Coverage for src/ramses_tx/__init__.py: 84%

1#!/usr/bin/env python3 

2"""RAMSES RF - a RAMSES-II protocol decoder & analyser. 

3`ramses_tx` takes care of the RF protocol (lower) layer. 

4""" 

5 

6from __future__ import annotations 

7 

8import asyncio 

9from functools import partial 

10from typing import TYPE_CHECKING, Any 

11 

12from .address import ( 

13 ALL_DEV_ADDR, 

14 ALL_DEVICE_ID, 

15 NON_DEV_ADDR, 

16 NON_DEVICE_ID, 

17 Address, 

18 is_valid_dev_id, 

19) 

20from .command import CODE_API_MAP, Command 

21from .const import ( 

22 DEV_ROLE_MAP, 

23 DEV_TYPE_MAP, 

24 F9, 

25 FA, 

26 FC, 

27 FF, 

28 SZ_ACTIVE_HGI, 

29 SZ_DEVICE_ROLE, 

30 SZ_DOMAIN_ID, 

31 SZ_ZONE_CLASS, 

32 SZ_ZONE_IDX, 

33 SZ_ZONE_MASK, 

34 SZ_ZONE_TYPE, 

35 ZON_ROLE_MAP, 

36 DevRole, 

37 DevType, 

38 IndexT, 

39 Priority, 

40 VerbT, 

41 ZoneRole, 

42) 

43from .gateway import Engine 

44from .logger import set_pkt_logging 

45from .message import Message 

46from .packet import PKT_LOGGER, Packet 

47from .protocol import PortProtocol, ReadProtocol, protocol_factory 

48from .ramses import ( 

49 _2411_PARAMS_SCHEMA, 

50 CODES_BY_DEV_SLUG, 

51 CODES_SCHEMA, 

52 SZ_DATA_TYPE, 

53 SZ_DATA_UNIT, 

54 SZ_DESCRIPTION, 

55 SZ_MAX_VALUE, 

56 SZ_MIN_VALUE, 

57 SZ_PRECISION, 

58) 

59from .schemas import SZ_BOUND_TO, SZ_SERIAL_PORT, DeviceIdT, DeviceListT 

60from .transport import ( 

61 FileTransport, 

62 PortTransport, 

63 RamsesTransportT, 

64 is_hgi80, 

65 transport_factory, 

66) 

67from .typing import QosParams 

68from .version import VERSION 

69 

70from .const import ( # isort: skip 

71 I_, 

72 RP, 

73 RQ, 

74 W_, 

75 Code, 

76) 

77 

78 

79__all__ = [ 

80 "VERSION", 

81 "Engine", 

82 # 

83 "SZ_ACTIVE_HGI", 

84 "SZ_DEVICE_ROLE", 

85 "SZ_DOMAIN_ID", 

86 "SZ_SERIAL_PORT", 

87 "SZ_ZONE_CLASS", 

88 "SZ_ZONE_IDX", 

89 "SZ_ZONE_MASK", 

90 "SZ_ZONE_TYPE", 

91 "SZ_BOUND_TO", 

92 # Schema-related constants 

93 "SZ_DATA_UNIT", 

94 "SZ_DESCRIPTION", 

95 "SZ_DATA_TYPE", 

96 "SZ_MAX_VALUE", 

97 "SZ_MIN_VALUE", 

98 "SZ_PRECISION", 

99 "_2411_PARAMS_SCHEMA", 

100 # 

101 "ALL_DEV_ADDR", 

102 "ALL_DEVICE_ID", 

103 "NON_DEV_ADDR", 

104 "NON_DEVICE_ID", 

105 # 

106 "CODE_API_MAP", 

107 "CODES_BY_DEV_SLUG", # shouldn't export this 

108 "CODES_SCHEMA", 

109 "DEV_ROLE_MAP", 

110 "DEV_TYPE_MAP", 

111 "ZON_ROLE_MAP", 

112 # 

113 "I_", 

114 "RP", 

115 "RQ", 

116 "W_", 

117 "F9", 

118 "FA", 

119 "FC", 

120 "FF", 

121 # 

122 "DeviceIdT", 

123 "DeviceListT", 

124 "DevRole", 

125 "DevType", 

126 "IndexT", 

127 "VerbT", 

128 "ZoneRole", 

129 # 

130 "Address", 

131 "Code", 

132 "Command", 

133 "Message", 

134 "Packet", 

135 "Priority", 

136 "QosParams", 

137 # 

138 "PortProtocol", 

139 "ReadProtocol", 

140 "RamsesProtocolT", 

141 "extract_known_hgi_id", 

142 "protocol_factory", 

143 # 

144 "FileTransport", 

145 "PortTransport", 

146 "RamsesTransportT", 

147 "is_hgi80", 

148 "transport_factory", 

149 # 

150 "is_valid_dev_id", 

151 "set_pkt_logging_config", 

152] 

153 

154 

155if TYPE_CHECKING: 

156 from logging import Logger 

157 

158 

159async def set_pkt_logging_config(**config: Any) -> Logger: 

160 """ 

161 Set up ramses packet logging to a file or port. 

162 Must run async in executor to prevent HA blocking call opening packet log file (issue #200) 

163 

164 :param config: if file_name is included, opens packet_log file 

165 :return: a logging.Logger 

166 """ 

167 loop = asyncio.get_running_loop() 

168 await loop.run_in_executor(None, partial(set_pkt_logging, PKT_LOGGER, **config)) 

169 return PKT_LOGGER 

170 

171 

172def extract_known_hgi_id( 

173 include_list: DeviceListT, 

174 /, 

175 *, 

176 disable_warnings: bool = False, 

177 strict_checking: bool = False, 

178) -> DeviceIdT | None: 

179 return PortProtocol._extract_known_hgi_id( 

180 include_list, 

181 disable_warnings=disable_warnings, 

182 strict_checking=strict_checking, 

183 )