Coverage for custom_components/supernotify/transport.py: 32%

1import logging 

2import time 

3from abc import abstractmethod 

4from traceback import format_exception 

5from typing import TYPE_CHECKING, Any 

6from urllib.parse import urlparse 

7 

8from homeassistant.components.notify.const import ATTR_TARGET 

9from homeassistant.const import ( 

10 ATTR_ENTITY_ID, 

11 ATTR_FRIENDLY_NAME, 

12 ATTR_NAME, 

13) 

14from homeassistant.exceptions import IntegrationError 

15from homeassistant.helpers.typing import ConfigType 

16from homeassistant.util import dt as dt_util 

17 

18from . import ( 

19 ATTR_ENABLED, 

20 CONF_DELIVERY_DEFAULTS, 

21) 

22from .common import CallRecord 

23from .context import Context 

24from .hass_api import HomeAssistantAPI 

25from .model import DebugTrace, DeliveryConfig, SuppressionReason, Target, TargetRequired, TransportConfig, TransportFeature 

26 

27if TYPE_CHECKING: 

28 import datetime as dt 

29 

30 from .delivery import Delivery, DeliveryRegistry 

31 from .people import PeopleRegistry 

32 

33_LOGGER = logging.getLogger(__name__) 

34 

35 

36class Transport: 

37 """Base class for delivery transports. 

38 

39 Sub classes integrste with Home Assistant notification services 

40 or alternative notification mechanisms. 

41 """ 

42 

43 name: str 

44 

45 @abstractmethod 

46 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None: 

47 self.hass_api: HomeAssistantAPI = context.hass_api 

48 self.people_registry: PeopleRegistry = context.people_registry 

49 self.delivery_registry: DeliveryRegistry = context.delivery_registry 

50 self.context: Context = context 

51 transport_config = transport_config or {} 

52 self.transport_config = TransportConfig(transport_config, class_config=self.default_config) 

53 

54 self.delivery_defaults: DeliveryConfig = self.transport_config.delivery_defaults 

55 self.config_enabled = self.transport_config.enabled 

56 self.enabled = self.config_enabled 

57 self.alias = self.transport_config.alias 

58 self.last_error_at: dt.datetime | None = None 

59 self.last_error_in: str | None = None 

60 self.last_error_message: str | None = None 

61 self.error_count: int = 0 

62 

63 async def initialize(self) -> None: 

64 """Async post-construction initialization""" 

65 if self.name is None: 

66 raise IntegrationError("Invalid nameless transport adaptor subclass") 

67 

68 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]: # noqa: ARG002 

69 return {} 

70 

71 @property 

72 def supported_features(self) -> TransportFeature: 

73 return TransportFeature.MESSAGE | TransportFeature.TITLE 

74 

75 @property 

76 def targets(self) -> Target: 

77 return self.delivery_defaults.target if self.delivery_defaults.target is not None else Target() 

78 

79 @property 

80 def default_config(self) -> TransportConfig: 

81 return TransportConfig() 

82 

83 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None: # noqa: ARG002 

84 return None 

85 

86 def validate_action(self, action: str | None) -> bool: 

87 """Override in subclass if transport has fixed action or doesn't require one""" 

88 return action == self.delivery_defaults.action 

89 

90 def attributes(self) -> dict[str, Any]: 

91 attrs: dict[str, Any] = { 

92 ATTR_NAME: self.name, 

93 ATTR_ENABLED: self.enabled, 

94 CONF_DELIVERY_DEFAULTS: self.delivery_defaults, 

95 } 

96 if self.alias: 

97 attrs[ATTR_FRIENDLY_NAME] = self.alias 

98 if self.last_error_at: 

99 attrs["last_error_at"] = self.last_error_at 

100 attrs["last_error_in"] = self.last_error_in 

101 attrs["last_error_message"] = self.last_error_message 

102 attrs["error_count"] = self.error_count 

103 attrs.update(self.extra_attributes()) 

104 return attrs 

105 

106 def extra_attributes(self) -> dict[str, Any]: 

107 return {} 

108 

109 @abstractmethod 

110 async def deliver(self, envelope: "Envelope", debug_trace: DebugTrace | None = None) -> bool: # type: ignore # noqa: F821 

