Coverage for custom_components / supernotify / transport.py: 98%
126 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-11 22:18 +0000
1from __future__ import annotations
3import datetime as dt
4import logging
5import time
6import unicodedata
7from abc import abstractmethod
8from traceback import format_exception
9from typing import TYPE_CHECKING, Any
10from urllib.parse import urlparse
12from homeassistant.components.notify.const import ATTR_TARGET
13from homeassistant.const import (
14 ATTR_ENTITY_ID,
15 ATTR_FRIENDLY_NAME,
16 ATTR_NAME,
17)
18from homeassistant.exceptions import IntegrationError
19from homeassistant.util import dt as dt_util
21from custom_components.supernotify.model import (
22 DebugTrace,
23 Target,
24 TargetRequired,
25 TransportConfig,
26 TransportFeature,
27)
29from .common import CallRecord
30from .const import (
31 ATTR_ENABLED,
32 CONF_DELIVERY_DEFAULTS,
33)
34from .model import DeliveryConfig, SuppressionReason
36if TYPE_CHECKING:
37 from homeassistant.helpers.typing import ConfigType
39 from .context import Context
40 from .delivery import Delivery, DeliveryRegistry
41 from .hass_api import HomeAssistantAPI
42 from .people import PeopleRegistry
44_LOGGER = logging.getLogger(__name__)
47class Transport:
48 """Base class for delivery transports.
50 Sub classes integrste with Home Assistant notification services
51 or alternative notification mechanisms.
52 """
54 name: str
56 @abstractmethod
57 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
58 self.hass_api: HomeAssistantAPI = context.hass_api
59 self.people_registry: PeopleRegistry = context.people_registry
60 self.delivery_registry: DeliveryRegistry = context.delivery_registry
61 self.context: Context = context
62 transport_config = transport_config or {}
63 self.transport_config = TransportConfig(transport_config, class_config=self.default_config)
65 self.delivery_defaults: DeliveryConfig = self.transport_config.delivery_defaults
66 self.config_enabled = self.transport_config.enabled
67 self.enabled = self.config_enabled
68 self.alias = self.transport_config.alias
69 self.last_error_at: dt.datetime | None = None
70 self.last_error_in: str | None = None
71 self.last_error_message: str | None = None
72 self.error_count: int = 0
74 async def initialize(self) -> None:
75 """Async post-construction initialization"""
76 if self.name is None:
77 raise IntegrationError("Invalid nameless transport adaptor subclass")
79 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]: # noqa: ARG002
80 return {}
82 @property
83 def supported_features(self) -> TransportFeature:
84 return TransportFeature.MESSAGE | TransportFeature.TITLE
86 @property
87 def targets(self) -> Target:
88 return self.delivery_defaults.target if self.delivery_defaults.target is not None else Target()
90 @property
91 def default_config(self) -> TransportConfig:
92 return TransportConfig()
94 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None: # noqa: ARG002
95 return None
97 def validate_action(self, action: str | None) -> bool:
98 """Override in subclass if transport has fixed action or doesn't require one"""
99 return action == self.delivery_defaults.action
101 def attributes(self) -> dict[str, Any]:
102 attrs: dict[str, Any] = {
103 ATTR_NAME: self.name,
104 ATTR_ENABLED: self.enabled,
105 CONF_DELIVERY_DEFAULTS: self.delivery_defaults,
106 }
107 if self.alias:
108 attrs[ATTR_FRIENDLY_NAME] = self.alias
109 if self.last_error_at:
110 attrs["last_error_at"] = self.last_error_at
111 attrs["last_error_in"] = self.last_error_in
112 attrs["last_error_message"] = self.last_error_message
113 attrs["error_count"] = self.error_count
114 attrs.update(self.extra_attributes())
115 return attrs
117 def extra_attributes(self) -> dict[str, Any]:
118 return {}
120 @abstractmethod
121 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # type: ignore # noqa: F821
122 """Delivery implementation
124 Args:
125 ----
126 envelope (Envelope): envelope to be delivered
127 debug_trace (DebugTrace): debug info collector
129 """
131 def set_action_data(self, action_data: dict[str, Any], key: str, data: Any | None) -> Any:
132 if data is not None:
133 action_data[key] = data
134 return action_data
136 async def call_action(
137 self,
138 envelope: Envelope, # type: ignore # noqa: F821
139 qualified_action: str | None = None,
140 action_data: dict[str, Any] | None = None,
141 target_data: dict[str, Any] | None = None,
142 implied_target: bool = False, # True if the qualified action implies a target
143 ) -> bool:
144 action_data = action_data or {}
145 start_time = time.time()
146 domain = service = None
147 delivery: Delivery = envelope.delivery
148 try:
149 qualified_action = qualified_action or delivery.action
150 if not qualified_action:
151 _LOGGER.debug(
152 "SUPERNOTIFY skipping %s action call with no service, targets %s",
153 envelope.delivery.name,
154 action_data.get(ATTR_TARGET),
155 )
156 envelope.skipped = 1
157 envelope.skip_reason = SuppressionReason.NO_ACTION
158 return False
159 if (
160 delivery.target_required == TargetRequired.ALWAYS
161 and not action_data.get(ATTR_TARGET)
162 and not action_data.get(ATTR_ENTITY_ID)
163 and not implied_target
164 and not target_data
165 ):
166 _LOGGER.debug(
167 "SUPERNOTIFY skipping %s action call for service %s, missing targets",
168 envelope.delivery.name,
169 qualified_action,
170 )
171 envelope.skipped = 1
172 envelope.skip_reason = SuppressionReason.NO_TARGET
173 return False
175 domain, service = qualified_action.split(".", 1)
176 start_time = time.time()
177 timestamp: dt.datetime | None = None
178 if target_data:
179 # home-assistant messes with the service_data passed by ref
180 service_data_as_sent = dict(action_data)
181 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
182 service_response = await self.hass_api.call_service(
183 domain, service, service_data=action_data, target=target_data, debug=delivery.debug
184 )
185 envelope.calls.append(
186 CallRecord(
187 timestamp,
188 time.time() - start_time,
189 domain,
190 service,
191 debug=delivery.debug,
192 action_data=service_data_as_sent,
193 target_data=target_data,
194 service_response=service_response,
195 )
196 )
197 else:
198 service_data_as_sent = dict(action_data)
199 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
200 service_response = await self.hass_api.call_service(
201 domain, service, service_data=action_data, debug=delivery.debug
202 )
203 envelope.calls.append(
204 CallRecord(
205 timestamp,
206 time.time() - start_time,
207 domain,
208 service,
209 debug=delivery.debug,
210 action_data=service_data_as_sent,
211 service_response=service_response,
212 )
213 )
215 envelope.delivered = 1
216 return True
217 except Exception as e:
218 self.record_error(str(e), method="call_action")
219 envelope.failed_calls.append(
220 CallRecord(
221 timestamp,
222 time.time() - start_time,
223 domain,
224 service,
225 action_data,
226 target_data,
227 exception=str(e),
228 )
229 )
230 _LOGGER.exception("SUPERNOTIFY Failed to notify %s via %s, data=%s", self.name, qualified_action, action_data)
231 envelope.error_count += 1
232 envelope.delivery_error = format_exception(e)
233 return False
235 def record_error(self, message: str, method: str) -> None:
236 self.last_error_at = dt_util.utcnow()
237 self.last_error_message = message
238 self.last_error_in = method
239 self.error_count += 1
241 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None:
242 """Simplify text for delivery transports with speaking or plain text interfaces"""
243 if not text:
244 return None
245 if strip_urls:
246 words = text.split()
247 text = " ".join(word for word in words if not urlparse(word).scheme)
248 text = text.translate(str.maketrans("_", " ", "()£$<>"))
249 text = "".join(c for c in text if unicodedata.category(c) not in ("So", "Sk", "Sm", "Mn"))
250 _LOGGER.debug("SUPERNOTIFY Simplified text to: %s", text)
251 return text