From e5c78b29e983748a1cea32f6e93ab832a5bcae46 Mon Sep 17 00:00:00 2001 From: Robert Svensson Date: Fri, 6 Oct 2023 23:43:25 +0200 Subject: [PATCH] Rework API request (#238) * Test * Rework basic_device_info * Make basic device info pass mypy and tests * Adapt api discovery * Some more trial * Combine ApiRequest2/3 Change data to content * Skip being too smart for now, api handler classes will need to combine request response parsing * Fix pir sensor * Fix stream profiles * Fix view areas * Fix mqtt * Fix light control * Remove 2 from ApiHandler2 and ApiRequest2 --- axis/vapix/interfaces/api_discovery.py | 18 +- axis/vapix/interfaces/api_handler.py | 18 +- axis/vapix/interfaces/basic_device_info.py | 17 +- axis/vapix/interfaces/light_control.py | 155 ++-- axis/vapix/interfaces/mqtt.py | 30 +- .../interfaces/pir_sensor_configuration.py | 26 +- axis/vapix/interfaces/stream_profiles.py | 19 +- axis/vapix/interfaces/view_areas.py | 30 +- axis/vapix/models/api.py | 39 +- axis/vapix/models/api_discovery.py | 112 ++- axis/vapix/models/basic_device_info.py | 128 ++- axis/vapix/models/light_control.py | 788 ++++++++++++------ axis/vapix/models/mqtt.py | 171 ++-- axis/vapix/models/pir_sensor_configuration.py | 191 +++-- axis/vapix/models/stream_profile.py | 116 ++- axis/vapix/models/view_area.py | 204 +++-- axis/vapix/vapix.py | 15 +- tests/test_api_discovery.py | 7 +- tests/test_basic_device_info.py | 7 +- tests/test_light_control.py | 33 +- tests/test_stream_profiles.py | 4 + tests/test_vapix.py | 14 +- tests/test_view_areas.py | 8 + 23 files changed, 1448 insertions(+), 702 deletions(-) diff --git a/axis/vapix/interfaces/api_discovery.py b/axis/vapix/interfaces/api_discovery.py index 73ec8ab6..ede096e0 100644 --- a/axis/vapix/interfaces/api_discovery.py +++ b/axis/vapix/interfaces/api_discovery.py @@ -5,9 +5,12 @@ """ from ..models.api_discovery import ( + API_VERSION, Api, ApiId, + GetAllApisResponse, GetSupportedVersionsRequest, + GetSupportedVersionsResponse, ListApisRequest, ListApisT, ) @@ -18,13 +21,20 @@ class ApiDiscoveryHandler(ApiHandler[Api]): """API Discovery for Axis devices.""" api_id = ApiId.API_DISCOVERY - api_request = ListApisRequest() + default_api_version = API_VERSION + + async def _api_request(self) -> dict[str, Api]: + """Get default data of API discovery.""" + return await self.get_api_list() async def get_api_list(self) -> ListApisT: """List all APIs registered on API Discovery service.""" - discovery_item = self[self.api_id.value] - return await self.vapix.request2(ListApisRequest(discovery_item.version)) + bytes_data = await self.vapix.new_request(ListApisRequest()) + response = GetAllApisResponse.decode(bytes_data) + return {api.id: api for api in response.data} async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data diff --git a/axis/vapix/interfaces/api_handler.py b/axis/vapix/interfaces/api_handler.py index 4a822e26..dd0e276f 100644 --- a/axis/vapix/interfaces/api_handler.py +++ b/axis/vapix/interfaces/api_handler.py @@ -1,6 +1,6 @@ """API handler class and base class for an API endpoint.""" -from abc import ABC +from abc import ABC, abstractmethod from typing import ( TYPE_CHECKING, Any, @@ -15,14 +15,13 @@ from ..models.api_discovery import ApiId from ..vapix import Vapix -from ..models.api import ApiItemT, ApiRequest +from ..models.api import ApiItemT class ApiHandler(ABC, Generic[ApiItemT]): """Base class for a map of API Items.""" api_id: "ApiId" - api_request: ApiRequest[dict[str, ApiItemT]] | None default_api_version: str | None = None def __init__(self, vapix: "Vapix") -> None: @@ -35,18 +34,23 @@ def supported(self) -> bool: """Is API supported by the device.""" return self.api_id.value in self.vapix.api_discovery + @property def api_version(self) -> str | None: """Latest API version supported.""" - if (discovery_item := self.vapix.api_discovery[self.api_id.value]) is not None: + if ( + discovery_item := self.vapix.api_discovery.get(self.api_id.value) + ) is not None: return discovery_item.version return self.default_api_version + @abstractmethod + async def _api_request(self) -> dict[str, ApiItemT]: + """Get API data method defined by subsclass.""" + async def update(self) -> None: """Refresh data.""" - if self.api_request is None: - return + self._items = await self._api_request() self.initialized = True - self._items = await self.vapix.request2(self.api_request) def items(self) -> ItemsView[str, ApiItemT]: """Return items.""" diff --git a/axis/vapix/interfaces/basic_device_info.py b/axis/vapix/interfaces/basic_device_info.py index ce73ecc6..73a3bd93 100644 --- a/axis/vapix/interfaces/basic_device_info.py +++ b/axis/vapix/interfaces/basic_device_info.py @@ -7,10 +7,13 @@ from ..models.api_discovery import ApiId from ..models.basic_device_info import ( + API_VERSION, DeviceInformation, GetAllPropertiesRequest, + GetAllPropertiesResponse, GetAllPropertiesT, GetSupportedVersionsRequest, + GetSupportedVersionsResponse, ) from .api_handler import ApiHandler @@ -19,18 +22,26 @@ class BasicDeviceInfoHandler(ApiHandler[DeviceInformation]): """Basic device information for Axis devices.""" api_id = ApiId.BASIC_DEVICE_INFO - api_request = GetAllPropertiesRequest() + default_api_version = API_VERSION + + async def _api_request(self) -> dict[str, DeviceInformation]: + """Get default data of basic device information.""" + return await self.get_all_properties() async def get_all_properties(self) -> GetAllPropertiesT: """List all properties of basic device info.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( GetAllPropertiesRequest(discovery_item.version) ) + response = GetAllPropertiesResponse.decode(bytes_data) + return {"0": response.data} async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data @property def architecture(self) -> str: diff --git a/axis/vapix/interfaces/light_control.py b/axis/vapix/interfaces/light_control.py index d18b1164..45d92784 100644 --- a/axis/vapix/interfaces/light_control.py +++ b/axis/vapix/interfaces/light_control.py @@ -13,17 +13,29 @@ DisableLightRequest, EnableLightRequest, GetCurrentAngleOfIlluminationRequest, + GetCurrentAngleOfIlluminationResponse, GetCurrentIntensityRequest, + GetCurrentIntensityResponse, GetIndividualIntensityRequest, - GetLightInformation, + GetIndividualIntensityResponse, + GetLightInformationRequest, + GetLightInformationResponse, GetLightStatusRequest, + GetLightStatusResponse, GetLightSynchronizeDayNightModeRequest, + GetLightSynchronizeDayNightModeResponse, GetManualAngleOfIlluminationRequest, + GetManualAngleOfIlluminationResponse, GetManualIntensityRequest, - GetServiceCapabilities, + GetManualIntensityResponse, + GetServiceCapabilitiesRequest, + GetServiceCapabilitiesResponse, GetSupportedVersionsRequest, - GetValidAngleOfIllumination, + GetSupportedVersionsResponse, + GetValidAngleOfIlluminationRequest, + GetValidAngleOfIlluminationResponse, GetValidIntensityRequest, + GetValidIntensityResponse, LightInformation, Range, ServiceCapabilities, @@ -41,78 +53,90 @@ class LightHandler(ApiHandler[LightInformation]): """Light control for Axis devices.""" api_id = ApiId.LIGHT_CONTROL - api_request = GetLightInformation() default_api_version = API_VERSION + @property + def api_version(self) -> str: + """Temporary override to complete PR. + + REMOVE ME. + """ + if not self.supported(): + return API_VERSION + discovery_item = self.vapix.api_discovery[self.api_id.value] + return discovery_item.version + + async def _api_request(self) -> dict[str, LightInformation]: + """Get default data of stream profiles.""" + return await self.get_light_information() + async def get_light_information(self) -> dict[str, LightInformation]: """List the light control information.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2(GetLightInformation(api_version)) + bytes_data = await self.vapix.new_request( + GetLightInformationRequest(self.api_version) + ) + return GetLightInformationResponse.decode(bytes_data).data async def get_service_capabilities(self) -> ServiceCapabilities: """List the light control information.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2(GetServiceCapabilities(api_version)) + bytes_data = await self.vapix.new_request( + GetServiceCapabilitiesRequest(self.api_version) + ) + return GetServiceCapabilitiesResponse.decode(bytes_data).data async def activate_light(self, light_id: str) -> None: """Activate the light.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - ActivateLightRequest(api_version, light_id=light_id) + await self.vapix.new_request( + ActivateLightRequest(self.api_version, light_id=light_id) ) async def deactivate_light(self, light_id: str) -> None: """Deactivate the light.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - DeactivateLightRequest(api_version, light_id=light_id) + await self.vapix.new_request( + DeactivateLightRequest(self.api_version, light_id=light_id) ) async def enable_light(self, light_id: str) -> None: """Activate the light.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - EnableLightRequest(api_version, light_id=light_id) + await self.vapix.new_request( + EnableLightRequest(self.api_version, light_id=light_id) ) async def disable_light(self, light_id: str) -> None: """Deactivate the light.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - DisableLightRequest(api_version, light_id=light_id) + await self.vapix.new_request( + DisableLightRequest(self.api_version, light_id=light_id) ) async def get_light_status(self, light_id: str) -> bool: """Get light status if its on or off.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetLightStatusRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetLightStatusRequest(self.api_version, light_id=light_id) ) + return GetLightStatusResponse.decode(bytes_data).data async def set_automatic_intensity_mode(self, light_id: str, enabled: bool) -> None: """Enable the automatic light intensity control.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetAutomaticIntensityModeRequest( - api_version, + self.api_version, light_id=light_id, enabled=enabled, ) ) async def get_valid_intensity(self, light_id: str) -> Range: - """Enable the automatic light intensity control.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetValidIntensityRequest(api_version, light_id=light_id) + """Get valid intensity range for light.""" + bytes_data = await self.vapix.new_request( + GetValidIntensityRequest(self.api_version, light_id=light_id) ) + return GetValidIntensityResponse.decode(bytes_data).data async def set_manual_intensity(self, light_id: str, intensity: int) -> None: """Manually sets the intensity.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetManualIntensityRequest( - api_version, + self.api_version, light_id=light_id, intensity=intensity, ) @@ -120,19 +144,18 @@ async def set_manual_intensity(self, light_id: str, intensity: int) -> None: async def get_manual_intensity(self, light_id: str) -> int: """Enable the automatic light intensity control.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetManualIntensityRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetManualIntensityRequest(self.api_version, light_id=light_id) ) + return GetManualIntensityResponse.decode(bytes_data).data async def set_individual_intensity( self, light_id: str, led_id: int, intensity: int ) -> None: """Manually sets the intensity for an individual LED.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetIndividualIntensityRequest( - api_version, + self.api_version, light_id=light_id, led_id=led_id, intensity=intensity, @@ -141,21 +164,21 @@ async def set_individual_intensity( async def get_individual_intensity(self, light_id: str, led_id: int) -> int: """Receives the intensity from the setIndividualIntensity request.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( GetIndividualIntensityRequest( - api_version, + self.api_version, light_id=light_id, led_id=led_id, ) ) + return GetIndividualIntensityResponse.decode(bytes_data).data async def get_current_intensity(self, light_id: str) -> int: """Receives the intensity from the setIndividualIntensity request.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetCurrentIntensityRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetCurrentIntensityRequest(self.api_version, light_id=light_id) ) + return GetCurrentIntensityResponse.decode(bytes_data).data async def set_automatic_angle_of_illumination_mode( self, light_id: str, enabled: bool @@ -165,19 +188,18 @@ async def set_automatic_angle_of_illumination_mode( Using this mode means that the angle of illumination is the same as the camera’s angle of view. """ - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetAutomaticAngleOfIlluminationModeRequest( - api_version, light_id=light_id, enabled=enabled + self.api_version, light_id=light_id, enabled=enabled ) ) async def get_valid_angle_of_illumination(self, light_id: str) -> list[Range]: """List the valid angle of illumination values.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetValidAngleOfIllumination(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetValidAngleOfIlluminationRequest(self.api_version, light_id=light_id) ) + return GetValidAngleOfIlluminationResponse.decode(bytes_data).data async def set_manual_angle_of_illumination( self, light_id: str, angle_of_illumination: int @@ -187,10 +209,9 @@ async def set_manual_angle_of_illumination( This is useful when the angle of illumination needs to be different from the camera’s view angle. """ - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetManualAngleOfIlluminationModeRequest( - api_version, + self.api_version, light_id=light_id, angle_of_illumination=angle_of_illumination, ) @@ -198,36 +219,36 @@ async def set_manual_angle_of_illumination( async def get_manual_angle_of_illumination(self, light_id: str) -> int: """Get the angle of illumination.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetManualAngleOfIlluminationRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetManualAngleOfIlluminationRequest(self.api_version, light_id=light_id) ) + return GetManualAngleOfIlluminationResponse.decode(bytes_data).data async def get_current_angle_of_illumination(self, light_id: str) -> int: """Receive the current angle of illumination.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetCurrentAngleOfIlluminationRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetCurrentAngleOfIlluminationRequest(self.api_version, light_id=light_id) ) + return GetCurrentAngleOfIlluminationResponse.decode(bytes_data).data async def set_light_synchronization_day_night_mode( self, light_id: str, enabled: bool ) -> None: """Enable automatic synchronization with the day/night mode.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( + await self.vapix.new_request( SetLightSynchronizeDayNightModeRequest( - api_version, light_id=light_id, enabled=enabled + self.api_version, light_id=light_id, enabled=enabled ) ) async def get_light_synchronization_day_night_mode(self, light_id: str) -> bool: """Check if the automatic synchronization is enabled with the day/night mode.""" - api_version = self.api_version() or self.default_api_version - return await self.vapix.request2( - GetLightSynchronizeDayNightModeRequest(api_version, light_id=light_id) + bytes_data = await self.vapix.new_request( + GetLightSynchronizeDayNightModeRequest(self.api_version, light_id=light_id) ) + return GetLightSynchronizeDayNightModeResponse.decode(bytes_data).data async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + return GetSupportedVersionsResponse.decode(bytes_data).data diff --git a/axis/vapix/interfaces/mqtt.py b/axis/vapix/interfaces/mqtt.py index f2b03694..84b014a6 100644 --- a/axis/vapix/interfaces/mqtt.py +++ b/axis/vapix/interfaces/mqtt.py @@ -15,7 +15,9 @@ EventFilter, EventPublicationConfig, GetClientStatusRequest, + GetClientStatusResponse, GetEventPublicationConfigRequest, + GetEventPublicationConfigResponse, ) from .api_handler import ApiHandler @@ -51,42 +53,50 @@ def mqtt_json_to_event(msg: str) -> dict[str, Any]: } -class MqttClientHandler(ApiHandler): +class MqttClientHandler(ApiHandler[Any]): """MQTT Client for Axis devices.""" api_id = ApiId.MQTT_CLIENT - api_request = None + default_api_version = API_VERSION + + async def _api_request(self) -> dict[str, None]: + """Get API data method defined by subsclass.""" + raise NotImplementedError async def configure_client(self, client_config: ClientConfig) -> None: """Configure MQTT Client.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + await self.vapix.new_request( ConfigureClientRequest(discovery_item.version, client_config=client_config) ) async def activate(self) -> None: """Activate MQTT Client.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2(ActivateClientRequest(discovery_item.version)) + await self.vapix.new_request(ActivateClientRequest(discovery_item.version)) async def deactivate(self) -> None: """Deactivate MQTT Client.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( - DeactivateClientRequest(discovery_item.version) - ) + await self.vapix.new_request(DeactivateClientRequest(discovery_item.version)) async def get_client_status(self) -> ClientConfigStatus: """Get MQTT Client status.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2(GetClientStatusRequest(discovery_item.version)) + bytes_data = await self.vapix.new_request( + GetClientStatusRequest(discovery_item.version) + ) + response = GetClientStatusResponse.decode(bytes_data) + return response.data async def get_event_publication_config(self) -> EventPublicationConfig: """Get MQTT Client event publication config.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( GetEventPublicationConfigRequest(discovery_item.version) ) + response = GetEventPublicationConfigResponse.decode(bytes_data) + return response.data async def configure_event_publication( self, topics: list[str] = DEFAULT_TOPICS @@ -97,6 +107,6 @@ async def configure_event_publication( [{"topicFilter": topic} for topic in topics] ) config = EventPublicationConfig(event_filter_list=event_filters) - return await self.vapix.request2( + await self.vapix.new_request( ConfigureEventPublicationRequest(discovery_item.version, config=config) ) diff --git a/axis/vapix/interfaces/pir_sensor_configuration.py b/axis/vapix/interfaces/pir_sensor_configuration.py index 1637c595..fa2b6293 100644 --- a/axis/vapix/interfaces/pir_sensor_configuration.py +++ b/axis/vapix/interfaces/pir_sensor_configuration.py @@ -6,9 +6,13 @@ from ..models.api_discovery import ApiId from ..models.pir_sensor_configuration import ( + API_VERSION, GetSensitivityRequest, + GetSensitivityResponse, GetSupportedVersionsRequest, + GetSupportedVersionsResponse, ListSensorsRequest, + ListSensorsResponse, ListSensorsT, PirSensorConfiguration, SetSensitivityRequest, @@ -20,27 +24,39 @@ class PirSensorConfigurationHandler(ApiHandler[PirSensorConfiguration]): """PIR sensor configuration for Axis devices.""" api_id = ApiId.PIR_SENSOR_CONFIGURATION - api_request = ListSensorsRequest() + default_api_version = API_VERSION + + async def _api_request(self) -> ListSensorsT: + """Get default data of PIR sensor configuration.""" + return await self.list_sensors() async def list_sensors(self) -> ListSensorsT: """List all PIR sensors of device.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2(ListSensorsRequest(discovery_item.version)) + bytes_data = await self.vapix.new_request( + ListSensorsRequest(discovery_item.version) + ) + response = ListSensorsResponse.decode(bytes_data) + return response.data async def get_sensitivity(self, id: int) -> float | None: """Retrieve configured sensitivity of specific sensor.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( GetSensitivityRequest(id, discovery_item.version) ) + response = GetSensitivityResponse.decode(bytes_data) + return response.data async def set_sensitivity(self, id: int, sensitivity: float) -> None: """Configure sensitivity of specific sensor.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + await self.vapix.new_request( SetSensitivityRequest(id, sensitivity, discovery_item.version) ) async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data diff --git a/axis/vapix/interfaces/stream_profiles.py b/axis/vapix/interfaces/stream_profiles.py index 401c5639..c055531c 100644 --- a/axis/vapix/interfaces/stream_profiles.py +++ b/axis/vapix/interfaces/stream_profiles.py @@ -10,8 +10,11 @@ from ..models.api_discovery import ApiId from ..models.stream_profile import ( + API_VERSION, GetSupportedVersionsRequest, + GetSupportedVersionsResponse, ListStreamProfilesRequest, + ListStreamProfilesResponse, ListStreamProfilesT, StreamProfile, ) @@ -22,15 +25,23 @@ class StreamProfilesHandler(ApiHandler[StreamProfile]): """API Discovery for Axis devices.""" api_id = ApiId.STREAM_PROFILES - api_request = ListStreamProfilesRequest() + default_api_version = API_VERSION + + async def _api_request(self) -> ListStreamProfilesT: + """Get default data of stream profiles.""" + return await self.list_stream_profiles() async def list_stream_profiles(self) -> ListStreamProfilesT: - """List all APIs registered on API Discovery service.""" + """List all stream profiles.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( ListStreamProfilesRequest(discovery_item.version) ) + response = ListStreamProfilesResponse.decode(bytes_data) + return response.data async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data diff --git a/axis/vapix/interfaces/view_areas.py b/axis/vapix/interfaces/view_areas.py index 996ba482..593132de 100644 --- a/axis/vapix/interfaces/view_areas.py +++ b/axis/vapix/interfaces/view_areas.py @@ -13,7 +13,9 @@ Geometry, GetSupportedConfigVersionsRequest, GetSupportedVersionsRequest, + GetSupportedVersionsResponse, ListViewAreasRequest, + ListViewAreasResponse, ListViewAreasT, ResetGeometryRequest, SetGeometryRequest, @@ -26,12 +28,20 @@ class ViewAreaHandler(ApiHandler[ViewArea]): """View areas for Axis devices.""" api_id = ApiId.VIEW_AREA - api_request = ListViewAreasRequest() + # api_request = ListViewAreasRequest() + + async def _api_request(self) -> ListViewAreasT: + """Get default data of stream profiles.""" + return await self.list_view_areas() async def list_view_areas(self) -> ListViewAreasT: """List all view areas of device.""" discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2(ListViewAreasRequest(discovery_item.version)) + bytes_data = await self.vapix.new_request( + ListViewAreasRequest(discovery_item.version) + ) + response = ListViewAreasResponse.decode(bytes_data) + return response.data async def set_geometry(self, id: int, geometry: Geometry) -> ListViewAreasT: """Set geometry of a view area. @@ -40,13 +50,15 @@ async def set_geometry(self, id: int, geometry: Geometry) -> ListViewAreasT: Method: POST """ discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( SetGeometryRequest( id=id, geometry=geometry, api_version=discovery_item.version, ) ) + response = ListViewAreasResponse.decode(bytes_data) + return response.data async def reset_geometry(self, id: int) -> ListViewAreasT: """Restore geometry of a view area back to default values. @@ -55,14 +67,20 @@ async def reset_geometry(self, id: int) -> ListViewAreasT: Method: POST """ discovery_item = self.vapix.api_discovery[self.api_id.value] - return await self.vapix.request2( + bytes_data = await self.vapix.new_request( ResetGeometryRequest(id=id, api_version=discovery_item.version) ) + response = ListViewAreasResponse.decode(bytes_data) + return response.data async def get_supported_versions(self) -> list[str]: """List supported API versions.""" - return await self.vapix.request2(GetSupportedVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data async def get_supported_config_versions(self) -> list[str]: """List supported configure API versions.""" - return await self.vapix.request2(GetSupportedConfigVersionsRequest()) + bytes_data = await self.vapix.new_request(GetSupportedConfigVersionsRequest()) + response = GetSupportedVersionsResponse.decode(bytes_data) + return response.data diff --git a/axis/vapix/models/api.py b/axis/vapix/models/api.py index f88107ec..22fefb37 100644 --- a/axis/vapix/models/api.py +++ b/axis/vapix/models/api.py @@ -2,7 +2,9 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field -from typing import Any, Callable, Generic, List, TypeVar +from typing import Callable, Generic, List, TypeVar + +from typing_extensions import Self CONTEXT = "Axis library" @@ -19,19 +21,40 @@ class ApiItem(ABC): @dataclass -class ApiRequest(ABC, Generic[ApiDataT]): +class ApiResponseSupportDecode(ABC): + """Response from API request.""" + + @classmethod + @abstractmethod + def decode(cls, bytes_data: bytes) -> Self: + """Decode data to class object.""" + + +@dataclass +class ApiResponse(ApiResponseSupportDecode, Generic[ApiDataT]): + """Response from API request. + + Class with generic can't be used in a TypeVar("X", bound=class) statement. + """ + + data: ApiDataT + # error: str + + +ApiResponseT = TypeVar("ApiResponseT", bound=ApiResponseSupportDecode) + + +@dataclass +class ApiRequest(ABC): """Create API request body.""" method: str = field(init=False) path: str = field(init=False) - data: dict[str, Any] = field(init=False) - - content_type: str = field(init=False) - error_codes: dict[int, str] = field(init=False) + @property @abstractmethod - def process_raw(self, raw: bytes) -> ApiDataT: - """Process raw data.""" + def content(self) -> bytes: + """Request content.""" class APIItem: diff --git a/axis/vapix/models/api_discovery.py b/axis/vapix/models/api_discovery.py index 8b6b71de..31046036 100644 --- a/axis/vapix/models/api_discovery.py +++ b/axis/vapix/models/api_discovery.py @@ -5,9 +5,9 @@ import logging import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.0" @@ -113,7 +113,7 @@ class ApiDescriptionT(TypedDict): docLink: str id: str name: str - status: str + status: NotRequired[str] version: str @@ -173,12 +173,49 @@ def api_id(self) -> ApiId: """ID of API.""" return ApiId(self.id) + @classmethod + def decode(cls, raw: ApiDescriptionT) -> Self: + """Decode dict to class object.""" + return cls( + id=raw["id"], + name=raw["name"], + status=ApiStatus(raw.get("status", "")), + version=raw["version"], + ) + + @classmethod + def decode_from_list(cls, raw: list[ApiDescriptionT]) -> list[Self]: + """Decode list[dict] to list of class objects.""" + return [cls.decode(item) for item in raw] + ListApisT = dict[str, Api] @dataclass -class ListApisRequest(ApiRequest[ListApisT]): +class GetAllApisResponse(ApiResponse[list[Api]]): + """Response object for basic device info.""" + + api_version: str + context: str + method: str + data: list[Api] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: ListApisResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=Api.decode_from_list(data["data"]["apiList"]), + ) + + +@dataclass +class ListApisRequest(ApiRequest): """Request object for listing API descriptions.""" method = "post" @@ -189,31 +226,42 @@ class ListApisRequest(ApiRequest[ListApisT]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getApiList", - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getApiList", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" - def process_raw(self, raw: bytes) -> ListApisT: + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: """Prepare API description dictionary.""" - data: ListApisResponseT = orjson.loads(raw) - apis = data.get("data", {}).get("apiList", []) - return { - api["id"]: Api( - id=api["id"], - name=api["name"], - status=ApiStatus(api.get("status", "")), - version=api["version"], - ) - for api in apis - } + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -223,14 +271,12 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) diff --git a/axis/vapix/models/basic_device_info.py b/axis/vapix/models/basic_device_info.py index f773f2d4..09ff66c8 100644 --- a/axis/vapix/models/basic_device_info.py +++ b/axis/vapix/models/basic_device_info.py @@ -3,9 +3,9 @@ from dataclasses import dataclass import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.1" @@ -99,13 +99,56 @@ class DeviceInformation(ApiItem): version: str web_url: str + @classmethod + def decode(cls, raw: DeviceInformationDescriptionT) -> Self: + """Decode dict to class object.""" + return cls( + id="0", + architecture=raw["Architecture"], + brand=raw["Brand"], + build_date=raw["BuildDate"], + hardware_id=raw["HardwareID"], + product_full_name=raw["ProdFullName"], + product_number=raw["ProdNbr"], + product_short_name=raw["ProdShortName"], + product_type=raw["ProdType"], + product_variant=raw["ProdVariant"], + serial_number=raw["SerialNumber"], + soc=raw["Soc"], + soc_serial_number=raw["SocSerialNumber"], + version=raw["Version"], + web_url=raw["WebURL"], + ) + GetAllPropertiesT = dict[str, DeviceInformation] @dataclass -class GetAllPropertiesRequest(ApiRequest[GetAllPropertiesT]): - """Request object for listing API descriptions.""" +class GetAllPropertiesResponse(ApiResponse[DeviceInformation]): + """Response object for basic device info.""" + + api_version: str + context: str + method: str + data: DeviceInformation + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetAllPropertiesResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=DeviceInformation.decode(data["data"]["propertyList"]), + ) + + +@dataclass +class GetAllPropertiesRequest(ApiRequest): + """Request object for basic device info.""" method = "post" path = "/axis-cgi/basicdeviceinfo.cgi" @@ -115,41 +158,42 @@ class GetAllPropertiesRequest(ApiRequest[GetAllPropertiesT]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getAllProperties", - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getAllProperties", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" + + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None - def process_raw(self, raw: bytes) -> GetAllPropertiesT: + @classmethod + def decode(cls, bytes_data: bytes) -> Self: """Prepare API description dictionary.""" - data: GetAllPropertiesResponseT = orjson.loads(raw) - device_information = data.get("data", {}).get("propertyList", {}) - return { - "0": DeviceInformation( - id="0", - architecture=device_information["Architecture"], - brand=device_information["Brand"], - build_date=device_information["BuildDate"], - hardware_id=device_information["HardwareID"], - product_full_name=device_information["ProdFullName"], - product_number=device_information["ProdNbr"], - product_short_name=device_information["ProdShortName"], - product_type=device_information["ProdType"], - product_variant=device_information["ProdVariant"], - serial_number=device_information["SerialNumber"], - soc=device_information["Soc"], - soc_serial_number=device_information["SocSerialNumber"], - version=device_information["Version"], - web_url=device_information["WebURL"], - ) - } + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -159,14 +203,12 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) diff --git a/axis/vapix/models/light_control.py b/axis/vapix/models/light_control.py index c8309dec..aa030af9 100644 --- a/axis/vapix/models/light_control.py +++ b/axis/vapix/models/light_control.py @@ -1,11 +1,11 @@ """Light Control API data model.""" -from dataclasses import dataclass +from dataclasses import dataclass, field import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.1" @@ -198,9 +198,9 @@ class LightInformation(ApiItem): error_info: str @classmethod - def from_dict(cls, data: LightInformationT) -> "LightInformation": + def from_dict(cls, data: LightInformationT) -> Self: """Create light information object from dict.""" - return LightInformation( + return cls( id=data["lightID"], enabled=data["enabled"], light_state=data["lightState"], @@ -216,9 +216,9 @@ def from_dict(cls, data: LightInformationT) -> "LightInformation": ) @classmethod - def from_list(cls, data: list[LightInformationT]) -> dict[str, "LightInformation"]: + def from_list(cls, data: list[LightInformationT]) -> dict[str, Self]: """Create light information objects from list.""" - lights = [LightInformation.from_dict(item) for item in data] + lights = [cls.from_dict(item) for item in data] return {light.id: light for light in lights} @@ -241,7 +241,29 @@ def from_list(cls, data: list[RangeT]) -> list["Range"]: @dataclass -class GetLightInformation(ApiRequest[dict[str, LightInformation]]): +class ApiVersion: + """Handle API version.""" + + _default_api_version: str = field(init=False) + api_version: str | None = None + + @property + def _api_version(self) -> str: + """API version or default API version.""" + if self.api_version is not None: + return self.api_version + return self._default_api_version + + +@dataclass +class _ApiVersion(ApiVersion): + """Light control API version.""" + + _default_api_version = API_VERSION + + +@dataclass +class GetLightInformationRequest(ApiRequest): """Request object for getting light information.""" method = "post" @@ -252,18 +274,39 @@ class GetLightInformation(ApiRequest[dict[str, LightInformation]]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getLightInformation", - } + assert self.api_version is not None + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getLightInformation", + } + ) + - def process_raw(self, raw: bytes) -> dict[str, LightInformation]: - """Prepare light information dictionary.""" - data: GetLightInformationResponseT = orjson.loads(raw) - return LightInformation.from_list(data["data"]["items"]) +@dataclass +class GetLightInformationResponse(ApiResponse[dict[str, LightInformation]]): + """Response object for getting light information.""" + + api_version: str + context: str + method: str + data: dict[str, LightInformation] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetLightInformationResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=LightInformation.from_list(data["data"]["items"]), + ) @dataclass @@ -297,8 +340,8 @@ def from_dict(cls, data: ServiceCapabilitiesT) -> "ServiceCapabilities": @dataclass -class GetServiceCapabilities(ApiRequest[ServiceCapabilities]): - """Request object for getting light information.""" +class GetServiceCapabilitiesRequest(ApiRequest): + """Request object for getting service capabilities.""" method = "post" path = "/axis-cgi/lightcontrol.cgi" @@ -308,22 +351,42 @@ class GetServiceCapabilities(ApiRequest[ServiceCapabilities]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getServiceCapabilities", - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getServiceCapabilities", + } + ) - def process_raw(self, raw: bytes) -> ServiceCapabilities: - """Prepare light information dictionary.""" - data: GetServiceCapabilitiesResponseT = orjson.loads(raw) - return ServiceCapabilities.from_dict(data["data"]) + +@dataclass +class GetServiceCapabilitiesResponse(ApiResponse[ServiceCapabilities]): + """Response object for getting service capabilities.""" + + api_version: str + context: str + method: str + data: ServiceCapabilities + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetServiceCapabilitiesResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=ServiceCapabilities.from_dict(data["data"]), + ) @dataclass -class ActivateLightRequest(ApiRequest[None]): +class ActivateLightRequest(ApiRequest): """Request object for activating light.""" method = "post" @@ -335,67 +398,76 @@ class ActivateLightRequest(ApiRequest[None]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "activateLight", - "params": {"lightID": self.light_id}, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "activateLight", + "params": {"lightID": self.light_id}, + } + ) @dataclass class DeactivateLightRequest(ActivateLightRequest): """Request object for activating light.""" - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "deactivateLight", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "deactivateLight", + "params": {"lightID": self.light_id}, + } + ) @dataclass class EnableLightRequest(ActivateLightRequest): """Request object for enabling light.""" - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "enableLight", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "enableLight", + "params": {"lightID": self.light_id}, + } + ) @dataclass class DisableLightRequest(ActivateLightRequest): """Request object for disabling light.""" - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "disableLight", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "disableLight", + "params": {"lightID": self.light_id}, + } + ) @dataclass -class GetLightStatusRequest(ApiRequest[bool]): +class GetLightStatusRequest(ApiRequest): """Request object for getting light status.""" method = "post" @@ -407,24 +479,44 @@ class GetLightStatusRequest(ApiRequest[bool]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getLightStatus", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getLightStatus", + "params": {"lightID": self.light_id}, + } + ) + - def process_raw(self, raw: bytes) -> bool: - """If light is on or off.""" - data: GetLightStatusResponseT = orjson.loads(raw) - return data["data"]["status"] +@dataclass +class GetLightStatusResponse(ApiResponse[bool]): + """Response object for getting light status.""" + + api_version: str + context: str + method: str + data: bool + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetLightStatusResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["status"], + ) @dataclass -class SetAutomaticIntensityModeRequest(ApiRequest[None]): +class SetAutomaticIntensityModeRequest(ApiRequest): """Enable the automatic light intensity control.""" method = "post" @@ -437,24 +529,24 @@ class SetAutomaticIntensityModeRequest(ApiRequest[None]): light_id: str | None = None enabled: bool | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.enabled is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setAutomaticIntensityMode", - "params": {"lightID": self.light_id, "enabled": self.enabled}, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setAutomaticIntensityMode", + "params": {"lightID": self.light_id, "enabled": self.enabled}, + } + ) @dataclass -class GetValidIntensityRequest(ApiRequest[Range]): - """Request object for getting light status.""" +class GetValidIntensityRequest(ApiRequest): + """Request object for getting valid intensity range of light.""" method = "post" path = "/axis-cgi/lightcontrol.cgi" @@ -465,24 +557,44 @@ class GetValidIntensityRequest(ApiRequest[Range]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getValidIntensity", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getValidIntensity", + "params": {"lightID": self.light_id}, + } + ) - def process_raw(self, raw: bytes) -> Range: - """If light is on or off.""" - data: GetValidRangesResponseT = orjson.loads(raw) - return Range.from_dict(data["data"]["ranges"][0]) + +@dataclass +class GetValidIntensityResponse(ApiResponse[Range]): + """Response object for getting valid intensity range of light.""" + + api_version: str + context: str + method: str + data: Range + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetValidRangesResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=Range.from_dict(data["data"]["ranges"][0]), + ) @dataclass -class SetManualIntensityRequest(ApiRequest[None]): +class SetManualIntensityRequest(ApiRequest): """Set manual light intensity.""" method = "post" @@ -495,23 +607,23 @@ class SetManualIntensityRequest(ApiRequest[None]): light_id: str | None = None intensity: int | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.intensity is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setManualIntensity", - "params": {"lightID": self.light_id, "intensity": self.intensity}, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setManualIntensity", + "params": {"lightID": self.light_id, "intensity": self.intensity}, + } + ) @dataclass -class GetManualIntensityRequest(ApiRequest[int]): +class GetManualIntensityRequest(ApiRequest): """Request object for getting manual intensity.""" method = "post" @@ -523,24 +635,44 @@ class GetManualIntensityRequest(ApiRequest[int]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getManualIntensity", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getManualIntensity", + "params": {"lightID": self.light_id}, + } + ) - def process_raw(self, raw: bytes) -> int: - """If light is on or off.""" - data: GetIntensityResponseT = orjson.loads(raw) - return data["data"]["intensity"] + +@dataclass +class GetManualIntensityResponse(ApiResponse[int]): + """Response object for getting manual intensity.""" + + api_version: str + context: str + method: str + data: int + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetIntensityResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["intensity"], + ) @dataclass -class SetIndividualIntensityRequest(ApiRequest[None]): +class SetIndividualIntensityRequest(ApiRequest): """Set individual light intensity.""" method = "post" @@ -554,28 +686,28 @@ class SetIndividualIntensityRequest(ApiRequest[None]): led_id: int | None = None intensity: int | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.led_id is not None assert self.intensity is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setIndividualIntensity", - "params": { - "lightID": self.light_id, - "LEDID": self.led_id, - "intensity": self.intensity, - }, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setIndividualIntensity", + "params": { + "lightID": self.light_id, + "LEDID": self.led_id, + "intensity": self.intensity, + }, + } + ) @dataclass -class GetIndividualIntensityRequest(ApiRequest[int]): +class GetIndividualIntensityRequest(ApiRequest): """Request object for getting individual intensity.""" method = "post" @@ -588,26 +720,46 @@ class GetIndividualIntensityRequest(ApiRequest[int]): light_id: str | None = None led_id: int | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.led_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getIndividualIntensity", - "params": {"lightID": self.light_id, "LEDID": self.led_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getIndividualIntensity", + "params": {"lightID": self.light_id, "LEDID": self.led_id}, + } + ) - def process_raw(self, raw: bytes) -> int: - """Process light intensity.""" - data: GetIntensityResponseT = orjson.loads(raw) - return data["data"]["intensity"] + +@dataclass +class GetIndividualIntensityResponse(ApiResponse[int]): + """Response object for getting individual intensity.""" + + api_version: str + context: str + method: str + data: int + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetIntensityResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["intensity"], + ) @dataclass -class GetCurrentIntensityRequest(ApiRequest[int]): - """Request object for getting manual intensity.""" +class GetCurrentIntensityRequest(ApiRequest): + """Request object for getting current intensity.""" method = "post" path = "/axis-cgi/lightcontrol.cgi" @@ -618,24 +770,44 @@ class GetCurrentIntensityRequest(ApiRequest[int]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getCurrentIntensity", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getCurrentIntensity", + "params": {"lightID": self.light_id}, + } + ) + - def process_raw(self, raw: bytes) -> int: - """If light is on or off.""" - data: GetIntensityResponseT = orjson.loads(raw) - return data["data"]["intensity"] +@dataclass +class GetCurrentIntensityResponse(ApiResponse[int]): + """Response object for getting current intensity.""" + + api_version: str + context: str + method: str + data: int + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetIntensityResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["intensity"], + ) @dataclass -class SetAutomaticAngleOfIlluminationModeRequest(ApiRequest[None]): +class SetAutomaticAngleOfIlluminationModeRequest(ApiRequest): """Enable the automatic angle of illumination control.""" method = "post" @@ -648,24 +820,24 @@ class SetAutomaticAngleOfIlluminationModeRequest(ApiRequest[None]): light_id: str | None = None enabled: bool | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.enabled is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setAutomaticAngleOfIlluminationMode", - "params": {"lightID": self.light_id, "enabled": self.enabled}, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setAutomaticAngleOfIlluminationMode", + "params": {"lightID": self.light_id, "enabled": self.enabled}, + } + ) @dataclass -class GetValidAngleOfIllumination(ApiRequest[list[Range]]): - """Request object for getting angle of illumination range.""" +class GetValidAngleOfIlluminationRequest(ApiRequest): + """Request object for getting valid angle of illumination range.""" method = "post" path = "/axis-cgi/lightcontrol.cgi" @@ -676,24 +848,44 @@ class GetValidAngleOfIllumination(ApiRequest[list[Range]]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getValidAngleOfIllumination", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getValidAngleOfIllumination", + "params": {"lightID": self.light_id}, + } + ) + - def process_raw(self, raw: bytes) -> list[Range]: - """If light is on or off.""" - data: GetValidRangesResponseT = orjson.loads(raw) - return Range.from_list(data["data"]["ranges"]) +@dataclass +class GetValidAngleOfIlluminationResponse(ApiResponse[list[Range]]): + """Response object for getting valid angle of illumination range.""" + + api_version: str + context: str + method: str + data: list[Range] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetValidRangesResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=Range.from_list(data["data"]["ranges"]), + ) @dataclass -class SetManualAngleOfIlluminationModeRequest(ApiRequest[None]): +class SetManualAngleOfIlluminationModeRequest(ApiRequest): """Set the manual angle of illumination.""" method = "post" @@ -706,26 +898,26 @@ class SetManualAngleOfIlluminationModeRequest(ApiRequest[None]): light_id: str | None = None angle_of_illumination: int | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.angle_of_illumination is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setManualAngleOfIllumination", - "params": { - "lightID": self.light_id, - "angleOfIllumination": self.angle_of_illumination, - }, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setManualAngleOfIllumination", + "params": { + "lightID": self.light_id, + "angleOfIllumination": self.angle_of_illumination, + }, + } + ) @dataclass -class GetManualAngleOfIlluminationRequest(ApiRequest[int]): +class GetManualAngleOfIlluminationRequest(ApiRequest): """Request object for getting manual angle of illumination.""" method = "post" @@ -737,24 +929,44 @@ class GetManualAngleOfIlluminationRequest(ApiRequest[int]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getManualAngleOfIllumination", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getManualAngleOfIllumination", + "params": {"lightID": self.light_id}, + } + ) + - def process_raw(self, raw: bytes) -> int: - """Angle of illumination.""" - data: GetAngleOfIlluminationResponseT = orjson.loads(raw) - return data["data"]["angleOfIllumination"] +@dataclass +class GetManualAngleOfIlluminationResponse(ApiResponse[int]): + """Response object for getting manual angle of illumination.""" + + api_version: str + context: str + method: str + data: int + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetAngleOfIlluminationResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["angleOfIllumination"], + ) @dataclass -class GetCurrentAngleOfIlluminationRequest(ApiRequest[int]): +class GetCurrentAngleOfIlluminationRequest(ApiRequest): """Request object for getting current angle of illumination.""" method = "post" @@ -766,24 +978,44 @@ class GetCurrentAngleOfIlluminationRequest(ApiRequest[int]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getCurrentAngleOfIllumination", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getCurrentAngleOfIllumination", + "params": {"lightID": self.light_id}, + } + ) - def process_raw(self, raw: bytes) -> int: - """Angle of illumination.""" - data: GetAngleOfIlluminationResponseT = orjson.loads(raw) - return data["data"]["angleOfIllumination"] + +@dataclass +class GetCurrentAngleOfIlluminationResponse(ApiResponse[int]): + """Response object for getting current angle of illumination.""" + + api_version: str + context: str + method: str + data: int + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetAngleOfIlluminationResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["angleOfIllumination"], + ) @dataclass -class SetLightSynchronizeDayNightModeRequest(ApiRequest[None]): +class SetLightSynchronizeDayNightModeRequest(ApiRequest): """Enable automatic synchronization with the day/night mode.""" method = "post" @@ -796,24 +1028,24 @@ class SetLightSynchronizeDayNightModeRequest(ApiRequest[None]): light_id: str | None = None enabled: bool | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None assert self.enabled is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setLightSynchronizationDayNightMode", - "params": {"lightID": self.light_id, "enabled": self.enabled}, - } - - def process_raw(self, raw: bytes) -> None: - """No return data to process.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setLightSynchronizationDayNightMode", + "params": {"lightID": self.light_id, "enabled": self.enabled}, + } + ) @dataclass -class GetLightSynchronizeDayNightModeRequest(ApiRequest[bool]): - """Request object for getting current angle of illumination.""" +class GetLightSynchronizeDayNightModeRequest(ApiRequest): + """Request object for getting day night mode synchronization setting.""" method = "post" path = "/axis-cgi/lightcontrol.cgi" @@ -824,24 +1056,44 @@ class GetLightSynchronizeDayNightModeRequest(ApiRequest[bool]): context: str = CONTEXT light_id: str | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.light_id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getLightSynchronizationDayNightMode", - "params": {"lightID": self.light_id}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getLightSynchronizationDayNightMode", + "params": {"lightID": self.light_id}, + } + ) + - def process_raw(self, raw: bytes) -> bool: - """If light is on or off.""" - data: GetLightSynchronizationDayNightModeResponseT = orjson.loads(raw) - return data["data"]["synchronize"] +@dataclass +class GetLightSynchronizeDayNightModeResponse(ApiResponse[bool]): + """Response object for getting day night mode synchronization setting.""" + + api_version: str + context: str + method: str + data: bool + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetLightSynchronizationDayNightModeResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["synchronize"], + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -851,14 +1103,34 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" + + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare API description dictionary.""" + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) diff --git a/axis/vapix/models/mqtt.py b/axis/vapix/models/mqtt.py index 27d5742b..820805fd 100644 --- a/axis/vapix/models/mqtt.py +++ b/axis/vapix/models/mqtt.py @@ -3,9 +3,9 @@ from dataclasses import dataclass import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiRequest +from .api import CONTEXT, ApiRequest, ApiResponse API_VERSION = "1.0" @@ -375,7 +375,7 @@ def to_dict(self) -> EventPublicationConfigT: @dataclass -class ConfigureClientRequest(ApiRequest[None]): +class ConfigureClientRequest(ApiRequest): """Request object for configuring MQTT client.""" method = "post" @@ -387,22 +387,22 @@ class ConfigureClientRequest(ApiRequest[None]): context: str = CONTEXT client_config: ClientConfig | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.client_config is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "configureClient", - "params": self.client_config.to_dict(), - } - - def process_raw(self, raw: bytes) -> None: - """Prepare view area dictionary.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "configureClient", + "params": self.client_config.to_dict(), + } + ) @dataclass -class ActivateClientRequest(ApiRequest[None]): +class ActivateClientRequest(ApiRequest): """Request object for activating MQTT client.""" method = "post" @@ -413,33 +413,58 @@ class ActivateClientRequest(ApiRequest[None]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "activateClient", - } - - def process_raw(self, raw: bytes) -> None: - """Prepare view area dictionary.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "activateClient", + } + ) @dataclass class DeactivateClientRequest(ActivateClientRequest): """Request object for deactivating MQTT client.""" - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "deactivateClient", - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "deactivateClient", + } + ) + + +@dataclass +class GetClientStatusResponse(ApiResponse[ClientConfigStatus]): + """Response object for get client status request.""" + + api_version: str + context: str + method: str + data: ClientConfigStatus + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetClientStatusResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=ClientConfigStatus.from_dict(data["data"]), + ) @dataclass -class GetClientStatusRequest(ApiRequest[ClientConfigStatus]): +class GetClientStatusRequest(ApiRequest): """Request object for getting MQTT client status.""" method = "post" @@ -450,22 +475,44 @@ class GetClientStatusRequest(ApiRequest[ClientConfigStatus]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getClientStatus", - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getClientStatus", + } + ) - def process_raw(self, raw: bytes) -> ClientConfigStatus: - """Prepare view area dictionary.""" - data: GetClientStatusResponseT = orjson.loads(raw) - return ClientConfigStatus.from_dict(data["data"]) + +@dataclass +class GetEventPublicationConfigResponse(ApiResponse[EventPublicationConfig]): + """Response object for event publication config get request.""" + + api_version: str + context: str + method: str + data: EventPublicationConfig + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetEventPublicationConfigResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=EventPublicationConfig.from_dict( + data["data"]["eventPublicationConfig"] + ), + ) @dataclass -class GetEventPublicationConfigRequest(ApiRequest[EventPublicationConfig]): +class GetEventPublicationConfigRequest(ApiRequest): """Request object for getting MQTT event publication config.""" method = "post" @@ -476,22 +523,20 @@ class GetEventPublicationConfigRequest(ApiRequest[EventPublicationConfig]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getEventPublicationConfig", - } - - def process_raw(self, raw: bytes) -> EventPublicationConfig: - """Prepare view area dictionary.""" - data: GetEventPublicationConfigResponseT = orjson.loads(raw) - return EventPublicationConfig.from_dict(data["data"]["eventPublicationConfig"]) + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getEventPublicationConfig", + } + ) @dataclass -class ConfigureEventPublicationRequest(ApiRequest[None]): +class ConfigureEventPublicationRequest(ApiRequest): """Request object for configuring event publication over MQTT.""" method = "post" @@ -503,15 +548,15 @@ class ConfigureEventPublicationRequest(ApiRequest[None]): context: str = CONTEXT config: EventPublicationConfig | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.config is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "configureEventPublication", - "params": self.config.to_dict(), - } - - def process_raw(self, raw: bytes) -> None: - """Prepare view area dictionary.""" + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "configureEventPublication", + "params": self.config.to_dict(), + } + ) diff --git a/axis/vapix/models/pir_sensor_configuration.py b/axis/vapix/models/pir_sensor_configuration.py index e5c7d908..feab58ba 100644 --- a/axis/vapix/models/pir_sensor_configuration.py +++ b/axis/vapix/models/pir_sensor_configuration.py @@ -6,9 +6,9 @@ from dataclasses import dataclass import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.0" @@ -107,12 +107,48 @@ class PirSensorConfiguration(ApiItem): configurable: bool sensitivity: float | None = None + @classmethod + def decode(cls, raw: PirSensorConfigurationT) -> Self: + """Decode dict to class object.""" + return cls( + id=raw["id"], + configurable=raw["sensitivityConfigurable"], + sensitivity=raw.get("sensitivity"), + ) + + @classmethod + def decode_from_list(cls, raw: list[PirSensorConfigurationT]) -> dict[str, Self]: + """Decode list[dict] to list of class objects.""" + return {item.id: item for item in [cls.decode(item) for item in raw]} + ListSensorsT = dict[str, PirSensorConfiguration] @dataclass -class ListSensorsRequest(ApiRequest[ListSensorsT]): +class ListSensorsResponse(ApiResponse[ListSensorsT]): + """Response object for list sensors response.""" + + api_version: str + context: str + method: str + data: ListSensorsT + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: ListSensorsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=PirSensorConfiguration.decode_from_list(data["data"]["sensors"]), + ) + + +@dataclass +class ListSensorsRequest(ApiRequest): """Request object for listing PIR sensors.""" method = "post" @@ -123,30 +159,42 @@ class ListSensorsRequest(ApiRequest[ListSensorsT]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "listSensors", - } - - def process_raw(self, raw: bytes) -> ListSensorsT: - """Prepare Pir sensor configuration dictionary.""" - data: ListSensorsResponseT = orjson.loads(raw) - sensors = data.get("data", {}).get("sensors", []) - return { - sensor["id"]: PirSensorConfiguration( - id=sensor["id"], - configurable=sensor["sensitivityConfigurable"], - sensitivity=sensor.get("sensitivity"), - ) - for sensor in sensors - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "listSensors", + } + ) + + +@dataclass +class GetSensitivityResponse(ApiResponse[float | None]): + """Response object for get sensitivity response.""" + + api_version: str + context: str + method: str + data: float | None + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetSensitivityResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data["data"]["sensitivity"], + ) @dataclass -class GetSensitivityRequest(ApiRequest[float | None]): +class GetSensitivityRequest(ApiRequest): """Request object for getting PIR sensor sensitivity.""" method = "post" @@ -158,25 +206,23 @@ class GetSensitivityRequest(ApiRequest[float | None]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "getSensitivity", - "params": { - "id": self.id, - }, - } - - def process_raw(self, raw: bytes) -> float | None: - """Prepare sensitivity value.""" - data: GetSensitivityResponseT = orjson.loads(raw) - return data.get("data", {}).get("sensitivity") + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "getSensitivity", + "params": { + "id": self.id, + }, + } + ) @dataclass -class SetSensitivityRequest(ApiRequest[None]): +class SetSensitivityRequest(ApiRequest): """Request object for setting PIR sensor sensitivity.""" method = "post" @@ -189,25 +235,24 @@ class SetSensitivityRequest(ApiRequest[None]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setSensitivity", - "params": { - "id": self.id, - "sensitivity": self.sensitivity, - }, - } - - def process_raw(self, raw: bytes) -> None: - """No expected data in response.""" - return None + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setSensitivity", + "params": { + "id": self.id, + "sensitivity": self.sensitivity, + }, + } + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -217,14 +262,34 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" + + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) diff --git a/axis/vapix/models/stream_profile.py b/axis/vapix/models/stream_profile.py index 07fd2d30..1679b0ae 100644 --- a/axis/vapix/models/stream_profile.py +++ b/axis/vapix/models/stream_profile.py @@ -8,9 +8,9 @@ from dataclasses import dataclass, field import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.0" @@ -80,12 +80,49 @@ class StreamProfile(ApiItem): description: str parameters: str + @classmethod + def decode(cls, raw: StreamProfileT) -> Self: + """Decode dict to class object.""" + return cls( + id=raw["name"], + name=raw["name"], + description=raw["description"], + parameters=raw["parameters"], + ) + + @classmethod + def decode_from_list(cls, raw: list[StreamProfileT]) -> dict[str, Self]: + """Decode list[dict] to list of class objects.""" + return {item.id: item for item in [cls.decode(item) for item in raw]} + ListStreamProfilesT = dict[str, StreamProfile] @dataclass -class ListStreamProfilesRequest(ApiRequest[ListStreamProfilesT]): +class ListStreamProfilesResponse(ApiResponse[ListStreamProfilesT]): + """Response object for list sensors response.""" + + api_version: str + context: str + method: str + data: ListStreamProfilesT + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: ListStreamProfilesResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=StreamProfile.decode_from_list(data["data"].get("streamProfile", [])), + ) + + +@dataclass +class ListStreamProfilesRequest(ApiRequest): """Request object for listing stream profiles descriptions.""" method = "post" @@ -97,33 +134,22 @@ class ListStreamProfilesRequest(ApiRequest[ListStreamProfilesT]): context: str = CONTEXT profiles: list[str] = field(default_factory=list) - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" profile_list = [{"name": profile} for profile in self.profiles] - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "list", - "params": {"streamProfileName": profile_list}, - } - - def process_raw(self, raw: bytes) -> ListStreamProfilesT: - """Prepare API description dictionary.""" - data: ListStreamProfilesResponseT = orjson.loads(raw) - stream_profiles = data.get("data", {}).get("streamProfile", []) - return { - stream_profile["name"]: StreamProfile( - id=stream_profile["name"], - name=stream_profile["name"], - description=stream_profile["description"], - parameters=stream_profile["parameters"], - ) - for stream_profile in stream_profiles - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "list", + "params": {"streamProfileName": profile_list}, + } + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -133,14 +159,34 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" + + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) diff --git a/axis/vapix/models/view_area.py b/axis/vapix/models/view_area.py index 0358ca4e..e92d0d93 100644 --- a/axis/vapix/models/view_area.py +++ b/axis/vapix/models/view_area.py @@ -3,9 +3,9 @@ from dataclasses import dataclass import orjson -from typing_extensions import NotRequired, TypedDict +from typing_extensions import NotRequired, Self, TypedDict -from .api import CONTEXT, ApiItem, ApiRequest +from .api import CONTEXT, ApiItem, ApiRequest, ApiResponse API_VERSION = "1.0" @@ -138,12 +138,67 @@ class ViewArea(ApiItem): max_size: Size | None = None grid: Geometry | None = None + @classmethod + def decode(cls, raw: ViewAreaT) -> Self: + """Decode dict to class object.""" + + def create_geometry(item: GeometryT | None) -> Geometry | None: + """Create geometry object.""" + if item is None: + return None + return Geometry.from_dict(item) + + def create_size(item: SizeT | None) -> Size | None: + """Create size object.""" + if item is None: + return None + return Size.from_dict(item) + + return cls( + id=str(raw["id"]), + camera=raw["camera"], + source=raw["source"], + configurable=raw["configurable"], + canvas_size=create_size(raw.get("canvasSize")), + rectangular_geometry=create_geometry(raw.get("rectangularGeometry")), + min_size=create_size(raw.get("minSize")), + max_size=create_size(raw.get("maxSize")), + grid=create_geometry(raw.get("grid")), + ) + + @classmethod + def decode_from_list(cls, raw: list[ViewAreaT]) -> dict[str, Self]: + """Decode list[dict] to list of class objects.""" + return {str(item.id): item for item in [cls.decode(item) for item in raw]} + ListViewAreasT = dict[str, ViewArea] @dataclass -class ListViewAreasRequest(ApiRequest[ListViewAreasT]): +class ListViewAreasResponse(ApiResponse[ListViewAreasT]): + """Response object for list view areas response.""" + + api_version: str + context: str + method: str + data: ListViewAreasT + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: ListViewAreasResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=ViewArea.decode_from_list(data.get("data", {}).get("viewAreas", [])), + ) + + +@dataclass +class ListViewAreasRequest(ApiRequest): """Request object for listing view areas.""" method = "post" @@ -154,45 +209,16 @@ class ListViewAreasRequest(ApiRequest[ListViewAreasT]): api_version: str = API_VERSION context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "list", - } - - def process_raw(self, raw: bytes) -> ListViewAreasT: - """Prepare view area dictionary.""" - data: ListViewAreasResponseT = orjson.loads(raw) - items = data.get("data", {}).get("viewAreas", []) - - def create_geometry(item: GeometryT | None) -> Geometry | None: - """Create geometry object.""" - if item is None: - return None - return Geometry.from_dict(item) - - def create_size(item: SizeT | None) -> Size | None: - """Create size object.""" - if item is None: - return None - return Size.from_dict(item) - - return { - str(item["id"]): ViewArea( - id=str(item["id"]), - camera=item["camera"], - source=item["source"], - configurable=item["configurable"], - canvas_size=create_size(item.get("canvasSize")), - rectangular_geometry=create_geometry(item.get("rectangularGeometry")), - min_size=create_size(item.get("minSize")), - max_size=create_size(item.get("maxSize")), - grid=create_geometry(item.get("grid")), - ) - for item in items - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "list", + } + ) @dataclass @@ -205,25 +231,28 @@ class SetGeometryRequest(ListViewAreasRequest): id: int | None = None geometry: Geometry | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.id is not None and self.geometry is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "setGeometry", - "params": { - "viewArea": { - "id": self.id, - "rectangularGeometry": { - "horizontalOffset": self.geometry.horizontal_offset, - "horizontalSize": self.geometry.horizontal_size, - "verticalOffset": self.geometry.vertical_offset, - "verticalSize": self.geometry.vertical_size, - }, - } - }, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "setGeometry", + "params": { + "viewArea": { + "id": self.id, + "rectangularGeometry": { + "horizontalOffset": self.geometry.horizontal_offset, + "horizontalSize": self.geometry.horizontal_size, + "verticalOffset": self.geometry.vertical_offset, + "verticalSize": self.geometry.vertical_size, + }, + } + }, + } + ) @dataclass @@ -235,19 +264,22 @@ class ResetGeometryRequest(ListViewAreasRequest): id: int | None = None - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" assert self.id is not None - self.data = { - "apiVersion": self.api_version, - "context": self.context, - "method": "resetGeometry", - "params": {"viewArea": {"id": self.id}}, - } + return orjson.dumps( + { + "apiVersion": self.api_version, + "context": self.context, + "method": "resetGeometry", + "params": {"viewArea": {"id": self.id}}, + } + ) @dataclass -class GetSupportedVersionsRequest(ApiRequest[list[str]]): +class GetSupportedVersionsRequest(ApiRequest): """Request object for listing supported API versions.""" method = "post" @@ -257,17 +289,37 @@ class GetSupportedVersionsRequest(ApiRequest[list[str]]): context: str = CONTEXT - def __post_init__(self) -> None: + @property + def content(self) -> bytes: """Initialize request data.""" - self.data = { - "context": self.context, - "method": "getSupportedVersions", - } - - def process_raw(self, raw: bytes) -> list[str]: - """Process supported versions.""" - data: GetSupportedVersionsResponseT = orjson.loads(raw) - return data.get("data", {}).get("apiVersions", []) + return orjson.dumps( + { + "context": self.context, + "method": "getSupportedVersions", + } + ) + + +@dataclass +class GetSupportedVersionsResponse(ApiResponse[list[str]]): + """Response object for supported versions.""" + + api_version: str + context: str + method: str + data: list[str] + # error: ErrorDataT | None = None + + @classmethod + def decode(cls, bytes_data: bytes) -> Self: + """Prepare response data.""" + data: GetSupportedVersionsResponseT = orjson.loads(bytes_data) + return cls( + api_version=data["apiVersion"], + context=data["context"], + method=data["method"], + data=data.get("data", {}).get("apiVersions", []), + ) @dataclass diff --git a/axis/vapix/vapix.py b/axis/vapix/vapix.py index 682c5615..f14b284f 100644 --- a/axis/vapix/vapix.py +++ b/axis/vapix/vapix.py @@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Callable import httpx -import orjson from packaging import version import xmltodict @@ -42,7 +41,6 @@ if TYPE_CHECKING: from ..device import AxisDevice - from .models.api import ApiDataT LOGGER = logging.getLogger(__name__) @@ -156,6 +154,8 @@ async def do_api_request(api: ApiHandler) -> None: await api.update() except Unauthorized: # Probably a viewer account pass + except NotImplementedError: + pass apis: tuple[ApiHandler, ...] = ( self.basic_device_info, @@ -331,14 +331,13 @@ async def request( return {} - async def request2(self, api_request: ApiRequest["ApiDataT"]) -> "ApiDataT": + async def new_request(self, api_request: ApiRequest) -> bytes: """Make a request to the device.""" - bytes_data = await self.do_request( - api_request.method, - api_request.path, - content=orjson.dumps(api_request.data), + return await self.do_request( + method=api_request.method, + path=api_request.path, + content=api_request.content, ) - return api_request.process_raw(bytes_data) async def do_request( self, diff --git a/tests/test_api_discovery.py b/tests/test_api_discovery.py index da5a25f8..a4ae98b8 100644 --- a/tests/test_api_discovery.py +++ b/tests/test_api_discovery.py @@ -23,7 +23,7 @@ def api_discovery(axis_device: AxisDevice) -> ApiDiscoveryHandler: @respx.mock @pytest.mark.asyncio -async def test_get_api_list(api_discovery): +async def test_get_api_list(api_discovery: ApiDiscoveryHandler): """Test get_api_list call.""" route = respx.post(f"http://{HOST}:80/axis-cgi/apidiscovery.cgi").respond( json=response_getApiList, @@ -56,7 +56,7 @@ async def test_get_api_list(api_discovery): @respx.mock @pytest.mark.asyncio -async def test_get_supported_versions(api_discovery): +async def test_get_supported_versions(api_discovery: ApiDiscoveryHandler): """Test get_supported_versions.""" route = respx.post(f"http://{HOST}:80/axis-cgi/apidiscovery.cgi").respond( json=response_getSupportedVersions, @@ -77,6 +77,7 @@ async def test_get_supported_versions(api_discovery): response_getApiList = { "method": "getApiList", "apiVersion": "1.0", + "context": "Axis library", "data": { "apiList": [ { @@ -174,6 +175,8 @@ async def test_get_supported_versions(api_discovery): } response_getSupportedVersions = { + "apiVersion": "1.0", + "context": "Axis library", "method": "getSupportedVersions", "data": {"apiVersions": ["1.0"]}, } diff --git a/tests/test_basic_device_info.py b/tests/test_basic_device_info.py index f6ad152a..999d6164 100644 --- a/tests/test_basic_device_info.py +++ b/tests/test_basic_device_info.py @@ -36,7 +36,7 @@ async def test_get_all_properties(basic_device_info: BasicDeviceInfoHandler): assert route.calls.last.request.url.path == "/axis-cgi/basicdeviceinfo.cgi" assert json.loads(route.calls.last.request.content) == { "method": "getAllProperties", - "apiVersion": "1.1", + "apiVersion": "1.0", "context": "Axis library", } @@ -64,6 +64,8 @@ async def test_get_supported_versions(basic_device_info: BasicDeviceInfoHandler) """Test get supported versions api.""" route = respx.post(f"http://{HOST}:80/axis-cgi/basicdeviceinfo.cgi").respond( json={ + "apiVersion": "1.1", + "context": "Axis library", "method": "getSupportedVersions", "data": {"apiVersions": ["1.1"]}, }, @@ -74,6 +76,7 @@ async def test_get_supported_versions(basic_device_info: BasicDeviceInfoHandler) assert route.calls.last.request.method == "POST" assert route.calls.last.request.url.path == "/axis-cgi/basicdeviceinfo.cgi" assert json.loads(route.calls.last.request.content) == { + # "apiVersion": "1.1", "context": "Axis library", "method": "getSupportedVersions", } @@ -83,6 +86,8 @@ async def test_get_supported_versions(basic_device_info: BasicDeviceInfoHandler) response_getAllProperties = { "apiVersion": "1.1", + "context": "Axis library", + "method": "getAllProperties", "data": { "propertyList": { "Architecture": "armv7hf", diff --git a/tests/test_light_control.py b/tests/test_light_control.py index 12f47e51..2c2402ca 100644 --- a/tests/test_light_control.py +++ b/tests/test_light_control.py @@ -11,15 +11,27 @@ from axis.device import AxisDevice from axis.vapix.interfaces.light_control import LightHandler +from axis.vapix.models.api_discovery import Api from .conftest import HOST @pytest.fixture -def light_control(axis_device: AxisDevice) -> LightHandler: +async def light_control(axis_device: AxisDevice) -> LightHandler: """Return the light_control mock object.""" - axis_device.vapix.api_discovery = api_discovery_mock = MagicMock() - api_discovery_mock.__getitem__().version = "1.0" + axis_device.vapix.api_discovery._items = { + api.id: api + for api in [ + Api.decode( + { + "id": "light-control", + "version": "1.0", + "name": "Light Control", + "docLink": "https://www.axis.com/partner_pages/vapix_library/#/", + } + ) + ] + } return axis_device.vapix.light_control @@ -29,6 +41,7 @@ async def test_update(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getLightInformation", "data": { "items": [ @@ -56,7 +69,7 @@ async def test_update(light_control): assert route.calls.last.request.url.path == "/axis-cgi/lightcontrol.cgi" assert json.loads(route.calls.last.request.content) == { "method": "getLightInformation", - "apiVersion": "1.1", + "apiVersion": "1.0", "context": "Axis library", } @@ -81,6 +94,7 @@ async def test_get_service_capabilities(light_control: LightHandler): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getServiceCapabilities", "data": { "automaticIntensitySupport": True, @@ -120,6 +134,7 @@ async def test_get_light_information(light_control: LightHandler): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getLightInformation", "data": { "items": [ @@ -266,6 +281,7 @@ async def test_get_light_status(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getLightStatus", "data": {"status": False}, }, @@ -292,6 +308,7 @@ async def test_set_automatic_intensity_mode(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "setAutomaticIntensityMode", "data": {}, }, @@ -316,6 +333,7 @@ async def test_get_manual_intensity(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getManualIntensity", "data": {"intensity": 1000}, }, @@ -342,6 +360,7 @@ async def test_set_manual_intensity(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "setManualIntensity", "data": {}, }, @@ -366,6 +385,7 @@ async def test_get_valid_intensity(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getValidIntensity", "data": {"ranges": [{"low": 0, "high": 1000}]}, }, @@ -417,6 +437,7 @@ async def test_get_individual_intensity(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getIndividualIntensity", "data": {"intensity": 1000}, }, @@ -443,6 +464,7 @@ async def test_get_current_intensity(light_control): route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ "apiVersion": "1.1", + "context": "Axis library", "method": "getCurrentIntensity", "data": {"intensity": 1000}, }, @@ -651,6 +673,8 @@ async def test_get_supported_versions(light_control): """Test get supported versions api.""" route = respx.post(f"http://{HOST}:80/axis-cgi/lightcontrol.cgi").respond( json={ + "apiVersion": "1.1", + "context": "Axis library", "method": "getSupportedVersions", "data": {"apiVersions": ["1.1"]}, }, @@ -671,6 +695,7 @@ async def test_get_supported_versions(light_control): response_getLightInformation = { "apiVersion": "1.1", + "context": "Axis library", "method": "getLightInformation", "data": { "items": [ diff --git a/tests/test_stream_profiles.py b/tests/test_stream_profiles.py index 0c654b37..cd2f5126 100644 --- a/tests/test_stream_profiles.py +++ b/tests/test_stream_profiles.py @@ -59,6 +59,7 @@ async def test_list_stream_profiles_no_profiles( json={ "method": "list", "apiVersion": "1.0", + "context": "", "data": { "maxProfiles": 0, }, @@ -91,6 +92,7 @@ async def test_get_supported_versions(stream_profiles: StreamProfilesHandler) -> response_list = { "method": "list", "apiVersion": "1.0", + "context": "", "data": { "streamProfile": [ { @@ -105,6 +107,8 @@ async def test_get_supported_versions(stream_profiles: StreamProfilesHandler) -> response_getSupportedVersions = { + "apiVersion": "1.0", + "context": "Axis library", "method": "getSupportedVersions", "data": {"apiVersions": ["1.0"]}, } diff --git a/tests/test_vapix.py b/tests/test_vapix.py index 3c8b8ef0..3310e16c 100644 --- a/tests/test_vapix.py +++ b/tests/test_vapix.py @@ -67,7 +67,12 @@ async def test_initialize(vapix: Vapix): json=stream_profiles_response, ) respx.post(f"http://{HOST}:80/axis-cgi/viewarea/info.cgi").respond( - json={"apiVersion": "1.0", "method": "list", "data": {"viewAreas": []}} + json={ + "apiVersion": "1.0", + "context": "", + "method": "list", + "data": {"viewAreas": []}, + } ) respx.get( @@ -132,7 +137,12 @@ async def test_initialize_api_discovery(vapix: Vapix): json=stream_profiles_response, ) respx.post(f"http://{HOST}:80/axis-cgi/viewarea/info.cgi").respond( - json={"apiVersion": "1.0", "method": "list", "data": {"viewAreas": []}} + json={ + "apiVersion": "1.0", + "context": "", + "method": "list", + "data": {"viewAreas": []}, + } ) await vapix.initialize_api_discovery() diff --git a/tests/test_view_areas.py b/tests/test_view_areas.py index c1e543b0..24c44af9 100644 --- a/tests/test_view_areas.py +++ b/tests/test_view_areas.py @@ -127,6 +127,7 @@ async def test_get_supported_versions(view_areas: ViewAreaHandler): route = respx.post(f"http://{HOST}:80{URL_INFO}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "data": {"apiVersions": ["1.0"]}, }, @@ -291,6 +292,7 @@ async def test_get_supported_config_versions(view_areas: ViewAreaHandler): route = respx.post(f"http://{HOST}:80{URL_CONFIG}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "data": {"apiVersions": ["1.0"]}, }, @@ -319,6 +321,7 @@ async def test_general_error_101(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_INFO}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 101, @@ -342,6 +345,7 @@ async def test_general_error_102(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_INFO}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 102, @@ -365,6 +369,7 @@ async def test_general_error_103(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_INFO}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 103, @@ -388,6 +393,7 @@ async def test_method_specific_error_200(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_CONFIG}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 200, @@ -409,6 +415,7 @@ async def test_method_specific_error_201(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_CONFIG}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 201, @@ -430,6 +437,7 @@ async def test_method_specific_error_202(view_areas: ViewAreaHandler): respx.post(f"http://{HOST}:80{URL_CONFIG}").respond( json={ "apiVersion": "1.0", + "context": "", "method": "getSupportedVersions", "error": { "code": 202,