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