diff --git a/pyoverkiz/client.py b/pyoverkiz/client.py index b51f0ded..060e5116 100644 --- a/pyoverkiz/client.py +++ b/pyoverkiz/client.py @@ -39,6 +39,7 @@ from pyoverkiz.models import ( Action, Command, + Definition, Device, Event, Execution, @@ -178,13 +179,18 @@ class OverkizClient: setup: Setup | None devices: list[Device] gateways: list[Gateway] - event_listener_id: str | None session: ClientSession _ssl: ssl.SSLContext | bool = True _auth: AuthStrategy _action_queue: ActionQueue | None = None + _event_listener_id: str | None settings: OverkizClientSettings + @property + def event_listener_id(self) -> str | None: + """Return the current event listener ID (read-only).""" + return self._event_listener_id + def __init__( self, *, @@ -207,7 +213,7 @@ def __init__( self.setup: Setup | None = None self.devices: list[Device] = [] self.gateways: list[Gateway] = [] - self.event_listener_id: str | None = None + self._event_listener_id: str | None = None self.session = session or ClientSession(headers={"User-Agent": USER_AGENT}) self._ssl = verify_ssl @@ -407,19 +413,23 @@ async def get_execution_history(self) -> list[HistoryExecution]: return structure_response(response, list[HistoryExecution]) @retry_on_auth_error - async def get_device_definition(self, deviceurl: str) -> dict[str, Any] | None: + async def get_device_definition(self, device_url: str) -> Definition | None: """Retrieve a particular setup device definition.""" response: dict = await self._get( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}" + f"setup/devices/{urllib.parse.quote_plus(device_url)}" ) - return response.get("definition") + raw = response.get("definition") + if raw is None: + return None + + return structure_response(raw, Definition) @retry_on_auth_error - async def get_state(self, deviceurl: str) -> list[State]: + async def get_state(self, device_url: str) -> list[State]: """Retrieve states of requested device.""" response = await self._get( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/states" + f"setup/devices/{urllib.parse.quote_plus(device_url)}/states" ) return structure_response(response, list[State]) @@ -429,10 +439,10 @@ async def refresh_states(self) -> None: await self._post("setup/devices/states/refresh") @retry_on_auth_error - async def refresh_device_states(self, deviceurl: str) -> None: + async def refresh_device_states(self, device_url: str) -> None: """Ask the box to refresh all states of the given device for protocols supporting that operation.""" await self._post( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/states/refresh" + f"setup/devices/{urllib.parse.quote_plus(device_url)}/states/refresh" ) @retry_on_concurrent_requests @@ -448,7 +458,7 @@ async def register_event_listener(self) -> str: """ response = await self._post("events/register") listener_id = cast(str, response.get("id")) - self.event_listener_id = listener_id + self._event_listener_id = listener_id return listener_id @@ -471,8 +481,8 @@ async def unregister_event_listener(self) -> None: API response status is always 200, even on unknown listener ids. """ - await self._post(f"events/{self.event_listener_id}/unregister") - self.event_listener_id = None + await self._post(f"events/{self._event_listener_id}/unregister") + self._event_listener_id = None @retry_on_auth_error async def get_current_execution(self, exec_id: str) -> Execution | None: @@ -780,38 +790,40 @@ async def get_devices_not_up_to_date(self) -> list[Device]: return structure_response(response, list[Device]) @retry_on_auth_error - async def get_device_firmware_status(self, deviceurl: str) -> FirmwareStatus | None: + async def get_device_firmware_status( + self, device_url: str + ) -> FirmwareStatus | None: """Check if a device's firmware is up to date. Returns None if the device does not support firmware status checks. """ try: response = await self._get( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/firmwareUpToDate" + f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpToDate" ) except UnsupportedOperationError: return None return structure_response(response, FirmwareStatus) @retry_on_auth_error - async def get_device_firmware_update_capability(self, deviceurl: str) -> bool: + async def get_device_firmware_update_capability(self, device_url: str) -> bool: """Check if a device supports firmware updates. Returns False if the device does not support this query. """ try: response = await self._get( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/firmwareUpdateCapability" + f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpdateCapability" ) except UnsupportedOperationError: return False return cast(bool, response["supportsFirmwareUpdate"]) @retry_on_auth_error - async def update_device_firmware(self, deviceurl: str) -> None: + async def update_device_firmware(self, device_url: str) -> None: """Update a device's firmware to the next available version.""" await self._put( - f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/updateFirmware" + f"setup/devices/{urllib.parse.quote_plus(device_url)}/updateFirmware" ) @retry_on_auth_error diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index 88a22a2e..f4d74bef 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -3,6 +3,7 @@ from __future__ import annotations from importlib.metadata import version +from types import MappingProxyType from pyoverkiz.enums import Server from pyoverkiz.enums.server import APIType @@ -45,117 +46,119 @@ Server.SOMFY_AMERICA, ] -SUPPORTED_SERVERS: dict[str, ServerConfig] = { - Server.ATLANTIC_COZYTOUCH: ServerConfig( - server=Server.ATLANTIC_COZYTOUCH, - name="Atlantic Cozytouch", - endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Atlantic", - api_type=APIType.CLOUD, - ), - Server.BRANDT: ServerConfig( - server=Server.BRANDT, - name="Brandt Smart Control", - endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Brandt", - api_type=APIType.CLOUD, - ), - Server.FLEXOM: ServerConfig( - server=Server.FLEXOM, - name="Flexom", - endpoint="https://ha108-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Bouygues", - api_type=APIType.CLOUD, - ), - Server.HEXAOM_HEXACONNECT: ServerConfig( - server=Server.HEXAOM_HEXACONNECT, - name="Hexaom HexaConnect", - endpoint="https://ha5-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hexaom", - api_type=APIType.CLOUD, - ), - Server.HI_KUMO_ASIA: ServerConfig( - server=Server.HI_KUMO_ASIA, - name="Hitachi Hi Kumo (Asia)", - endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - api_type=APIType.CLOUD, - ), - Server.HI_KUMO_EUROPE: ServerConfig( - server=Server.HI_KUMO_EUROPE, - name="Hitachi Hi Kumo (Europe)", - endpoint="https://ha117-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - api_type=APIType.CLOUD, - ), - Server.HI_KUMO_OCEANIA: ServerConfig( - server=Server.HI_KUMO_OCEANIA, - name="Hitachi Hi Kumo (Oceania)", - endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - api_type=APIType.CLOUD, - ), - Server.NEXITY: ServerConfig( - server=Server.NEXITY, - name="Nexity Eugénie", - endpoint="https://ha106-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Nexity", - api_type=APIType.CLOUD, - ), - Server.REXEL: ServerConfig( - server=Server.REXEL, - name="Rexel Energeasy Connect", - endpoint=REXEL_BACKEND_API, - manufacturer="Rexel", - api_type=APIType.CLOUD, - ), - Server.SAUTER_COZYTOUCH: ServerConfig( # duplicate of Atlantic Cozytouch - server=Server.SAUTER_COZYTOUCH, - name="Sauter Cozytouch", - endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Sauter", - api_type=APIType.CLOUD, - ), - Server.SIMU_LIVEIN2: ServerConfig( # alias of https://tahomalink.com - server=Server.SIMU_LIVEIN2, - name="SIMU (LiveIn2)", - endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - api_type=APIType.CLOUD, - ), - Server.SOMFY_EUROPE: ServerConfig( # alias of https://tahomalink.com - server=Server.SOMFY_EUROPE, - name="Somfy (Europe)", - endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - api_type=APIType.CLOUD, - ), - Server.SOMFY_AMERICA: ServerConfig( - server=Server.SOMFY_AMERICA, - name="Somfy (North America)", - endpoint="https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - api_type=APIType.CLOUD, - ), - Server.SOMFY_OCEANIA: ServerConfig( - server=Server.SOMFY_OCEANIA, - name="Somfy (Oceania)", - endpoint="https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - api_type=APIType.CLOUD, - ), - Server.THERMOR_COZYTOUCH: ServerConfig( # duplicate of Atlantic Cozytouch - server=Server.THERMOR_COZYTOUCH, - name="Thermor Cozytouch", - endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Thermor", - api_type=APIType.CLOUD, - ), - Server.UBIWIZZ: ServerConfig( - server=Server.UBIWIZZ, - name="Ubiwizz", - endpoint="https://ha129-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Decelect", - api_type=APIType.CLOUD, - ), -} +SUPPORTED_SERVERS: MappingProxyType[str, ServerConfig] = MappingProxyType( + { + Server.ATLANTIC_COZYTOUCH: ServerConfig( + server=Server.ATLANTIC_COZYTOUCH, + name="Atlantic Cozytouch", + endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Atlantic", + api_type=APIType.CLOUD, + ), + Server.BRANDT: ServerConfig( + server=Server.BRANDT, + name="Brandt Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Brandt", + api_type=APIType.CLOUD, + ), + Server.FLEXOM: ServerConfig( + server=Server.FLEXOM, + name="Flexom", + endpoint="https://ha108-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Bouygues", + api_type=APIType.CLOUD, + ), + Server.HEXAOM_HEXACONNECT: ServerConfig( + server=Server.HEXAOM_HEXACONNECT, + name="Hexaom HexaConnect", + endpoint="https://ha5-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hexaom", + api_type=APIType.CLOUD, + ), + Server.HI_KUMO_ASIA: ServerConfig( + server=Server.HI_KUMO_ASIA, + name="Hitachi Hi Kumo (Asia)", + endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + api_type=APIType.CLOUD, + ), + Server.HI_KUMO_EUROPE: ServerConfig( + server=Server.HI_KUMO_EUROPE, + name="Hitachi Hi Kumo (Europe)", + endpoint="https://ha117-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + api_type=APIType.CLOUD, + ), + Server.HI_KUMO_OCEANIA: ServerConfig( + server=Server.HI_KUMO_OCEANIA, + name="Hitachi Hi Kumo (Oceania)", + endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + api_type=APIType.CLOUD, + ), + Server.NEXITY: ServerConfig( + server=Server.NEXITY, + name="Nexity Eugénie", + endpoint="https://ha106-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Nexity", + api_type=APIType.CLOUD, + ), + Server.REXEL: ServerConfig( + server=Server.REXEL, + name="Rexel Energeasy Connect", + endpoint=REXEL_BACKEND_API, + manufacturer="Rexel", + api_type=APIType.CLOUD, + ), + Server.SAUTER_COZYTOUCH: ServerConfig( # duplicate of Atlantic Cozytouch + server=Server.SAUTER_COZYTOUCH, + name="Sauter Cozytouch", + endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Sauter", + api_type=APIType.CLOUD, + ), + Server.SIMU_LIVEIN2: ServerConfig( # alias of https://tahomalink.com + server=Server.SIMU_LIVEIN2, + name="SIMU (LiveIn2)", + endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + api_type=APIType.CLOUD, + ), + Server.SOMFY_EUROPE: ServerConfig( # alias of https://tahomalink.com + server=Server.SOMFY_EUROPE, + name="Somfy (Europe)", + endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + api_type=APIType.CLOUD, + ), + Server.SOMFY_AMERICA: ServerConfig( + server=Server.SOMFY_AMERICA, + name="Somfy (North America)", + endpoint="https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + api_type=APIType.CLOUD, + ), + Server.SOMFY_OCEANIA: ServerConfig( + server=Server.SOMFY_OCEANIA, + name="Somfy (Oceania)", + endpoint="https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + api_type=APIType.CLOUD, + ), + Server.THERMOR_COZYTOUCH: ServerConfig( # duplicate of Atlantic Cozytouch + server=Server.THERMOR_COZYTOUCH, + name="Thermor Cozytouch", + endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Thermor", + api_type=APIType.CLOUD, + ), + Server.UBIWIZZ: ServerConfig( + server=Server.UBIWIZZ, + name="Ubiwizz", + endpoint="https://ha129-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Decelect", + api_type=APIType.CLOUD, + ), + } +) diff --git a/pyoverkiz/models.py b/pyoverkiz/models.py index 4896fff7..b01d9ba1 100644 --- a/pyoverkiz/models.py +++ b/pyoverkiz/models.py @@ -27,6 +27,7 @@ from pyoverkiz.enums.command import OverkizCommand from pyoverkiz.enums.protocol import Protocol from pyoverkiz.enums.server import APIType, Server +from pyoverkiz.exceptions import OverkizError from pyoverkiz.obfuscate import obfuscate_email, obfuscate_id, obfuscate_string from pyoverkiz.types import DATA_TYPE_TO_PYTHON, CommandParameterValue, StateType @@ -363,7 +364,7 @@ def from_device_url(cls, device_url: str) -> DeviceIdentifier: """Parse a device URL into its structured identifier components.""" match = DEVICE_URL_RE.fullmatch(device_url) if not match: - raise ValueError(f"Invalid device URL: {device_url}") + raise OverkizError(f"Invalid device URL: {device_url}") subsystem_id = ( int(match.group("subsystemId")) if match.group("subsystemId") else None @@ -414,7 +415,9 @@ def __attrs_post_init__(self) -> None: self.widget = UIWidget(self.definition.widget_name) if self.ui_class is None or self.widget is None: - raise ValueError(f"Device {self.device_url} is missing ui_class or widget") + raise OverkizError( + f"Device {self.device_url} is missing ui_class or widget" + ) def supports_command(self, command: str | OverkizCommand) -> bool: """Check if device supports a command.""" @@ -573,7 +576,7 @@ class Execution: id: str description: str owner: str = field(repr=obfuscate_email) - state: str + state: ExecutionState action_group: ActionGroup | None = None start_time: int | None = None execution_type: ExecutionType | None = None @@ -722,17 +725,17 @@ class Location: postal_code: str | None = field(repr=obfuscate_string, default=None) address_line1: str | None = field(repr=obfuscate_string, default=None) address_line2: str | None = field(repr=obfuscate_string, default=None) - timezone: str = "" + timezone: str | None = None longitude: str | None = field(repr=obfuscate_string, default=None) latitude: str | None = field(repr=obfuscate_string, default=None) - twilight_mode: int = 0 - twilight_angle: str = "" + twilight_mode: int | None = None + twilight_angle: str | None = None twilight_city: str | None = None - summer_solstice_dusk_minutes: str = "" - winter_solstice_dusk_minutes: str = "" + summer_solstice_dusk_minutes: str | None = None + winter_solstice_dusk_minutes: str | None = None twilight_offset_enabled: bool = False - dawn_offset: int = 0 - dusk_offset: int = 0 + dawn_offset: int | None = None + dusk_offset: int | None = None country_code: str | None = field(repr=obfuscate_string, default=None) tariff_settings: dict[str, Any] | None = None diff --git a/pyoverkiz/obfuscate.py b/pyoverkiz/obfuscate.py index a2fb6967..1e320d02 100644 --- a/pyoverkiz/obfuscate.py +++ b/pyoverkiz/obfuscate.py @@ -22,52 +22,61 @@ def obfuscate_string(value: str) -> str: return re.sub(r"[a-zA-Z0-9_.-]*", "*", str(value)) +def _obfuscate_value(key: str, value: Any, mask_next_value: bool) -> tuple[Any, bool]: + """Return (obfuscated_value, mask_next_value) for a single key/value pair.""" + result = value + + if key in {"gatewayId", "id", "deviceURL"}: + result = obfuscate_id(value) + elif key in { + "label", + "city", + "country", + "postalCode", + "addressLine1", + "addressLine2", + "longitude", + "latitude", + }: + result = obfuscate_string(value) + elif mask_next_value and key == "value": + result = obfuscate_string(value) + return result, False + + if result in ( + "core:NameState", + "homekit:SetupCode", + "homekit:SetupPayload", + "core:SSIDState", + "core:NetworkMacState", + ): + mask_next_value = True + + if isinstance(result, dict): + result = obfuscate_sensitive_data(result) + elif isinstance(result, list): + result = [ + obfuscate_sensitive_data(item) if isinstance(item, dict) else item + for item in result + ] + + return result, mask_next_value + + def obfuscate_sensitive_data( data: dict[str, Any] | list[dict[str, Any]], ) -> dict[str, Any] | list[dict[str, Any]]: - """Mask Overkiz JSON data to remove sensitive data.""" + """Return a copy of Overkiz JSON data with sensitive values masked.""" if isinstance(data, list): return cast( list[dict[str, Any]], [obfuscate_sensitive_data(item) for item in data] ) + result: dict[str, Any] = {} mask_next_value = False for key, value in data.items(): - if key in {"gatewayId", "id", "deviceURL"}: - data[key] = obfuscate_id(value) - - if key in { - "label", - "city", - "country", - "postalCode", - "addressLine1", - "addressLine2", - "longitude", - "latitude", - }: - data[key] = obfuscate_string(value) - - if value in ( - "core:NameState", - "homekit:SetupCode", - "homekit:SetupPayload", - "core:SSIDState", - "core:NetworkMacState", - ): - mask_next_value = True - - if mask_next_value and key == "value": - data[key] = obfuscate_string(value) - mask_next_value = False - - # Mask homekit:SetupCode and homekit:SetupPayload - if isinstance(value, dict): - obfuscate_sensitive_data(value) - elif isinstance(value, list): - for val in value: - if isinstance(val, dict): - obfuscate_sensitive_data(val) - - return data + obfuscated, mask_next_value = _obfuscate_value(key, value, mask_next_value) + result[key] = obfuscated + + return result diff --git a/pyoverkiz/utils.py b/pyoverkiz/utils.py index 2427f6aa..e538c09e 100644 --- a/pyoverkiz/utils.py +++ b/pyoverkiz/utils.py @@ -12,16 +12,17 @@ def create_local_server_config( *, host: str, + server: Server | str = Server.SOMFY_DEVELOPER_MODE, name: str = "Somfy Developer Mode", manufacturer: str = "Somfy", configuration_url: str | None = None, ) -> ServerConfig: - """Generate server configuration for a local API (Somfy Developer mode).""" + """Generate server configuration for a local API.""" return create_server_config( name=name, endpoint=f"https://{host}{LOCAL_API_PATH}", manufacturer=manufacturer, - server=Server.SOMFY_DEVELOPER_MODE, + server=server, configuration_url=configuration_url, api_type=APIType.LOCAL, ) diff --git a/tests/test_client.py b/tests/test_client.py index 89468747..bb3b6c63 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -129,7 +129,7 @@ async def test_backoff_refresh_listener_on_listener_error( self, client: OverkizClient ) -> None: """Ensure listener backoff retries and triggers `register_event_listener()`.""" - client.event_listener_id = "listener-1" + client._event_listener_id = "listener-1" client.register_event_listener = AsyncMock(return_value="listener-2") with ( diff --git a/utils/generate_enums.py b/utils/generate_enums.py index 8c38aca1..d32f6425 100644 --- a/utils/generate_enums.py +++ b/utils/generate_enums.py @@ -326,7 +326,7 @@ def clean_description(desc: str) -> str: "class UIProfile(UnknownEnumMixin, StrEnum):", ' """', " UI Profiles define device capabilities through commands and states.", - " ", + "", " Each profile describes what a device can do (commands) and what information", " it provides (states). Form factor indicates if the profile is tied to a", " specific physical device type.",