Coverage for tests/tests/test_parsers.py: 0%

1#!/usr/bin/env python3 

2"""RAMSES RF - Test the payload parsers.""" 

3 

4from pathlib import Path, PurePath 

5 

6import pytest 

7 

8from ramses_tx.const import Code 

9from ramses_tx.message import Message 

10from ramses_tx.packet import Packet 

11 

12from .helpers import TEST_DIR 

13 

14WORK_DIR = f"{TEST_DIR}/parsers" 

15 

16HAS_ARRAY = "has_array" 

17HAS_IDX = "has_idx" 

18HAS_PAYLOAD = "has_payload" 

19IS_FRAGMENT = "is_fragment" 

20META_KEYS = (HAS_ARRAY, HAS_IDX, HAS_PAYLOAD, IS_FRAGMENT) 

21 

22 

23def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: 

24 def id_fnc(param: Path) -> str: 

25 return PurePath(param).name 

26 

27 metafunc.parametrize("f_name", sorted(Path(WORK_DIR).glob("*.log")), ids=id_fnc) 

28 

29 

30def _proc_log_line(log_line: str) -> None: 

31 pkt_line, pkt_eval, *_ = list( 

32 map(str.strip, log_line.split("#", maxsplit=1) + [""]) 

33 ) 

34 

35 if not pkt_line: 

36 return 

37 

38 pkt = Packet.from_file(pkt_line[:26], pkt_line[27:]) 

39 msg = Message(pkt) 

40 

41 # assert bool(msg._is_fragment) == pkt._is_fragment 

42 # assert bool(msg._idx): dict == pkt._idx: Optional[bool | str] # not useful 

43 

44 if not pkt_eval: 

45 return 

46 try: 

47 pkt_dict = eval(pkt_eval) 

48 except SyntaxError: 

49 if "{" in pkt_eval: # if so, there is an issue with the log line 

50 raise # that should be addressed 

51 return 

52 

53 if isinstance(pkt_dict, list) or not any(k for k in pkt_dict if k in META_KEYS): 

54 assert msg.payload == pkt_dict, msg._pkt 

55 return 

56 

57 assert HAS_ARRAY not in pkt_dict or pkt._has_array == pkt_dict[HAS_ARRAY] 

58 assert HAS_IDX not in pkt_dict or pkt._idx == pkt_dict[HAS_IDX] 

59 assert HAS_PAYLOAD not in pkt_dict or pkt._has_payload == pkt_dict[HAS_PAYLOAD] 

60 assert IS_FRAGMENT not in pkt_dict or pkt._is_fragment == pkt_dict[IS_FRAGMENT] 

61 

62 

63def _proc_log_line_pair_4e15(log_line: str, prev_msg: Message) -> Message | None: 

64 pkt_line, *_ = list(map(str.strip, log_line.split("#", maxsplit=1) + [""])) 

65 

66 if not pkt_line: 

67 return None 

68 

69 pkt = Packet.from_file(pkt_line[:26], pkt_line[27:]) 

70 this_msg = Message(pkt) 

71 

72 if not prev_msg or prev_msg.code != Code._4E15: 

73 return this_msg 

74 

75 if this_msg.code != Code._3EF0: 

76 return None 

77 

78 assert prev_msg.payload["is_cooling"] == this_msg.payload["cool_active"] 

79 assert prev_msg.payload["is_heating"] == this_msg.payload["ch_active"] 

80 assert prev_msg.payload["is_dhw_ing"] == this_msg.payload["dhw_active"] 

81 

82 return this_msg 

83 

84 

85def test_parsers_from_log_files(f_name: Path) -> None: 

86 with open(f_name) as f: 

87 while line := (f.readline()): 

88 _proc_log_line(line) 

89 

90 

91def _test_parser_31da(f_name: Path) -> None: 

92 # assert _31DA_FAN_INFO[int(payload[36:38], 16) & 0x1F] in ( 

93 # speed_capabilities(payload[30:34])["speed_capabilities"] 

94 # ) or ( 

95 # int(payload[36:38], 16) & 0x1F in (1, 2, 3) and int(payload[30:34], 16) & 2**14 

96 # ) or ( 

97 # int(payload[36:38], 16) & 0x1F in (11, 12, 13) and int(payload[30:34], 16) & 2**14 and int(payload[30:34], 16) & 2**13 

98 # ) or ( 

99 # int(payload[36:38], 16) & 0x1F in (0x00, 0x18, 0x15) 

100 # ), {_31DA_FAN_INFO[int(payload[36:38], 16) & 0x1F]: speed_capabilities(payload[30:34])} 

101 

102 # assert payload[36:38] not in ("0B", "0C", "0D") or payload[42:46] == "0000", ( 

103 # payload[36:38], payload[42:46] 

104 # ) 

105 

106 pass 

107 

108 

109def _test_parser_pairs_31d9_31da(f_name: Path) -> None: 

110 pass 

111 

112 

113def _test_parser_pairs_4e15_3ef0(f_name: Path) -> None: 

114 if "4e15" in str(f_name): 

115 with open(f_name) as f: 

116 msg = None 

117 while this_line := (f.readline()): 

118 msg = _proc_log_line_pair_4e15(this_line, msg) 

119 

120 # elif "01ff" in str(f_name): 

121 # with open(f_name) as f: 

122 # msg = None 

123 # while this_line := (f.readline()): 

124 # msg = _proc_log_line_pair_01ff(this_line, msg)