diff --git a/README.md b/README.md index 6e917e42..43f34940 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ Asynchronous Python client for communicating with a OJ Microline Thermostat. ## About -A Python package to control OJ Microline thermostats. It currently supports the WD5 series (OWD5, MWD5) and WG4 series (UWG4, AWG4). +A Python package to control OJ Microline thermostats. It currently supports the WD5 series (OWD5, MWD5), WG4 series (UWG4, AWG4), and WG5 series. ## Installation @@ -63,24 +63,34 @@ These fields are only available on WD5-series thermostats; for others, they may | :------- | :--- | :---------- | | `thermostat_id` | int | The unique identifier for this thermostat. | | `adaptive_mode` | boolean | If on then then the thermostat automatically changes heating start times to ensure that the required temperature has been reached at the beginning of any specific event. | -| `open_window_detection` | boolean | If on then the thermostat shuts off the heating for 30 minutes if an open window is detected. | +| `open_window_detection` | boolean | If on then the thermostat shuts off the heating for 30 minutes if an open window is detected. Also available on WG5-series thermostats. | | `daylight_saving_active` | boolean | If on, the "Daylight Saving Time" function of the thermostat will automatically adjust the clock to the daylight saving time for the "Region" chosen. | -| `sensor_mode` | integer | The currently set sensor mode of the thermostat, see below. | -| `temperature_floor` | integer | The temperature measured by the floor sensor. | -| `temperature_room` | integer | The temperature measured by the room sensor. | +| `sensor_mode` | integer | The currently set sensor mode of the thermostat, see below. Also available on WG5-series thermostats. | +| `temperature_floor` | integer | The temperature measured by the floor sensor. Also available on WG5-series thermostats. | +| `temperature_room` | integer | The temperature measured by the room sensor. Also available on WG5-series thermostats. | | `boost_temperature` | integer | If the regulation mode is set to boost mode, the thermostat will target this temperature. | | `boost_end_time` | datetime | If the regulation mode is set to boost mode, it will end at this time. | -| `frost_protection_temperature` | integer | If the regulation mode is set to frost protection mode, the thermostat will target this temperature. | -| `schedule` | Schedule | The schedule the thermostat currently uses. (This *could* be supported by WG4-series thermostats, it simply isn't implemented.) | +| `frost_protection_temperature` | integer | If the regulation mode is set to frost protection mode, the thermostat will target this temperature. Also available on WG5-series thermostats. | +| `schedule` | Schedule/dict | The schedule the thermostat currently uses. On WD5 this is a `Schedule` object; on WG5 it is a raw dict from the API. | | `energy` | list | The energy usage in kWh for the current day and the six previous days. (Note that the integrated tariff calculation needs to be disabled.) | -These fields are only available on WG4-series thermostats; for others, they may be `None`: +These fields are available on WG4 and WG5-series thermostats; for others, they may be `None`: | Variable | Type | Description | | :------- | :--- | :---------- | | `temperature` | integer | The current temperature; the thermostat uses the room sensor or floor sensor based on its configuration. Avoid using this directly; instead, call the `get_current_temperature()` method which also works for WD5-series thermostats. | | `set_point_temperature` | integer | The temperature the thermostat is targeting. Avoid using this directly; instead, call the `get_target_temperature()` method which also works for WD5-series thermostats. | +These fields are only available on WG5-series thermostats; for others, they may be `None`: + +| Variable | Type | Description | +| :------- | :--- | :---------- | +| `building_id` | string | The building UUID this thermostat belongs to. | +| `zone_uuid` | string | The zone UUID this thermostat belongs to (WG5 uses UUIDs rather than integer zone IDs). | +| `is_in_standby` | boolean | Whether the thermostat is in standby (frost protection) mode. | +| `schedule_id` | string | The UUID of the schedule assigned to this thermostat. | +| `schedule_name` | string | The name of the schedule assigned to this thermostat. | + #### Regulation modes | Integer | Constant | Description | @@ -115,6 +125,8 @@ Keep in mind that certain thermostats only support a subset of these modes; be s ## Usage +### WD5 Example + ```python import asyncio from time import sleep @@ -173,6 +185,44 @@ async def main(): await client.set_regulation_mode(resource, REGULATION_SCHEDULE) +if __name__ == "__main__": + asyncio.run(main()) +``` + +### WG5 Example + +```python +import asyncio + +from ojmicroline_thermostat import OJMicroline, Thermostat +from ojmicroline_thermostat.wg5 import WG5API +from ojmicroline_thermostat.const import ( + REGULATION_MANUAL, + REGULATION_SCHEDULE, +) + + +async def main(): + """Show example on using the OJMicroline client with a WG5 thermostat.""" + async with OJMicroline( + api=WG5API( + username="", + password="", + ), + ) as client: + thermostats: list[Thermostat] = await client.get_thermostats() + + for resource in thermostats: + print(f"{resource.name}: {resource.get_current_temperature() / 100}°C") + print(f" Target: {resource.get_target_temperature() / 100}°C") + + # Set to manual mode at 25°C + await client.set_regulation_mode(resource, REGULATION_MANUAL, 2500) + + # Set back to schedule + await client.set_regulation_mode(resource, REGULATION_SCHEDULE) + + if __name__ == "__main__": asyncio.run(main()) ``` diff --git a/ojmicroline_thermostat/__init__.py b/ojmicroline_thermostat/__init__.py index b0d23cb1..63145921 100644 --- a/ojmicroline_thermostat/__init__.py +++ b/ojmicroline_thermostat/__init__.py @@ -11,10 +11,12 @@ from .ojmicroline import OJMicroline from .wd5 import WD5API from .wg4 import WG4API +from .wg5 import WG5API __all__ = [ "WD5API", "WG4API", + "WG5API", "OJMicroline", "OJMicrolineAuthError", "OJMicrolineConnectionError", diff --git a/ojmicroline_thermostat/const.py b/ojmicroline_thermostat/const.py index 865c4016..d11ed046 100644 --- a/ojmicroline_thermostat/const.py +++ b/ojmicroline_thermostat/const.py @@ -15,3 +15,12 @@ WD5_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S" WG4_DATETIME_FORMAT = "%d/%m/%Y %H:%M:%S %z" COMFORT_DURATION = 240 + +WG5_SENSOR_MAP: dict[str, int] = { + "Floor": SENSOR_FLOOR, + "Room": SENSOR_ROOM, + "RoomFloor": SENSOR_ROOM_FLOOR, +} + +WG5_MODE_SCHEDULE = 1 +WG5_MODE_HOLD = 2 diff --git a/ojmicroline_thermostat/models/thermostat.py b/ojmicroline_thermostat/models/thermostat.py index 608e5437..3bcf0b6b 100644 --- a/ojmicroline_thermostat/models/thermostat.py +++ b/ojmicroline_thermostat/models/thermostat.py @@ -5,7 +5,7 @@ from __future__ import annotations from dataclasses import dataclass -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from math import ceil from time import gmtime, strftime from typing import Any @@ -22,6 +22,7 @@ SENSOR_ROOM, SENSOR_ROOM_FLOOR, WG4_DATETIME_FORMAT, + WG5_SENSOR_MAP, ) from .schedule import Schedule @@ -66,10 +67,17 @@ class Thermostat: boost_temperature: int | None = None energy: list[float] | None = None - # WG4-only fields: + # WG4/WG5 fields: temperature: int | None = None set_point_temperature: int | None = None + # WG5-only fields: + building_id: str | None = None + zone_uuid: str | None = None + is_in_standby: bool | None = None + schedule_id: str | None = None + schedule_name: str | None = None + @classmethod def from_wd5_json(cls, data: dict[str, Any]) -> Thermostat: """Return a new Thermostat instance based on JSON from the WD5-series API. @@ -178,6 +186,79 @@ def from_wg4_json(cls, data: dict[str, Any]) -> Thermostat: vacation_temperature=data["VacationTemperature"], ) + @classmethod + def from_wg5_json( + cls, + data: dict[str, Any], + building_id: str, + zone_name: str, + *, + away_active: bool = False, + ) -> Thermostat: + """Return a new Thermostat instance based on JSON from the WG5-series API. + + Args: + ---- + data: The thermostat control JSON data from the API. + building_id: The building UUID this thermostat belongs to. + zone_name: The zone name this thermostat belongs to. + away_active: Whether away mode is active for the building. + + Returns: + ------- + A Thermostat Object. + + """ + mode = data["mode"] + setpoint_centideg = round(data["setpoint"] * 100) + + if away_active or mode.get("isAwayActive"): + regulation_mode = REGULATION_VACATION + elif mode["isInStandby"]: + regulation_mode = REGULATION_FROST_PROTECTION + elif mode.get("fallbackMode") == "Auto": + regulation_mode = REGULATION_SCHEDULE + else: + regulation_mode = REGULATION_MANUAL + + return cls( + model="WG5", + serial_number=data["id"], + software_version="", + zone_name=zone_name, + zone_id=0, + zone_uuid=data["zoneId"], + building_id=building_id, + name=data["name"], + online=data["isOnline"], + heating=data["isHeatRelayActive"], + regulation_mode=regulation_mode, + supported_regulation_modes=[ + REGULATION_SCHEDULE, + REGULATION_MANUAL, + REGULATION_COMFORT, + REGULATION_VACATION, + REGULATION_FROST_PROTECTION, + ], + min_temperature=round(data["minimumPossibleSetPoint"] * 100), + max_temperature=round(data["maximumPossibleSetPoint"] * 100), + temperature=round(data["currentTemperature"] * 100), + set_point_temperature=setpoint_centideg, + manual_temperature=setpoint_centideg, + comfort_temperature=setpoint_centideg, + comfort_end_time=datetime.min.replace(tzinfo=UTC), + last_primary_mode_is_auto=mode.get("fallbackMode") == "Auto", + frost_protection_temperature=round( + data["frostProtectionTemperature"] * 100 + ), + sensor_mode=WG5_SENSOR_MAP.get(data.get("sensorApplication", "")), + open_window_detection=data.get("isOpenWindowDetected", False), + vacation_mode=away_active or mode.get("isAwayActive", False), + is_in_standby=mode["isInStandby"], + schedule_id=data.get("scheduleData", {}).get("scheduleId"), + schedule_name=data.get("scheduleData", {}).get("scheduleName"), + ) + def get_target_temperature(self) -> int: """Return the target temperature for the thermostat. diff --git a/ojmicroline_thermostat/ojmicroline.py b/ojmicroline_thermostat/ojmicroline.py index 39513297..a2058a25 100644 --- a/ojmicroline_thermostat/ojmicroline.py +++ b/ojmicroline_thermostat/ojmicroline.py @@ -5,6 +5,7 @@ import json import socket +from collections.abc import Awaitable, Callable from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any, Protocol, Self @@ -24,6 +25,8 @@ if TYPE_CHECKING: from .models import Thermostat +RequestFunc = Callable[..., Awaitable[Any]] + class OJMicrolineAPI(Protocol): """Implements support for a specific OJ Microline API. @@ -37,52 +40,84 @@ class OJMicrolineAPI(Protocol): host: str """The host to use for API requests, without the protocol prefix.""" - login_path: str - """HTTP path used to log in and fetch a session id.""" - - def login_body(self) -> dict[str, Any]: - """Compute HTTP body parameters used when posting to the login path.""" + request: RequestFunc + """Callable for making HTTP requests, set by OJMicroline.""" - get_thermostats_path: str - """HTTP path used to fetch a list of thermostats.""" + async def login(self) -> None: + """Perform authentication against the API.""" - def get_thermostats_params(self) -> dict[str, Any]: - """Compute additional query string params for fetching thermostats.""" + async def get_thermostats(self) -> list[Thermostat]: + """Fetch all thermostats.""" - def parse_thermostats_response(self, data: Any) -> list[Thermostat]: - """Parse an HTTP response containing thermostat data. + async def get_energy_usage(self, resource: Thermostat) -> list[float]: + """Fetch energy usage for a thermostat. Args: ---- - data: The JSON data contained in the response. + resource: The Thermostat model. """ - get_energy_usage_path: str - """HTTP path used to fetch energy usage data.""" - - def parse_energy_usage_response(self, data: Any) -> list[float]: - """Parse an HTTP response containing energy usage data. + async def set_regulation_mode( + self, + resource: Thermostat, + regulation_mode: int, + temperature: int | None, + duration: int, + ) -> bool: + """Set the regulation mode on a thermostat. Args: ---- - data: The JSON data contained in the response. - - Returns: A list with the energy usage values. + resource: The Thermostat model. + regulation_mode: The mode to set the thermostat to. + temperature: The temperature to set or None. + duration: The duration in minutes (comfort mode only). """ + +class SessionOJMicrolineAPI: + """Base class for session-based OJ Microline APIs (WD5, WG4). + + Provides login/get_thermostats/get_energy_usage/set_regulation_mode + implementations that use a session ID obtained via a login endpoint. + Subclasses provide the low-level details (paths, bodies, parsers). + """ + + host: str + request: RequestFunc + + # Subclass attributes: + login_path: str + get_thermostats_path: str + get_energy_usage_path: str update_regulation_mode_path: str - """HTTP path used to update a thermostat's mode""" - def update_regulation_mode_params(self, thermostat: Thermostat) -> dict[str, Any]: - """Compute additional query string params for posting to the update path. + # Session state: + _session_id: str | None = None + _session_calls_left: int = 0 + _session_calls: int = 300 - Args: - ---- - thermostat: The Thermostat model. + def login_body(self) -> dict[str, Any]: + """Compute HTTP body parameters used when posting to the login path.""" + raise NotImplementedError - """ + def get_thermostats_params(self) -> dict[str, Any]: + """Compute additional query string params for fetching thermostats.""" + raise NotImplementedError + + def parse_thermostats_response(self, data: Any) -> list[Thermostat]: + """Parse an HTTP response containing thermostat data.""" + raise NotImplementedError + + def parse_energy_usage_response(self, data: Any) -> list[float]: + """Parse an HTTP response containing energy usage data.""" + raise NotImplementedError + + def update_regulation_mode_params(self, thermostat: Thermostat) -> dict[str, Any]: + """Compute additional query string params for posting to the update path.""" + raise NotImplementedError def update_regulation_mode_body( self, @@ -91,30 +126,111 @@ def update_regulation_mode_body( temperature: int | None, duration: int, ) -> dict[str, Any]: - """Compute HTTP body parameters used when posting to the update path. + """Compute HTTP body parameters used when posting to the update path.""" + raise NotImplementedError - Args: - ---- - thermostat: The Thermostat model. - regulation_mode: The mode to set the thermostat to. - temperature: The temperature to set or None. - duration: The duration in minutes to set the temperature - for (comfort mode only). + def parse_update_regulation_mode_response(self, data: Any) -> bool: + """Parse the HTTP response received after updating the regulation mode.""" + raise NotImplementedError - Returns: A dict with values to be used in an HTTP POST body. + async def login(self) -> None: + """Get a valid session to do requests with the OJ Microline API. + + Raises + ------ + OJMicrolineAuthError: An error occurred while authenticating. """ + self._session_calls_left -= 1 + if self._session_calls_left < 0 or self._session_id is None: + data = await self.request( + self.login_path, + method=hdrs.METH_POST, + body=self.login_body(), + ) - def parse_update_regulation_mode_response(self, data: Any) -> bool: - """Parse the HTTP response received after updating the regulation mode. + if data["ErrorCode"] == 1: + msg = "Unable to create session, wrong username, password, API key or customer ID provided." # noqa: E501 + raise OJMicrolineAuthError(msg) - Args: - ---- - data: The JSON data contained in the response. + self._session_calls_left = self._session_calls + self._session_id = data["SessionId"] - Returns: True if the update succeeded. + async def get_thermostats(self) -> list[Thermostat]: + """Get all the thermostats. + + Returns + ------- + A list of Thermostats objects. """ + data = await self.request( + self.get_thermostats_path, + method=hdrs.METH_GET, + params={ + "sessionid": self._session_id, + **self.get_thermostats_params(), + }, + ) + thermostats = self.parse_thermostats_response(data) + + for thermostat in thermostats: + thermostat.energy = await self.get_energy_usage(thermostat) + + return thermostats + + async def get_energy_usage(self, resource: Thermostat) -> list[float]: + """Get the energy usage for the provided thermostat.""" + date_tomorrow = (datetime.now(tz=UTC) + timedelta(days=1)).strftime("%Y-%m-%d") + + if self.get_energy_usage_path == "": + return [] + + data = await self.request( + self.get_energy_usage_path, + method=hdrs.METH_POST, + params={ + "sessionid": self._session_id, + }, + body={ + **self.get_thermostats_params(), + "ThermostatID": resource.serial_number, + "ViewType": 2, + "DateTime": date_tomorrow, + "History": 0, + }, + ) + + return self.parse_energy_usage_response(data) + + async def set_regulation_mode( + self, + resource: Thermostat, + regulation_mode: int, + temperature: int | None, + duration: int, + ) -> bool: + """Set the regulation mode.""" + data = await self.request( + self.update_regulation_mode_path, + method=hdrs.METH_POST, + params={ + "sessionid": self._session_id, + **self.update_regulation_mode_params(resource), + }, + body=self.update_regulation_mode_body( + resource, + regulation_mode, + temperature, + duration, + ), + ) + + if not self.parse_update_regulation_mode_response(data): + msg = "Unable to set preset mode." + raise OJMicrolineError(msg) + + return True @dataclass @@ -127,15 +243,6 @@ class OJMicroline: __http_session: ClientSession | None = None __close_http_session: bool = False - # The session ID to perform calls with. - __session_id: str | None = None - - # Reset the login session when this hits zero. - __session_calls_left: int = 0 - - # Maximum number of requests within a single session. - __session_calls: int = 300 - def __init__( self, api: OJMicrolineAPI, session: ClientSession | None = None ) -> None: @@ -148,15 +255,17 @@ def __init__( """ self.__api = api + self.__api.request = self._request self.__http_session = session - async def _request( + async def _request( # pylint: disable=too-many-arguments self, uri: str, *, method: str = hdrs.METH_GET, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, ) -> Any: """Handle a request to the OJ Microline API. @@ -166,6 +275,7 @@ async def _request( method: HTTP method to use, for example, 'GET' params: Extra options to improve or limit the response. body: Data can be used in a POST and PATCH request. + headers: Additional HTTP headers to include. Returns: ------- @@ -184,21 +294,23 @@ async def _request( self.__http_session = ClientSession() self.__close_http_session = True - # Reduce the number of calls left by 1. - self.__session_calls_left -= 1 url = URL.build(scheme="https", host=self.__api.host, path="/").join( URL(uri) ) + request_headers = { + "Content-Type": "application/json; charset=utf-8", + "Accept": "application/json", + } + if headers: + request_headers.update(headers) + async with async_timeout.timeout(self.__request_timeout): response = await self.__http_session.request( method, url, params=params, - headers={ - "Content-Type": "application/json; charset=utf-8", - "Accept": "application/json", - }, + headers=request_headers, json=body, ssl=True, ) @@ -221,28 +333,14 @@ async def _request( return json.loads(await response.text()) async def login(self) -> None: - """Get a valid session to do requests with the OJ Microline API. + """Log in to the OJ Microline API. Raises ------ OJMicrolineAuthError: An error occurred while authenticating. """ - if self.__session_calls_left == 0 or self.__session_id is None: - # Get a new session. - data = await self._request( - self.__api.login_path, - method=hdrs.METH_POST, - body=self.__api.login_body(), - ) - - if data["ErrorCode"] == 1: - msg = "Unable to create session, wrong username, password, API key or customer ID provided." # noqa: E501 - raise OJMicrolineAuthError(msg) - - # Reset the number of session calls. - self.__session_calls_left = self.__session_calls - self.__session_id = data["SessionId"] + await self.__api.login() async def get_thermostats(self) -> list[Thermostat]: """Get all the thermostats. @@ -253,22 +351,7 @@ async def get_thermostats(self) -> list[Thermostat]: """ await self.login() - - data = await self._request( - self.__api.get_thermostats_path, - method=hdrs.METH_GET, - params={ - "sessionid": self.__session_id, - **self.__api.get_thermostats_params(), - }, - ) - thermostats = self.__api.parse_thermostats_response(data) - - # add energy data to each thermostat - for thermostat in thermostats: - thermostat.energy = await self.get_energy_usage(thermostat) - - return thermostats + return await self.__api.get_thermostats() async def get_energy_usage(self, resource: Thermostat) -> list[float]: """Get the energy usage. @@ -288,31 +371,8 @@ async def get_energy_usage(self, resource: Thermostat) -> list[float]: OJMicrolineError: An error occurred while fetching the energy usage. """ - if self.__session_id is None: - await self.login() - - date_tomorrow = (datetime.now(tz=UTC) + timedelta(days=1)).strftime("%Y-%m-%d") - - # If no path is defined, return an empty list. - if self.__api.get_energy_usage_path == "": - return [] - - data = await self._request( - self.__api.get_energy_usage_path, - method=hdrs.METH_POST, - params={ - "sessionid": self.__session_id, - }, - body={ - **self.__api.get_thermostats_params(), - "ThermostatID": resource.serial_number, - "ViewType": 2, - "DateTime": date_tomorrow, - "History": 0, - }, - ) - - return self.__api.parse_energy_usage_response(data) + await self.login() + return await self.__api.get_energy_usage(resource) async def set_regulation_mode( self, @@ -344,35 +404,15 @@ async def set_regulation_mode( OJMicrolineError: An error occurred while setting the regulation mode. """ - if self.__session_id is None: - await self.login() - - data = await self._request( - self.__api.update_regulation_mode_path, - method=hdrs.METH_POST, - params={ - "sessionid": self.__session_id, - **self.__api.update_regulation_mode_params(resource), - }, - body=self.__api.update_regulation_mode_body( - resource, - regulation_mode, - temperature, - duration, - ), + await self.login() + return await self.__api.set_regulation_mode( + resource, regulation_mode, temperature, duration ) - if not self.__api.parse_update_regulation_mode_response(data): - msg = "Unable to set preset mode." - raise OJMicrolineError(msg) - - return True - async def close(self) -> None: """Close open client session.""" if self.__http_session and self.__close_http_session: self.__close_http_session = False - self.__session_id = None await self.__http_session.close() async def __aenter__(self) -> Self: diff --git a/ojmicroline_thermostat/wd5.py b/ojmicroline_thermostat/wd5.py index 8f5b3e51..df75239d 100644 --- a/ojmicroline_thermostat/wd5.py +++ b/ojmicroline_thermostat/wd5.py @@ -17,11 +17,11 @@ ) from .exceptions import OJMicrolineResultsError from .models import Thermostat -from .ojmicroline import OJMicrolineAPI +from .ojmicroline import SessionOJMicrolineAPI @dataclass -class WD5API(OJMicrolineAPI): +class WD5API(SessionOJMicrolineAPI): """Controls OJ Microline WD5-series thermostats (OWD5, MWD5, etc.).""" def __init__( diff --git a/ojmicroline_thermostat/wg4.py b/ojmicroline_thermostat/wg4.py index 38092b1e..8f378326 100644 --- a/ojmicroline_thermostat/wg4.py +++ b/ojmicroline_thermostat/wg4.py @@ -9,11 +9,11 @@ from .const import REGULATION_COMFORT, REGULATION_MANUAL from .models import Thermostat -from .ojmicroline import OJMicrolineAPI +from .ojmicroline import SessionOJMicrolineAPI @dataclass -class WG4API(OJMicrolineAPI): +class WG4API(SessionOJMicrolineAPI): """Controls OJ Microline WG4-series thermostats (UWG4, AWG4, etc.).""" def __init__( diff --git a/ojmicroline_thermostat/wg5.py b/ojmicroline_thermostat/wg5.py new file mode 100644 index 00000000..a83f56d1 --- /dev/null +++ b/ojmicroline_thermostat/wg5.py @@ -0,0 +1,291 @@ +"""Implementation of OJMicrolineAPI for WG5-series thermostats.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import TYPE_CHECKING, Any + +import async_timeout +from aiohttp import ClientSession +from yarl import URL + +from .const import ( + REGULATION_COMFORT, + REGULATION_FROST_PROTECTION, + REGULATION_MANUAL, + REGULATION_SCHEDULE, + REGULATION_VACATION, + WG5_MODE_HOLD, + WG5_MODE_SCHEDULE, +) +from .exceptions import OJMicrolineAuthError, OJMicrolineError +from .models import Thermostat + +if TYPE_CHECKING: + from .ojmicroline import RequestFunc + +WG5_SCOPES = ( + "ugw.zones ugw.users ugw.profile ugw.buildings ugw.schedules " + "ugw.thermostats ugw.firmware ugw.linking openid role offline_access" +) + + +class WG5API: + """Controls OJ Microline WG5-series thermostats.""" + + request: RequestFunc + + def __init__( + self, + username: str, + password: str, + host: str = "user-api.ojmicroline.com", + identity_host: str = "identity.ojmicroline.com", + ) -> None: + """Create a new instance of the API object. + + Args: + ---- + username: The username to log in with. + password: The password for the username. + host: The host name used for API requests. + identity_host: The host name used for OAuth2 authentication. + + """ + self.username = username + self.password = password + self.host = host + self.identity_host = identity_host + self.client_id = "mobile_app_client" + self._access_token: str | None = None + self._refresh_token: str | None = None + self._token_expiry: datetime | None = None + + async def login(self) -> None: + """Authenticate via OAuth2 Resource Owner Password Credentials grant. + + Raises + ------ + OJMicrolineAuthError: Authentication failed. + + """ + if ( + self._access_token + and self._token_expiry + and self._token_expiry > datetime.now(tz=UTC) + ): + return + + async with ClientSession() as session: + url = URL.build( + scheme="https", host=self.identity_host, path="/connect/token" + ) + async with async_timeout.timeout(30): + response = await session.post( + url, + data={ + "grant_type": "password", + "username": self.username, + "password": self.password, + "client_id": self.client_id, + "scope": WG5_SCOPES, + }, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, + ssl=True, + ) + + if response.status != 200: + msg = "Unable to authenticate, wrong username or password." + raise OJMicrolineAuthError(msg) + + data = await response.json() + + self._access_token = data["access_token"] + self._refresh_token = data.get("refresh_token") + self._token_expiry = datetime.now(tz=UTC) + timedelta( + seconds=data["expires_in"] + ) + + async def _auth_request( + self, + uri: str, + *, + method: str = "GET", + body: dict[str, Any] | None = None, + ) -> Any: + """Make an authenticated request with Bearer token.""" + return await self.request( + uri, + method=method, + body=body, + headers={"Authorization": f"Bearer {self._access_token}"}, + ) + + async def _is_away_active(self, building_id: str) -> bool: + """Check if away mode is active for a building.""" + data = await self._auth_request(f"awaymode/{building_id}") + return "data" in data + + async def _fetch_thermostat( + self, + thermostat_id: str, + building_id: str, + zone_name: str, + *, + away_active: bool, + ) -> Thermostat: + """Fetch and enrich a single thermostat.""" + control_data = await self._auth_request( + f"thermostats/{thermostat_id}/control", + ) + detail_data = await self._auth_request( + f"thermostats/{thermostat_id}", + ) + schedule_id = control_data["data"].get("scheduleData", {}).get("scheduleId") + schedule_data = None + if schedule_id: + resp = await self._auth_request(f"schedules/{schedule_id}") + schedule_data = resp.get("data") + thermostat = Thermostat.from_wg5_json( + control_data["data"], + building_id=building_id, + zone_name=zone_name, + away_active=away_active, + ) + readouts = detail_data.get("data", {}).get("thermostatReadouts", {}) + thermostat.software_version = readouts.get("softwareVersionNumber", "") + floor = readouts.get("floorTemperature") + if floor is not None: + thermostat.temperature_floor = round(float(floor) * 100) + room = readouts.get("roomTemperature") + if room is not None: + thermostat.temperature_room = round(float(room) * 100) + thermostat.schedule = schedule_data + return thermostat + + async def get_thermostats(self) -> list[Thermostat]: + """Fetch all thermostats across all buildings.""" + buildings_data = await self._auth_request("buildings") + thermostats: list[Thermostat] = [] + + for building in buildings_data["data"]: + building_id = building["id"] + away_active = await self._is_away_active(building_id) + tree_data = await self._auth_request(f"buildings/{building_id}/tree") + for zone in tree_data["data"]["zones"]: + zone_name = zone["name"] + for thermostat_stub in zone["thermostats"]: + thermostat = await self._fetch_thermostat( + thermostat_stub["id"], + building_id, + zone_name, + away_active=away_active, + ) + thermostats.append(thermostat) + + return thermostats + + async def get_energy_usage(self, resource: Thermostat) -> list[float]: + """Fetch energy usage for a thermostat.""" + if not resource.building_id: + return [] + + now = datetime.now(tz=UTC) + start = now.replace(hour=0, minute=0, second=0, microsecond=0) + + data = await self._auth_request( + f"energy/usage/building/{resource.building_id}", + method="POST", + body={ + "ThermostatIds": [resource.serial_number], + "Start": start.strftime("%Y-%m-%dT%H:%M:%SZ"), + "End": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "Size": 0, + }, + ) + + return [ + float(bucket.get("consumedWattHours", 0)) + for bucket in data["data"]["histogram"] + ] + + async def set_regulation_mode( + self, + resource: Thermostat, + regulation_mode: int, + temperature: int | None, + duration: int, + ) -> bool: + """Set the regulation mode on a thermostat.""" + thermostat_id = resource.serial_number + + if regulation_mode == REGULATION_VACATION: + await self._auth_request( + "awaymode", + method="POST", + body={ + "BuildingId": resource.building_id, + "TemperatureLimitMax": 5.0, + "StartTime": datetime.now(tz=UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), + "EndTime": (datetime.now(tz=UTC) + timedelta(days=365)).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ), + "IsPlanned": False, + "ChangeTimestamp": "0001-01-01T00:00:00", + }, + ) + return True + + if regulation_mode == REGULATION_FROST_PROTECTION: + await self._auth_request( + f"thermostats/{thermostat_id}/standby/True", + method="PUT", + ) + return True + + # Disable away mode if currently active. + if resource.vacation_mode: + await self._auth_request( + f"awaymode/{resource.building_id}", + method="DELETE", + ) + + # Take out of standby if needed. + if resource.is_in_standby: + await self._auth_request( + f"thermostats/{thermostat_id}/standby/False", + method="PUT", + ) + + if regulation_mode == REGULATION_SCHEDULE: + body: dict[str, Any] = { + "ModeAction": WG5_MODE_SCHEDULE, + "ChangeTimestamp": "0001-01-01T00:00:00", + } + elif regulation_mode in (REGULATION_MANUAL, REGULATION_COMFORT): + setpoint = (temperature or resource.set_point_temperature or 0) / 100 + body = { + "ModeAction": WG5_MODE_HOLD, + "Setpoint": setpoint, + "ChangeTimestamp": "0001-01-01T00:00:00", + } + if regulation_mode == REGULATION_COMFORT: + hold_end = datetime.now(tz=UTC) + timedelta(minutes=duration) + body["HoldUntilEndTime"] = hold_end.strftime("%Y-%m-%dT%H:%M:%SZ") + body["HoldIsPermanent"] = False + else: + body["HoldUntilEndTime"] = None + body["HoldIsPermanent"] = True + else: + msg = f"Unsupported regulation mode for WG5: {regulation_mode}" + raise OJMicrolineError(msg) + + await self._auth_request( + f"thermostats/{thermostat_id}/mode", + method="PUT", + body=body, + ) + return True diff --git a/pyproject.toml b/pyproject.toml index e58d597e..8e0aed29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "ojmicroline-thermostat" -version = "3.3.0" +version = "4.0.0" description = "Asynchronous Python client controlling an OJ Microline Thermostat." authors = ["Robbin Janssen "] maintainers = ["Robbin Janssen "] diff --git a/tests/fixtures/wg5_building_tree.json b/tests/fixtures/wg5_building_tree.json new file mode 100644 index 00000000..9dc0be83 --- /dev/null +++ b/tests/fixtures/wg5_building_tree.json @@ -0,0 +1,22 @@ +{ + "data": { + "zones": [ + { + "thermostats": [ + { + "id": "2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "name": "Bathroom" + } + ], + "id": "ae85d8fc-1b4a-445a-bdf1-05f970fc3da4", + "name": "Default" + } + ], + "id": "57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "name": "My Home" + }, + "status": { + "code": "OK", + "traceId": "6835247adfb2e0d1ef8679cce39f1606" + } +} diff --git a/tests/fixtures/wg5_buildings.json b/tests/fixtures/wg5_buildings.json new file mode 100644 index 00000000..200b3501 --- /dev/null +++ b/tests/fixtures/wg5_buildings.json @@ -0,0 +1,20 @@ +{ + "page": { + "total": 1, + "totalPages": 1, + "size": 50, + "number": 0 + }, + "data": [ + { + "isDefault": true, + "userId": "87c522ce-f45f-4993-bf2a-74677d7114f9", + "id": "57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "name": "My Home" + } + ], + "status": { + "code": "OK", + "traceId": "bb18845ebf596416875793381d5bb28c" + } +} diff --git a/tests/fixtures/wg5_energy.json b/tests/fixtures/wg5_energy.json new file mode 100644 index 00000000..d27832ca --- /dev/null +++ b/tests/fixtures/wg5_energy.json @@ -0,0 +1,36 @@ +{ + "data": { + "histogram": [ + { + "consumedWattHours": 150, + "cost": 0.015, + "bucketStart": "2026-04-13T00:00:00Z" + }, + { + "consumedWattHours": 200, + "cost": 0.020, + "bucketStart": "2026-04-13T01:00:00Z" + }, + { + "consumedWattHours": 0, + "cost": 0.0, + "bucketStart": "2026-04-13T02:00:00Z" + } + ], + "priceSettings": { + "isDefault": true, + "price": 0.1, + "currency": "USD" + }, + "floorLoadSettings": [ + { + "thermostatId": "2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "floorLoad": 100 + } + ] + }, + "status": { + "code": "OK", + "traceId": "abc123" + } +} diff --git a/tests/fixtures/wg5_schedule.json b/tests/fixtures/wg5_schedule.json new file mode 100644 index 00000000..38929fc1 --- /dev/null +++ b/tests/fixtures/wg5_schedule.json @@ -0,0 +1,61 @@ +{ + "data": { + "baseTemperature": 23, + "schedules": [ + { + "days": [ + "Monday", + "Tuesday", + "Wednesday", + "Thursday", + "Friday" + ], + "schedules": [ + { + "start": "06:00:00", + "end": "09:00:00", + "temperature": 28 + }, + { + "start": "17:00:00", + "end": "23:00:00", + "temperature": 28 + } + ] + }, + { + "days": [ + "Saturday" + ], + "schedules": [ + { + "start": "08:00:00", + "end": "23:00:00", + "temperature": 28 + } + ] + }, + { + "days": [ + "Sunday" + ], + "schedules": [ + { + "start": "08:00:00", + "end": "23:00:00", + "temperature": 28 + } + ] + } + ], + "scheduleType": "WorkingDaysAndHolidaysSeparately", + "id": "f9aac20b-4e45-415b-9f56-e53014ee8639", + "name": "Default Schedule", + "creationDate": "2026-04-05T11:26:36Z", + "defaultScheduleType": "FloorSensor" + }, + "status": { + "code": "OK", + "traceId": "a5d720e9638ffae0e4fe322a7242fb1c" + } +} diff --git a/tests/fixtures/wg5_thermostat_control.json b/tests/fixtures/wg5_thermostat_control.json new file mode 100644 index 00000000..4a75b725 --- /dev/null +++ b/tests/fixtures/wg5_thermostat_control.json @@ -0,0 +1,40 @@ +{ + "data": { + "id": "2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "zoneId": "ae85d8fc-1b4a-445a-bdf1-05f970fc3da4", + "name": "Bathroom", + "createdDate": "2026-04-05T11:29:17Z", + "currentTemperature": 19, + "currentTemperatureFahrenheit": 66, + "minimumPossibleSetPoint": 5, + "maximumPossibleSetPoint": 40, + "isOnline": true, + "isAdaptiveHeatingActive": false, + "errorFlags": "NoErrors", + "hasError": false, + "isOpenWindowDetected": false, + "isHeatRelayActive": true, + "isFrostProtectionEnabled": true, + "frostProtectionTemperature": 5, + "isFollowingZone": false, + "mode": { + "isInStandby": false, + "isAwayActive": false, + "fallbackMode": "Manual", + "manualSetpoint": 27.78, + "changeTimestamp": "2026-04-13T23:40:43Z" + }, + "scheduleData": { + "scheduleId": "f9aac20b-4e45-415b-9f56-e53014ee8639", + "scheduleName": "Default Schedule", + "currentSetPoint": 23 + }, + "sensorApplication": "Floor", + "setpoint": 27.78, + "changeTimestamp": "2026-04-13T23:40:43Z" + }, + "status": { + "code": "OK", + "traceId": "3bbdb398bee4254f0801119ffe38a07c" + } +} diff --git a/tests/fixtures/wg5_thermostat_detail.json b/tests/fixtures/wg5_thermostat_detail.json new file mode 100644 index 00000000..14fda869 --- /dev/null +++ b/tests/fixtures/wg5_thermostat_detail.json @@ -0,0 +1,110 @@ +{ + "data": { + "id": "2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "name": "Bathroom", + "userId": "87c522ce-f45f-4993-bf2a-74677d7114f9", + "zoneId": "ae85d8fc-1b4a-445a-bdf1-05f970fc3da4", + "zoneName": "Default", + "buildingId": "57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "buildingName": "My Home", + "isOnline": true, + "hasError": false, + "hasUpdates": true, + "errors": "NoErrors", + "isFloorSensorAvailable": true, + "isAwayActive": false, + "serialNumber": "300030132", + "isAdaptiveHeatingActive": false, + "isOpenWindowDetected": false, + "isHeatRelayActive": false, + "distributorId": 10031, + "plConfigFile": "1047AA103", + "thermostatDisplaySettings": { + "temperatureUnit": "Fahrenheit", + "screensaver": "ActualTemperature", + "screenLockEnabled": false, + "displayOn": true, + "brightnessNormal": 6, + "brightnessScreensaver": 3, + "changeTimestamp": "2025-08-05T14:13:36Z" + }, + "thermostatHeatingSettings": { + "floorLoad": 100, + "floorSensorType": "Type10KOhm", + "floorSensorOffset": 0, + "roomSensorOffset": 0, + "isAdaptiveHeatingEnabled": false, + "isOpenWindowDetectionEnabled": true, + "changeTimestamp": "2026-04-14T01:17:18Z" + }, + "thermostatFloorSensorSettings": { + "value15deg": 4714, + "value20deg": 3749, + "value25deg": 3000, + "value30deg": 2417, + "changeTimestamp": "2025-08-05T14:13:36Z" + }, + "thermostatRegulationSettings": { + "sensorApplication": "Floor", + "temperatureLimitMax": 40, + "temperatureLimitMin": 5, + "floorType": "Tile", + "floorProtectionMax": 40, + "floorProtectionMin": 0, + "frostProtectionTemperature": 5, + "frostProtectionEnabled": true, + "changeTimestamp": "2026-04-05T04:20:33Z" + }, + "thermostatDateTimeSettings": { + "dst": "NorthAmerica", + "timeZoneMinutes": -300, + "changeTimestamp": "2026-04-05T11:43:35Z" + }, + "thermostatWifiSettings": { + "ssid": "TestWifi", + "authentication": "Wpa2", + "defaultGateway": "192.168.4.1", + "netmask": "255.255.252.0", + "ipv4": "192.168.4.24", + "ipv6": "fe80::223:38ff:fe62:d449", + "dhcp": false, + "macWifi": "00:23:38:62:D4:49", + "macBle": "00:23:38:62:D4:4A", + "preferedWifiChannel": 0, + "numberOfTimesWifiApConnectWasCalled": 7, + "bleState": 1, + "changeTimestamp": "2026-04-13T01:38:30Z" + }, + "thermostatReadouts": { + "softwareVersionNumber": "1047A108", + "productType": "UWG5", + "floorTemperature": 19, + "floorTemperatureFahrenheit": 67, + "roomTemperature": 21, + "roomTemperatureFahrenheit": 70, + "currentTime": "2026-04-14T01:31:11Z", + "distributorId": "10031", + "wifiSoftwareVersion": "6.02.00-004" + }, + "mode": { + "isInStandby": true, + "isAwayActive": false, + "fallbackMode": "Auto", + "manualSetpoint": 26, + "changeTimestamp": "0001-01-01T00:00:00" + }, + "scheduleData": { + "scheduleId": "f9aac20b-4e45-415b-9f56-e53014ee8639", + "scheduleName": "Default Schedule", + "nextEventStartTime": "23:00:00", + "nextEventDayOfWeek": "Monday", + "nextEventSetPoint": 23, + "currentSetPoint": 28 + }, + "setpoint": 28 + }, + "status": { + "code": "OK", + "traceId": "d078903141fa63e29fff471c0e3c514f" + } +} diff --git a/tests/fixtures/wg5_token.json b/tests/fixtures/wg5_token.json new file mode 100644 index 00000000..2a2a4202 --- /dev/null +++ b/tests/fixtures/wg5_token.json @@ -0,0 +1,7 @@ +{ + "access_token": "eyJ0ZXN0IjoiZmFrZS1hY2Nlc3MtdG9rZW4ifQ", + "expires_in": 3600, + "token_type": "Bearer", + "refresh_token": "fake-refresh-token-for-testing", + "scope": "offline_access openid role ugw.buildings ugw.firmware ugw.linking ugw.profile ugw.schedules ugw.thermostats ugw.users ugw.zones" +} diff --git a/tests/ruff.toml b/tests/ruff.toml index 60b7bf55..08727e59 100644 --- a/tests/ruff.toml +++ b/tests/ruff.toml @@ -8,6 +8,7 @@ lint.extend-select = [ lint.extend-ignore = [ "S101", # Use of assert detected. As these are tests "SLF001", # Tests will access private/protected members + "S105", # Ignore possible hardcoded passwords "S106", # Ignore possible passwords "ANN001", # Ignore missing annotations ] diff --git a/tests/test_model_thermostat.py b/tests/test_model_thermostat.py index db8e3dbd..da2a68d5 100644 --- a/tests/test_model_thermostat.py +++ b/tests/test_model_thermostat.py @@ -207,6 +207,122 @@ async def test_thermostat_from_json_wd5_timezone_negative() -> None: assert thermostat.comfort_end_time.strftime("%z") == "-0200" +@pytest.mark.asyncio +async def test_thermostat_from_json_wg5() -> None: + """Make sure the data is accepted by the from_wg5_json method.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + ) + + for field in WG5_ONLY_FIELDS: + assert getattr(thermostat, field) is not None, ( + f"Expected {field} to be non-null" + ) + for field in WD5_EXCLUSIVE_FIELDS: + assert getattr(thermostat, field) is None, f"Expected {field} to be null" + + assert thermostat.model == "WG5" + assert thermostat.serial_number == "2cb3e6c5-8cf8-4e7e-943a-42618c34a506" + assert thermostat.software_version == "" + assert thermostat.zone_name == "Default" + assert thermostat.zone_id == 0 + assert thermostat.zone_uuid == "ae85d8fc-1b4a-445a-bdf1-05f970fc3da4" + assert thermostat.building_id == "57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a" + assert thermostat.name == "Bathroom" + assert thermostat.online is True + assert thermostat.heating is True + assert thermostat.regulation_mode == REGULATION_MANUAL + assert thermostat.supported_regulation_modes == [ + REGULATION_SCHEDULE, + REGULATION_MANUAL, + REGULATION_COMFORT, + REGULATION_VACATION, + REGULATION_FROST_PROTECTION, + ] + assert thermostat.min_temperature == 500 + assert thermostat.max_temperature == 4000 + assert thermostat.temperature == 1900 + assert thermostat.set_point_temperature == 2778 + assert thermostat.manual_temperature == 2778 + assert thermostat.comfort_temperature == 2778 + assert thermostat.frost_protection_temperature == 500 + assert thermostat.sensor_mode == SENSOR_FLOOR + assert thermostat.vacation_mode is False + assert thermostat.is_in_standby is False + assert thermostat.open_window_detection is False + assert thermostat.schedule_id == "f9aac20b-4e45-415b-9f56-e53014ee8639" + assert thermostat.schedule_name == "Default Schedule" + + # Test the getter methods: + assert thermostat.get_current_temperature() == 1900 + assert thermostat.get_target_temperature() == 2778 + + +@pytest.mark.asyncio +async def test_thermostat_from_json_wg5_standby() -> None: + """Make sure standby mode maps to REGULATION_FROST_PROTECTION.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + data["data"]["mode"]["isInStandby"] = True + + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + ) + + assert thermostat.regulation_mode == REGULATION_FROST_PROTECTION + assert thermostat.is_in_standby is True + + +@pytest.mark.asyncio +async def test_thermostat_from_json_wg5_away() -> None: + """Make sure away_active maps to REGULATION_VACATION.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + away_active=True, + ) + + assert thermostat.regulation_mode == REGULATION_VACATION + assert thermostat.vacation_mode is True + + +@pytest.mark.asyncio +async def test_thermostat_from_json_wg5_schedule() -> None: + """Make sure fallbackMode Auto maps to REGULATION_SCHEDULE.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + data["data"]["mode"]["fallbackMode"] = "Auto" + + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + ) + + assert thermostat.regulation_mode == REGULATION_SCHEDULE + assert thermostat.last_primary_mode_is_auto is True + assert thermostat.is_in_standby is False + + +@pytest.mark.asyncio +async def test_thermostat_get_current_energy_with_data() -> None: + """Test get_current_energy returns the first element when energy is set.""" + data = json.loads(load_fixtures("wd5_thermostat.json")) + thermostat = Thermostat.from_wd5_json(data) + thermostat.energy = [3.5, 2.1, 1.0] + + assert thermostat.get_current_energy() == 3.5 + + thermostat.energy = None + assert thermostat.get_current_energy() == 0.0 + + REQUIRED_FIELDS = [ "model", "serial_number", @@ -248,3 +364,28 @@ async def test_thermostat_from_json_wd5_timezone_negative() -> None: "frost_protection_temperature", "boost_temperature", ] + +WD5_EXCLUSIVE_FIELDS = [ + "thermostat_id", + "schedule", + "adaptive_mode", + "daylight_saving_active", + "temperature_floor", + "temperature_room", + "boost_end_time", + "boost_temperature", +] + +WG5_ONLY_FIELDS = [ + "building_id", + "zone_uuid", + "is_in_standby", + "schedule_id", + "schedule_name", + "temperature", + "set_point_temperature", + "sensor_mode", + "open_window_detection", + "frost_protection_temperature", + "vacation_mode", +] diff --git a/tests/test_ojmicroline.py b/tests/test_ojmicroline.py index 5766cf46..21b0f1ff 100644 --- a/tests/test_ojmicroline.py +++ b/tests/test_ojmicroline.py @@ -46,21 +46,15 @@ async def test_json_request(aresponses: ResponsesMockServer) -> None: assert response is not None await client.close() assert client._OJMicroline__close_http_session is False - assert client._OJMicroline__session_id is None @pytest.mark.asyncio async def test_timeout(monkeypatch, aresponses: ResponsesMockServer) -> None: """Test request timeout.""" - async def response_handler(_: aiohttp.ClientResponse) -> Response: + async def response_handler(_: aiohttp.ClientResponse) -> None: # Faking a timeout by sleeping await asyncio.sleep(0.2) - return Response( - status=200, - headers={"Content-Type": "application/json"}, - text=json.dumps({"ErrorCode": 0}), - ) aresponses.add("ojmicroline.test.host", "/test", "GET", response_handler) diff --git a/tests/test_wd5.py b/tests/test_wd5.py index bd780410..1fe39586 100644 --- a/tests/test_wd5.py +++ b/tests/test_wd5.py @@ -35,23 +35,18 @@ async def test_login(aresponses: ResponsesMockServer) -> None: ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) + client = OJMicroline(api=api, session=session) await client.login() - assert ( - client._OJMicroline__session_calls_left - == client._OJMicroline__session_calls - ) - assert client._OJMicroline__session_id == "f00br4" + assert api._session_calls_left == api._session_calls + assert api._session_id == "f00br4" @pytest.mark.asyncio @@ -68,26 +63,24 @@ async def test_login_failed(aresponses: ResponsesMockServer) -> None: ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) + client = OJMicroline(api=api, session=session) with pytest.raises(OJMicrolineAuthError): await client.login() - assert client._OJMicroline__session_calls_left == -1 - assert client._OJMicroline__session_id is None + assert api._session_calls_left == -1 + assert api._session_id is None @pytest.mark.asyncio -async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> None: +async def test_get_thermostats(aresponses: ResponsesMockServer) -> None: """Test get thermostats function and make sure fields are set.""" aresponses.add( "ojmicroline.test.host", @@ -121,19 +114,17 @@ async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") thermostats: list[Thermostat] = await client.get_thermostats() assert thermostats is not None @@ -142,9 +133,7 @@ async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> @pytest.mark.asyncio -async def test_get_thermostats_failed( - monkeypatch, aresponses: ResponsesMockServer -) -> None: +async def test_get_thermostats_failed(aresponses: ResponsesMockServer) -> None: """Test get thermostats function to handle an error.""" aresponses.add( "ojmicroline.test.host", @@ -157,30 +146,23 @@ async def test_get_thermostats_failed( ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) - - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) with pytest.raises(OJMicrolineResultsError): await client.get_thermostats() - assert client._OJMicroline__session_calls_left == 299 - @pytest.mark.asyncio -async def test_set_regulation_mode( - monkeypatch, aresponses: ResponsesMockServer -) -> None: +async def test_set_regulation_mode(aresponses: ResponsesMockServer) -> None: """Test updating the regulation mode.""" aresponses.add( "ojmicroline.test.host", @@ -196,19 +178,17 @@ async def test_set_regulation_mode( data = load_fixtures("wd5_thermostat.json") thermostat = Thermostat.from_wd5_json(json.loads(data)) - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") result = await client.set_regulation_mode(thermostat, REGULATION_MANUAL, 2500) assert result is True @@ -216,7 +196,7 @@ async def test_set_regulation_mode( @pytest.mark.asyncio async def test_set_regulation_mode_expect_login( - monkeypatch, aresponses: ResponsesMockServer + aresponses: ResponsesMockServer, ) -> None: """Test update the regulation mode with login method fired.""" aresponses.add( @@ -230,35 +210,63 @@ async def test_set_regulation_mode_expect_login( ), ) async with aiohttp.ClientSession() as session: - - def set_session_id() -> None: - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", + ) data = load_fixtures("wd5_thermostat.json") thermostat = Thermostat.from_wd5_json(json.loads(data)) + def set_session_id() -> None: + api._session_id = "f00b4r" + with patch.object( OJMicroline, "login", side_effect=set_session_id ) as mock_login: - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, - ) + client = OJMicroline(api=api, session=session) await client.set_regulation_mode(thermostat, REGULATION_MANUAL, None) mock_login.assert_called_once() @pytest.mark.asyncio -async def test_set_regulation_mode_failed( - monkeypatch, aresponses: ResponsesMockServer -) -> None: +async def test_get_energy_usage_failed(aresponses: ResponsesMockServer) -> None: + """Test energy usage when the API returns an error.""" + aresponses.add( + "ojmicroline.test.host", + "/api/EnergyUsage/GetEnergyUsage", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"ErrorCode": 1}), + ), + ) + async with aiohttp.ClientSession() as session: + data = load_fixtures("wd5_thermostat.json") + thermostat = Thermostat.from_wd5_json(json.loads(data)) + + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", + ) + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) + + with pytest.raises(OJMicrolineResultsError): + await client.get_energy_usage(thermostat) + + +@pytest.mark.asyncio +async def test_set_regulation_mode_failed(aresponses: ResponsesMockServer) -> None: """Test update the regulation mode when an error occurs.""" aresponses.add( "ojmicroline.test.host", @@ -274,21 +282,16 @@ async def test_set_regulation_mode_failed( data = load_fixtures("wd5_thermostat.json") thermostat = Thermostat.from_wd5_json(json.loads(data)) - client = OJMicroline( - api=WD5API( - host="ojmicroline.test.host", - api_key="ap1-k3y-v3ry-s3cret", - customer_id=1337, - username="py", - password="test", - ), - session=session, + api = WD5API( + host="ojmicroline.test.host", + api_key="ap1-k3y-v3ry-s3cret", + customer_id=1337, + username="py", + password="test", ) - - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) with pytest.raises(OJMicrolineError): await client.set_regulation_mode(thermostat, REGULATION_COMFORT, 2500, 360) - - assert client._OJMicroline__session_calls_left == 299 diff --git a/tests/test_wg4.py b/tests/test_wg4.py index 41b2a450..2c582abe 100644 --- a/tests/test_wg4.py +++ b/tests/test_wg4.py @@ -38,21 +38,16 @@ async def test_login(aresponses: ResponsesMockServer) -> None: ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", ) + client = OJMicroline(api=api, session=session) await client.login() - assert ( - client._OJMicroline__session_calls_left - == client._OJMicroline__session_calls - ) - assert client._OJMicroline__session_id == "f00br4" + assert api._session_calls_left == api._session_calls + assert api._session_id == "f00br4" @pytest.mark.asyncio @@ -69,24 +64,22 @@ async def test_login_failed(aresponses: ResponsesMockServer) -> None: ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", ) + client = OJMicroline(api=api, session=session) with pytest.raises(OJMicrolineAuthError): await client.login() - assert client._OJMicroline__session_calls_left == -1 - assert client._OJMicroline__session_id is None + assert api._session_calls_left == -1 + assert api._session_id is None @pytest.mark.asyncio -async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> None: +async def test_get_thermostats(aresponses: ResponsesMockServer) -> None: """Test get thermostats function and make sure fields are set.""" aresponses.add( "ojmicroline.test.host", @@ -99,17 +92,15 @@ async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> ), ) async with aiohttp.ClientSession() as session: - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", ) + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") thermostats: list[Thermostat] = await client.get_thermostats() assert thermostats is not None @@ -119,9 +110,7 @@ async def test_get_thermostats(monkeypatch, aresponses: ResponsesMockServer) -> @pytest.mark.asyncio -async def test_set_regulation_mode( - monkeypatch, aresponses: ResponsesMockServer -) -> None: +async def test_set_regulation_mode(aresponses: ResponsesMockServer) -> None: """Test updating the regulation mode.""" aresponses.add( "ojmicroline.test.host", @@ -137,17 +126,15 @@ async def test_set_regulation_mode( data = load_fixtures("wg4_thermostat.json") thermostat = Thermostat.from_wg4_json(json.loads(data)) - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", ) + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") result = await client.set_regulation_mode(thermostat, REGULATION_MANUAL, 2500) assert result is True @@ -155,7 +142,7 @@ async def test_set_regulation_mode( @pytest.mark.asyncio async def test_set_regulation_mode_expect_login( - monkeypatch, aresponses: ResponsesMockServer + aresponses: ResponsesMockServer, ) -> None: """Test update the regulation mode with login method fired.""" aresponses.add( @@ -169,33 +156,29 @@ async def test_set_regulation_mode_expect_login( ), ) async with aiohttp.ClientSession() as session: - - def set_session_id() -> None: - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", + ) data = load_fixtures("wg4_thermostat.json") thermostat = Thermostat.from_wg4_json(json.loads(data)) + def set_session_id() -> None: + api._session_id = "f00b4r" + with patch.object( OJMicroline, "login", side_effect=set_session_id ) as mock_login: - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, - ) + client = OJMicroline(api=api, session=session) await client.set_regulation_mode(thermostat, REGULATION_SCHEDULE) mock_login.assert_called_once() @pytest.mark.asyncio -async def test_set_regulation_mode_failed( - monkeypatch, aresponses: ResponsesMockServer -) -> None: +async def test_set_regulation_mode_failed(aresponses: ResponsesMockServer) -> None: """Test update the regulation mode when an error occurs.""" aresponses.add( "ojmicroline.test.host", @@ -211,19 +194,25 @@ async def test_set_regulation_mode_failed( data = load_fixtures("wg4_thermostat.json") thermostat = Thermostat.from_wg4_json(json.loads(data)) - client = OJMicroline( - api=WG4API( - host="ojmicroline.test.host", - username="py", - password="test", - ), - session=session, + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", ) - - monkeypatch.setattr(client, "_OJMicroline__session_calls_left", 300) - monkeypatch.setattr(client, "_OJMicroline__session_id", "f00b4r") + api._session_calls_left = 300 + api._session_id = "f00b4r" + client = OJMicroline(api=api, session=session) with pytest.raises(OJMicrolineError): await client.set_regulation_mode(thermostat, REGULATION_COMFORT, 2500, 360) - assert client._OJMicroline__session_calls_left == 299 + +@pytest.mark.asyncio +async def test_parse_energy_usage_response() -> None: + """Test that WG4 energy usage response parser returns an empty list.""" + api = WG4API( + host="ojmicroline.test.host", + username="py", + password="test", + ) + assert not api.parse_energy_usage_response({}) diff --git a/tests/test_wg5.py b/tests/test_wg5.py new file mode 100644 index 00000000..91258a67 --- /dev/null +++ b/tests/test_wg5.py @@ -0,0 +1,347 @@ +# pylint: disable=protected-access +# mypy: disable-error-code="attr-defined" +"""Integration test for the WG5API class.""" + +import json +from datetime import UTC, datetime, timedelta + +import aiohttp +import pytest +from aresponses import Response, ResponsesMockServer # type: ignore[import] +from ojmicroline_thermostat import ( + OJMicroline, + OJMicrolineAuthError, + Thermostat, +) +from ojmicroline_thermostat.const import REGULATION_MANUAL +from ojmicroline_thermostat.wg5 import WG5API + +from . import load_fixtures + + +@pytest.mark.asyncio +async def test_login(aresponses: ResponsesMockServer) -> None: + """Test the OAuth2 login method.""" + aresponses.add( + "identity.test.host", + "/connect/token", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_token.json"), + ), + ) + async with aiohttp.ClientSession() as session: + api = WG5API( + username="py", + password="test", + host="ojmicroline.test.host", + identity_host="identity.test.host", + ) + client = OJMicroline(api=api, session=session) + + await client.login() + assert api._access_token == "eyJ0ZXN0IjoiZmFrZS1hY2Nlc3MtdG9rZW4ifQ" + assert api._refresh_token == "fake-refresh-token-for-testing" + + +@pytest.mark.asyncio +async def test_login_failed(aresponses: ResponsesMockServer) -> None: + """Test the OAuth2 login method when it fails.""" + aresponses.add( + "identity.test.host", + "/connect/token", + "POST", + Response( + status=400, + headers={"Content-Type": "application/json"}, + text=json.dumps({"error": "invalid_grant"}), + ), + ) + async with aiohttp.ClientSession() as session: + api = WG5API( + username="py", + password="test", + host="ojmicroline.test.host", + identity_host="identity.test.host", + ) + client = OJMicroline(api=api, session=session) + + with pytest.raises(OJMicrolineAuthError): + await client.login() + + +def _add_login_response(aresponses: ResponsesMockServer) -> None: + """Add a successful login response to the mock server.""" + aresponses.add( + "identity.test.host", + "/connect/token", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_token.json"), + ), + ) + + +def _make_api() -> WG5API: + """Create a WG5API for tests.""" + return WG5API( + username="py", + password="test", + host="ojmicroline.test.host", + identity_host="identity.test.host", + ) + + +@pytest.mark.asyncio +async def test_login_skips_when_token_valid(aresponses: ResponsesMockServer) -> None: + """Test that login is skipped when a valid token exists.""" + _add_login_response(aresponses) + # Only one token response is mocked; a second login call would fail. + aresponses.add( + "ojmicroline.test.host", + "/buildings", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": []}), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/buildings", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": []}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + + # First call triggers login + await client.get_thermostats() + # Second call reuses the cached token (line 79) + await client.get_thermostats() + + +@pytest.mark.asyncio +async def test_get_thermostats(aresponses: ResponsesMockServer) -> None: + """Test fetching thermostats through the buildings->tree->control flow.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/buildings", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_buildings.json"), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/awaymode/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"status": {"code": "OK"}}), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/buildings/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a/tree", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_building_tree.json"), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/control", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_thermostat_control.json"), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_thermostat_detail.json"), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/schedules/f9aac20b-4e45-415b-9f56-e53014ee8639", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_schedule.json"), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + + thermostats: list[Thermostat] = await client.get_thermostats() + + assert len(thermostats) == 1 + thermostat = thermostats[0] + assert thermostat.name == "Bathroom" + assert thermostat.model == "WG5" + assert thermostat.serial_number == "2cb3e6c5-8cf8-4e7e-943a-42618c34a506" + assert thermostat.online is True + assert thermostat.heating is True + assert thermostat.regulation_mode == REGULATION_MANUAL + assert thermostat.temperature == 1900 + assert thermostat.set_point_temperature == 2778 + assert thermostat.building_id == "57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a" + assert thermostat.zone_name == "Default" + assert thermostat.software_version == "1047A108" + assert thermostat.temperature_floor == 1900 + assert thermostat.temperature_room == 2100 + assert thermostat.schedule is not None + assert thermostat.schedule["baseTemperature"] == 23 + assert len(thermostat.schedule["schedules"]) == 3 + + +@pytest.mark.asyncio +async def test_get_thermostats_no_schedule_no_readouts( + aresponses: ResponsesMockServer, +) -> None: + """Test fetching thermostats when schedule and readouts are missing.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/buildings", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_buildings.json"), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/awaymode/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"status": {"code": "OK"}}), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/buildings/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a/tree", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_building_tree.json"), + ), + ) + # Control data with no scheduleId + control_data = json.loads(load_fixtures("wg5_thermostat_control.json")) + control_data["data"]["scheduleData"] = {} + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/control", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps(control_data), + ), + ) + # Detail data with no readout temperatures + detail_data = json.loads(load_fixtures("wg5_thermostat_detail.json")) + del detail_data["data"]["thermostatReadouts"]["floorTemperature"] + del detail_data["data"]["thermostatReadouts"]["roomTemperature"] + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506", + "GET", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps(detail_data), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + + thermostats: list[Thermostat] = await client.get_thermostats() + + assert len(thermostats) == 1 + thermostat = thermostats[0] + assert thermostat.schedule is None + assert thermostat.temperature_floor is None + assert thermostat.temperature_room is None + + +@pytest.mark.asyncio +async def test_get_energy_usage_no_building_id() -> None: + """Test energy usage returns empty list when building_id is missing.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="", + zone_name="Default", + ) + + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + # No login mock needed; should return before making any request + api._access_token = "fake" + api._token_expiry = datetime.now(tz=UTC) + timedelta(hours=1) + + energy = await client.get_energy_usage(thermostat) + assert energy == [] + + +@pytest.mark.asyncio +async def test_get_energy_usage(aresponses: ResponsesMockServer) -> None: + """Test fetching energy usage for a WG5 thermostat.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/energy/usage/building/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_energy.json"), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + thermostat = Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + ) + + energy = await client.get_energy_usage(thermostat) + assert energy == [150.0, 200.0, 0.0] diff --git a/tests/test_wg5_update.py b/tests/test_wg5_update.py new file mode 100644 index 00000000..3094c5e9 --- /dev/null +++ b/tests/test_wg5_update.py @@ -0,0 +1,273 @@ +# pylint: disable=protected-access +# mypy: disable-error-code="attr-defined" +"""Test the WG5API update regulation mode methods.""" + +import json +from datetime import UTC, datetime, timedelta + +import aiohttp +import pytest +from aresponses import Response, ResponsesMockServer # type: ignore[import] +from freezegun import freeze_time +from ojmicroline_thermostat import ( + OJMicroline, + OJMicrolineError, + Thermostat, +) +from ojmicroline_thermostat.const import ( + REGULATION_BOOST, + REGULATION_COMFORT, + REGULATION_FROST_PROTECTION, + REGULATION_MANUAL, + REGULATION_SCHEDULE, + REGULATION_VACATION, +) +from ojmicroline_thermostat.wg5 import WG5API + +from . import load_fixtures + + +def _make_api() -> WG5API: + """Create a WG5API for tests.""" + return WG5API( + username="py", + password="test", + host="ojmicroline.test.host", + identity_host="identity.test.host", + ) + + +def _add_login_response(aresponses: ResponsesMockServer) -> None: + """Add a successful login response to the mock server.""" + aresponses.add( + "identity.test.host", + "/connect/token", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=load_fixtures("wg5_token.json"), + ), + ) + + +def _make_thermostat() -> Thermostat: + """Create a WG5 thermostat for tests.""" + data = json.loads(load_fixtures("wg5_thermostat_control.json")) + return Thermostat.from_wg5_json( + data["data"], + building_id="57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + zone_name="Default", + ) + + +@pytest.mark.asyncio +async def test_set_regulation_mode_manual(aresponses: ResponsesMockServer) -> None: + """Test setting manual mode with a temperature.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/mode", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-13T23:40:43Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + result = await client.set_regulation_mode(thermostat, REGULATION_MANUAL, 2600) + assert result is True + + +@pytest.mark.asyncio +@freeze_time("2026-04-13 23:40:00") +async def test_set_regulation_mode_comfort(aresponses: ResponsesMockServer) -> None: + """Test setting comfort mode with duration.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/mode", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-13T23:40:43Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + result = await client.set_regulation_mode( + thermostat, REGULATION_COMFORT, 2200, 120 + ) + assert result is True + + +@pytest.mark.asyncio +async def test_set_regulation_mode_schedule(aresponses: ResponsesMockServer) -> None: + """Test setting schedule mode.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/mode", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-13T23:40:43Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + result = await client.set_regulation_mode(thermostat, REGULATION_SCHEDULE) + assert result is True + + +@pytest.mark.asyncio +async def test_set_regulation_mode_exits_standby( + aresponses: ResponsesMockServer, +) -> None: + """Test that setting manual mode exits standby first.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/standby/False", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-14T01:00:00Z"}}), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/mode", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-14T01:00:00Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + thermostat.is_in_standby = True + + result = await client.set_regulation_mode(thermostat, REGULATION_MANUAL, 2600) + assert result is True + + +@pytest.mark.asyncio +async def test_set_regulation_mode_unsupported() -> None: + """Test that an unsupported regulation mode raises OJMicrolineError.""" + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + # Set token to avoid login attempt + api._access_token = "fake" + api._token_expiry = datetime.now(tz=UTC) + timedelta(hours=1) + + with pytest.raises(OJMicrolineError, match="Unsupported regulation mode"): + await client.set_regulation_mode(thermostat, REGULATION_BOOST) + + +@pytest.mark.asyncio +async def test_set_regulation_mode_frost_protection( + aresponses: ResponsesMockServer, +) -> None: + """Test setting frost protection (standby) mode.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/standby/True", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-13T23:41:50Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + result = await client.set_regulation_mode( + thermostat, REGULATION_FROST_PROTECTION + ) + assert result is True + + +@pytest.mark.asyncio +async def test_set_regulation_mode_vacation( + aresponses: ResponsesMockServer, +) -> None: + """Test enabling away (vacation) mode.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/awaymode", + "POST", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"status": {"code": "OK"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + + result = await client.set_regulation_mode(thermostat, REGULATION_VACATION) + assert result is True + + +@pytest.mark.asyncio +async def test_set_regulation_mode_disable_vacation( + aresponses: ResponsesMockServer, +) -> None: + """Test switching from vacation to schedule disables away mode.""" + _add_login_response(aresponses) + aresponses.add( + "ojmicroline.test.host", + "/awaymode/57dc7778-4ed3-4382-9e89-5cf2e0bc0f8a", + "DELETE", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"status": {"code": "OK"}}), + ), + ) + aresponses.add( + "ojmicroline.test.host", + "/thermostats/2cb3e6c5-8cf8-4e7e-943a-42618c34a506/mode", + "PUT", + Response( + status=200, + headers={"Content-Type": "application/json"}, + text=json.dumps({"data": {"changeTimestamp": "2026-04-14T01:00:00Z"}}), + ), + ) + async with aiohttp.ClientSession() as session: + api = _make_api() + client = OJMicroline(api=api, session=session) + thermostat = _make_thermostat() + thermostat.vacation_mode = True + + result = await client.set_regulation_mode(thermostat, REGULATION_SCHEDULE) + assert result is True