111 """Delivery implementation 

112 

113 Args: 

114 ---- 

115 envelope (Envelope): envelope to be delivered 

116 debug_trace (DebugTrace): debug info collector 

117 

118 """ 

119 

120 def set_action_data(self, action_data: dict[str, Any], key: str, data: Any | None) -> Any: 

121 if data is not None: 

122 action_data[key] = data 

123 return action_data 

124 

125 async def call_action( 

126 self, 

127 envelope: "Envelope", # type: ignore # noqa: F821 

128 qualified_action: str | None = None, 

129 action_data: dict[str, Any] | None = None, 

130 target_data: dict[str, Any] | None = None, 

131 implied_target: bool = False, # True if the qualified action implies a target 

132 ) -> bool: 

133 action_data = action_data or {} 

134 start_time = time.time() 

135 domain = service = None 

136 delivery: Delivery = envelope.delivery 

137 try: 

138 qualified_action = qualified_action or delivery.action 

139 if not qualified_action: 

140 _LOGGER.debug( 

141 "SUPERNOTIFY skipping %s action call with no service, targets %s", 

142 envelope.delivery.name, 

143 action_data.get(ATTR_TARGET), 

144 ) 

145 envelope.skipped = 1 

146 envelope.skip_reason = SuppressionReason.NO_ACTION 

147 return False 

148 if ( 

149 delivery.target_required == TargetRequired.ALWAYS 

150 and not action_data.get(ATTR_TARGET) 

151 and not action_data.get(ATTR_ENTITY_ID) 

152 and not implied_target 

153 and not target_data 

154 ): 

155 _LOGGER.debug( 

156 "SUPERNOTIFY skipping %s action call for service %s, missing targets", 

157 envelope.delivery.name, 

158 qualified_action, 

159 ) 

160 envelope.skipped = 1 

161 envelope.skip_reason = SuppressionReason.NO_TARGET 

162 return False 

163 

164 domain, service = qualified_action.split(".", 1) 

165 start_time = time.time() 

166 if target_data: 

167 # home-assistant messes with the service_data passed by ref 

168 service_data_as_sent = dict(action_data) 

169 service_response = await self.hass_api.call_service( 

170 domain, service, service_data=action_data, target=target_data, debug=delivery.debug 

171 ) 

172 envelope.calls.append( 

173 CallRecord( 

174 time.time() - start_time, 

175 domain, 

176 service, 

177 debug=delivery.debug, 

178 action_data=service_data_as_sent, 

179 target_data=target_data, 

180 service_response=service_response, 

181 ) 

182 ) 

183 else: 

184 service_data_as_sent = dict(action_data) 

185 service_response = await self.hass_api.call_service( 

186 domain, service, service_data=action_data, debug=delivery.debug 

187 ) 

188 envelope.calls.append( 

189 CallRecord( 

190 time.time() - start_time, 

191 domain, 

192 service, 

193 debug=delivery.debug, 

194 action_data=service_data_as_sent, 

195 service_response=service_response, 

196 ) 

197 ) 

198 

199 envelope.delivered = 1 

200 return True 

201 except Exception as e: 

202 self.record_error(str(e), method="call_action") 

203 envelope.failed_calls.append( 

204 CallRecord(time.time() - start_time, domain, service, action_data, target_data, exception=str(e)) 

205 ) 

206 _LOGGER.exception("SUPERNOTIFY Failed to notify %s via %s, data=%s", self.name, qualified_action, action_data) 

207 envelope.error_count += 1 

208 envelope.delivery_error = format_exception(e) 

209 return False 

210 

211 def record_error(self, message: str, method: str) -> None: 

212 self.last_error_at = dt_util.utcnow() 

213 self.last_error_message = message 

214 self.last_error_in = method 

215 self.error_count += 1 

216 

217 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None: 

218 """Simplify text for delivery transports with speaking or plain text interfaces""" 

219 if not text: 

220 return None 

221 if strip_urls: 

222 words = text.split() 

223 text = " ".join(word for word in words if not urlparse(word).scheme) 

224 text = text.translate(str.maketrans("_", " ", "()£$<>")) 

225 _LOGGER.debug("SUPERNOTIFY Simplified text to: %s", text) 

226 return text