diff --git a/docs/source/platypush/responses/camera.android.rst b/docs/source/platypush/responses/camera.android.rst deleted file mode 100644 index f4317c90d..000000000 --- a/docs/source/platypush/responses/camera.android.rst +++ /dev/null @@ -1,5 +0,0 @@ -``camera.android`` -============================================= - -.. automodule:: platypush.message.response.camera.android - :members: diff --git a/docs/source/platypush/responses/camera.rst b/docs/source/platypush/responses/camera.rst deleted file mode 100644 index 1946a5d03..000000000 --- a/docs/source/platypush/responses/camera.rst +++ /dev/null @@ -1,5 +0,0 @@ -``camera`` -===================================== - -.. automodule:: platypush.message.response.camera - :members: diff --git a/docs/source/responses.rst b/docs/source/responses.rst index 4baff0c48..0f96e27b4 100644 --- a/docs/source/responses.rst +++ b/docs/source/responses.rst @@ -6,8 +6,6 @@ Responses :maxdepth: 1 :caption: Responses: - platypush/responses/camera.rst - platypush/responses/camera.android.rst platypush/responses/google.drive.rst platypush/responses/pihole.rst platypush/responses/printer.cups.rst diff --git a/platypush/message/response/camera/__init__.py b/platypush/message/response/camera/__init__.py deleted file mode 100644 index f584452f9..000000000 --- a/platypush/message/response/camera/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -from platypush.message.response import Response - - -class CameraResponse(Response): - pass - - -# vim:sw=4:ts=4:et: diff --git a/platypush/message/response/camera/android.py b/platypush/message/response/camera/android.py deleted file mode 100644 index 7a750a5ed..000000000 --- a/platypush/message/response/camera/android.py +++ /dev/null @@ -1,191 +0,0 @@ -from typing import List - -from platypush.message.response.camera import CameraResponse - - -class AndroidCameraStatusResponse(CameraResponse): - """ - Example response: - - .. code-block:: json - - { - "stream_url": "https://192.168.1.30:8080/video", - "image_url": "https://192.168.1.30:8080/photo.jpg", - "audio_url": "https://192.168.1.30:8080/audio.wav", - "orientation": "landscape", - "idle": "off", - "audio_only": "off", - "overlay": "off", - "quality": "49", - "focus_homing": "off", - "ip_address": "192.168.1.30", - "motion_limit": "250", - "adet_limit": "200", - "night_vision": "off", - "night_vision_average": "2", - "night_vision_gain": "1.0", - "motion_detect": "off", - "motion_display": "off", - "video_chunk_len": "60", - "gps_active": "off", - "video_size": "1920x1080", - "mirror_flip": "none", - "ffc": "off", - "rtsp_video_formats": "", - "rtsp_audio_formats": "", - "video_connections": "0", - "audio_connections": "0", - "ivideon_streaming": "off", - "zoom": "100", - "crop_x": "50", - "crop_y": "50", - "coloreffect": "none", - "scenemode": "auto", - "focusmode": "continuous-video", - "whitebalance": "auto", - "flashmode": "off", - "antibanding": "off", - "torch": "off", - "focus_distance": "0.0", - "focal_length": "4.25", - "aperture": "1.7", - "filter_density": "0.0", - "exposure_ns": "9384", - "frame_duration": "33333333", - "iso": "100", - "manual_sensor": "off", - "photo_size": "1920x1080" - } - - """ - - attrs = { - 'orientation', 'idle', 'audio_only', 'overlay', 'quality', 'focus_homing', - 'ip_address', 'motion_limit', 'adet_limit', 'night_vision', - 'night_vision_average', 'night_vision_gain', 'motion_detect', - 'motion_display', 'video_chunk_len', 'gps_active', 'video_size', - 'mirror_flip', 'ffc', 'rtsp_video_formats', 'rtsp_audio_formats', - 'video_connections', 'audio_connections', 'ivideon_streaming', 'zoom', - 'crop_x', 'crop_y', 'coloreffect', 'scenemode', 'focusmode', - 'whitebalance', 'flashmode', 'antibanding', 'torch', 'focus_distance', - 'focal_length', 'aperture', 'filter_density', 'exposure_ns', - 'frame_duration', 'iso', 'manual_sensor', 'photo_size', - } - - def __init__(self, *args, - name: str = None, - stream_url: str = None, - image_url: str = None, - audio_url: str = None, - orientation: str = None, - idle: str = None, - audio_only: str = None, - overlay: str = None, - quality: str = None, - focus_homing: str = None, - ip_address: str = None, - motion_limit: str = None, - adet_limit: str = None, - night_vision: str = None, - night_vision_average: str = None, - night_vision_gain: str = None, - motion_detect: str = None, - motion_display: str = None, - video_chunk_len: str = None, - gps_active: str = None, - video_size: str = None, - mirror_flip: str = None, - ffc: str = None, - rtsp_video_formats: str = None, - rtsp_audio_formats: str = None, - video_connections: str = None, - audio_connections: str = None, - ivideon_streaming: str = None, - zoom: str = None, - crop_x: str = None, - crop_y: str = None, - coloreffect: str = None, - scenemode: str = None, - focusmode: str = None, - whitebalance: str = None, - flashmode: str = None, - antibanding: str = None, - torch: str = None, - focus_distance: str = None, - focal_length: str = None, - aperture: str = None, - filter_density: str = None, - exposure_ns: str = None, - frame_duration: str = None, - iso: str = None, - manual_sensor: str = None, - photo_size: str = None, - **kwargs): - - self.status = { - "name": name, - "stream_url": stream_url, - "image_url": image_url, - "audio_url": audio_url, - "orientation": orientation, - "idle": True if idle == "on" else False, - "audio_only": True if audio_only == "on" else False, - "overlay": True if overlay == "on" else False, - "quality": int(quality or 0), - "focus_homing": True if focus_homing == "on" else False, - "ip_address": ip_address, - "motion_limit": int(motion_limit or 0), - "adet_limit": int(adet_limit or 0), - "night_vision": True if night_vision == "on" else False, - "night_vision_average": float(night_vision_average or 0), - "night_vision_gain": float(night_vision_gain or 0), - "motion_detect": True if motion_detect == "on" else False, - "motion_display": True if motion_display == "on" else False, - "video_chunk_len": int(video_chunk_len or 0), - "gps_active": True if gps_active == "on" else False, - "video_size": video_size, - "mirror_flip": mirror_flip, - "ffc": True if ffc == "on" else False, - "rtsp_video_formats": rtsp_video_formats, - "rtsp_audio_formats": rtsp_audio_formats, - "video_connections": int(video_connections or 0), - "audio_connections": int(audio_connections or 0), - "ivideon_streaming": True if ivideon_streaming == "on" else False, - "zoom": int(zoom or 0), - "crop_x": int(crop_x or 0), - "crop_y": int(crop_y or 0), - "coloreffect": coloreffect, - "scenemode": scenemode, - "focusmode": focusmode, - "whitebalance": whitebalance, - "flashmode": True if flashmode == "on" else False, - "antibanding": True if antibanding == "on" else False, - "torch": True if torch == "on" else False, - "focus_distance": float(focus_distance or 0), - "focal_length": float(focal_length or 0), - "aperture": float(aperture or 0), - "filter_density": float(filter_density or 0), - "exposure_ns": int(exposure_ns or 0), - "frame_duration": int(frame_duration or 0), - "iso": int(iso or 0), - "manual_sensor": True if manual_sensor == "on" else False, - "photo_size": photo_size, - } - - super().__init__(*args, output=self.status, **kwargs) - - -class AndroidCameraStatusListResponse(CameraResponse): - def __init__(self, status: List[AndroidCameraStatusResponse], **kwargs): - self.status = [s.status for s in status] - super().__init__(output=self.status, **kwargs) - - -class AndroidCameraPictureResponse(CameraResponse): - def __init__(self, image_file: str, *args, **kwargs): - self.image_file = image_file - super().__init__(*args, output={'image_file': image_file}, **kwargs) - - -# vim:sw=4:ts=4:et: diff --git a/platypush/plugins/camera/android/ipcam/__init__.py b/platypush/plugins/camera/android/ipcam/__init__.py index 86acac63d..e65550dc0 100644 --- a/platypush/plugins/camera/android/ipcam/__init__.py +++ b/platypush/plugins/camera/android/ipcam/__init__.py @@ -3,18 +3,29 @@ import os import requests from requests.auth import HTTPBasicAuth -from typing import Optional, Union, Dict, List, Any +from typing import Optional, Sequence, Union, Dict, List, Any -from platypush.message.response.camera.android import AndroidCameraStatusResponse, AndroidCameraStatusListResponse, \ - AndroidCameraPictureResponse from platypush.plugins import Plugin, action +from platypush.schemas.camera.android.ipcam import CameraStatusSchema class AndroidIpcam: + """ + IPCam camera configuration. + """ + args = {} - def __init__(self, name: str, host: str, port: int = 8080, username: Optional[str] = None, - password: Optional[str] = None, timeout: int = 10, ssl: bool = True): + def __init__( + self, + name: str, + host: str, + port: int = 8080, + username: Optional[str] = None, + password: Optional[str] = None, + timeout: int = 10, + ssl: bool = True, + ): self.args = { 'name': name, 'host': host, @@ -41,12 +52,13 @@ class AndroidIpcam: self.args[key] = value def __str__(self): - return json.dumps(getattr(self, 'args') or {}) + return json.dumps(self.args or {}) @property def base_url(self) -> str: return 'http{ssl}://{host}:{port}/'.format( - ssl=('s' if self.ssl else ''), host=self.host, port=self.port) + ssl=('s' if self.ssl else ''), host=self.host, port=self.port + ) @property def stream_url(self) -> str: @@ -67,16 +79,20 @@ class CameraAndroidIpcamPlugin(Plugin): `IPCam `_. """ - def __init__(self, - host: Optional[str] = None, - port: Optional[int] = 8080, - username: Optional[str] = None, - password: Optional[str] = None, - timeout: int = 10, - ssl: bool = True, - cameras: Optional[Dict[str, Dict[str, Any]]] = None, - **kwargs): + def __init__( + self, + name: Optional[str] = None, + host: Optional[str] = None, + port: int = 8080, + username: Optional[str] = None, + password: Optional[str] = None, + timeout: int = 10, + ssl: bool = True, + cameras: Optional[Sequence[dict]] = None, + **kwargs, + ): """ + :param name: Custom name for the default camera (default: IP/hostname) :param host: Camera host name or address :param port: Camera port :param username: Camera username, if set @@ -84,28 +100,44 @@ class CameraAndroidIpcamPlugin(Plugin): :param timeout: Connection timeout :param ssl: Use HTTPS instead of HTTP :param cameras: Alternatively, you can specify a list of IPCam cameras as a - name->dict mapping. The keys will be unique names used to identify your - cameras, the values will contain dictionaries containing `host, `port`, - `username`, `password`, `timeout` and `ssl` attributes for each camera. + list of objects with ``name``, ``host``, ``port``, ``username``, + ``password``, ``timeout`` and ``ssl`` attributes. """ super().__init__(**kwargs) self.cameras: List[AndroidIpcam] = [] self._camera_name_to_idx: Dict[str, int] = {} if not cameras: - camera = AndroidIpcam(name=host, host=host, port=port, username=username, - password=password, timeout=timeout, ssl=ssl) + assert host, 'You need to specify at least one camera' + name = name or host + camera = AndroidIpcam( + name=name, + host=host, + port=port, + username=username, + password=password, + timeout=timeout, + ssl=ssl, + ) self.cameras.append(camera) - self._camera_name_to_idx[host] = 0 + self._camera_name_to_idx[name] = 0 else: - for name, camera in cameras.items(): - camera = AndroidIpcam(name=name, host=camera['host'], port=camera.get('port', port), - username=camera.get('username'), password=camera.get('password'), - timeout=camera.get('timeout', timeout), ssl=camera.get('ssl', ssl)) + for camera in cameras: + assert 'host' in camera, 'You need to specify the host for each camera' + name = camera.get('name', camera['host']) + camera = AndroidIpcam( + name=name, + host=camera['host'], + port=camera.get('port', port), + username=camera.get('username'), + password=camera.get('password'), + timeout=camera.get('timeout', timeout), + ssl=camera.get('ssl', ssl), + ) self._camera_name_to_idx[name] = len(self.cameras) self.cameras.append(camera) - def _get_camera(self, camera: Union[int, str] = None) -> AndroidIpcam: + def _get_camera(self, camera: Optional[Union[int, str]] = None) -> AndroidIpcam: if not camera: camera = 0 @@ -114,10 +146,14 @@ class CameraAndroidIpcamPlugin(Plugin): return self.cameras[self._camera_name_to_idx[camera]] - def _exec(self, url: str, camera: Union[int, str] = None, *args, **kwargs) -> Union[Dict[str, Any], bool]: + def _exec( + self, url: str, *args, camera: Optional[Union[int, str]] = None, **kwargs + ) -> Union[Dict[str, Any], bool]: cam = self._get_camera(camera) url = cam.base_url + url - response = requests.get(url, auth=cam.auth, timeout=cam.timeout, verify=False, *args, **kwargs) + response = requests.get( + url, auth=cam.auth, timeout=cam.timeout, verify=False, *args, **kwargs + ) response.raise_for_status() if response.headers.get('content-type') == 'application/json': @@ -125,87 +161,138 @@ class CameraAndroidIpcamPlugin(Plugin): return response.text.find('Ok') != -1 - @action - def change_setting(self, key: str, value: Union[str, int, bool], camera: Union[int, str] = None) -> bool: - """ - Change a setting. - :param key: Setting name - :param value: Setting value - :param camera: Camera index or configured name - :return: True on success, False otherwise - """ + def _change_setting( + self, + key: str, + value: Union[str, int, bool], + camera: Optional[Union[int, str]] = None, + ) -> bool: if isinstance(value, bool): payload = "on" if value else "off" else: payload = value - return self._exec("settings/{key}?set={payload}".format(key=key, payload=payload), camera=camera) + return bool( + self._exec( + "settings/{key}?set={payload}".format(key=key, payload=payload), + camera=camera, + ) + ) @action - def status(self, camera: Union[int, str] = None) -> AndroidCameraStatusListResponse: + def change_setting( + self, + key: str, + value: Union[str, int, bool], + camera: Optional[Union[int, str]] = None, + ) -> bool: + """ + Change a setting. + + :param key: Setting name + :param value: Setting value + :param camera: Camera index or configured name + :return: True on success, False otherwise + """ + return self._change_setting(key, value, camera=camera) + + @action + def status(self, camera: Optional[Union[int, str]] = None) -> List[dict]: """ :param camera: Camera index or name (default: status of all the cameras) - :return: True if the camera is available, False otherwise + :return: .. schema:: camera.android.ipcam.CameraStatusSchema(many=True) """ cameras = self._camera_name_to_idx.keys() if camera is None else [camera] statuses = [] for c in cameras: try: - if isinstance(camera, int): + print('****** HERE ******') + print(self._camera_name_to_idx) + if isinstance(c, int): cam = self.cameras[c] else: cam = self.cameras[self._camera_name_to_idx[c]] - status_data = self._exec('status.json', params={'show_avail': 1}, camera=cam.name).get('curvals', {}) - status = AndroidCameraStatusResponse( - name=cam.name, - stream_url=cam.stream_url, - image_url=cam.image_url, - audio_url=cam.audio_url, - **{k: v for k, v in status_data.items() - if k in AndroidCameraStatusResponse.attrs}) + response = self._exec( + 'status.json', params={'show_avail': 1}, camera=cam.name + ) + assert isinstance(response, dict), f'Invalid response: {response}' + + status_data = response.get('curvals', {}) + status = CameraStatusSchema().dump( + { + 'name': cam.name, + 'stream_url': cam.stream_url, + 'image_url': cam.image_url, + 'audio_url': cam.audio_url, + **status_data, + } + ) statuses.append(status) except Exception as e: - self.logger.warning('Could not get the status of {}: {}'.format(c, str(e))) + self.logger.warning( + 'Could not get the status of %s: %s: %s', c, type(e), e + ) - return AndroidCameraStatusListResponse(statuses) + return statuses @action - def set_front_facing_camera(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_front_facing_camera( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable the front-facing camera.""" - return self.change_setting('ffc', activate, camera=camera) + return self._change_setting('ffc', activate, camera=camera) @action - def set_torch(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_torch( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable the torch.""" url = 'enabletorch' if activate else 'disabletorch' - return self._exec(url, camera=camera) + return bool(self._exec(url, camera=camera)) @action - def set_focus(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_focus( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable the focus.""" url = 'focus' if activate else 'nofocus' - return self._exec(url, camera=camera) + return bool(self._exec(url, camera=camera)) @action - def start_recording(self, tag: Optional[str] = None, camera: Union[int, str] = None) -> bool: + def start_recording( + self, tag: Optional[str] = None, camera: Optional[Union[int, str]] = None + ) -> bool: """Start recording.""" - params = {'force': 1} - if tag: - params['tag'] = tag - - return self._exec('startvideo', params=params, camera=camera) + params = { + 'force': 1, + **({'tag': tag} if tag else {}), + } + return bool(self._exec('startvideo', params=params, camera=camera)) @action - def stop_recording(self, camera: Union[int, str] = None) -> bool: + def stop_recording(self, camera: Optional[Union[int, str]] = None) -> bool: """Stop recording.""" - return self._exec('stopvideo', params={'force': 1}, camera=camera) + return bool(self._exec('stopvideo', params={'force': 1}, camera=camera)) @action - def take_picture(self, image_file: str, camera: Union[int, str] = None) -> AndroidCameraPictureResponse: - """Take a picture and save it on the local device.""" + def take_picture( + self, image_file: str, camera: Optional[Union[int, str]] = None + ) -> dict: + """ + Take a picture and save it on the local device. + + :return: dict + + .. code-block:: json + + { + "image_file": "/path/to/image.jpg" + } + + """ cam = self._get_camera(camera) image_file = os.path.abspath(os.path.expanduser(image_file)) os.makedirs(os.path.dirname(image_file), exist_ok=True) @@ -214,47 +301,64 @@ class CameraAndroidIpcamPlugin(Plugin): with open(image_file, 'wb') as f: f.write(response.content) - return AndroidCameraPictureResponse(image_file=image_file) + + return {'image_file': image_file} @action - def set_night_vision(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_night_vision( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable night vision.""" - return self.change_setting('night_vision', activate, camera=camera) + return self._change_setting('night_vision', activate, camera=camera) @action - def set_overlay(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_overlay( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable video overlay.""" - return self.change_setting('overlay', activate, camera=camera) + return self._change_setting('overlay', activate, camera=camera) @action - def set_gps(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_gps( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable GPS.""" - return self.change_setting('gps_active', activate, camera=camera) + return self._change_setting('gps_active', activate, camera=camera) @action - def set_quality(self, quality: int = 100, camera: Union[int, str] = None) -> bool: + def set_quality( + self, quality: int = 100, camera: Optional[Union[int, str]] = None + ) -> bool: """Set video quality.""" - return self.change_setting('quality', int(quality), camera=camera) + return self._change_setting('quality', int(quality), camera=camera) @action - def set_motion_detect(self, activate: bool = True, camera: Union[int, str] = None) -> bool: + def set_motion_detect( + self, activate: bool = True, camera: Optional[Union[int, str]] = None + ) -> bool: """Enable/disable motion detect.""" - return self.change_setting('motion_detect', activate, camera=camera) + return self._change_setting('motion_detect', activate, camera=camera) @action - def set_orientation(self, orientation: str = 'landscape', camera: Union[int, str] = None) -> bool: + def set_orientation( + self, orientation: str = 'landscape', camera: Optional[Union[int, str]] = None + ) -> bool: """Set video orientation.""" - return self.change_setting('orientation', orientation, camera=camera) + return self._change_setting('orientation', orientation, camera=camera) @action - def set_zoom(self, zoom: float, camera: Union[int, str] = None) -> bool: + def set_zoom(self, zoom: float, camera: Optional[Union[int, str]] = None) -> bool: """Set the zoom level.""" - return self._exec('settings/ptz', params={'zoom': float(zoom)}, camera=camera) + return bool( + self._exec('settings/ptz', params={'zoom': float(zoom)}, camera=camera) + ) @action - def set_scenemode(self, scenemode: str = 'auto', camera: Union[int, str] = None) -> bool: + def set_scenemode( + self, scenemode: str = 'auto', camera: Optional[Union[int, str]] = None + ) -> bool: """Set video orientation.""" - return self.change_setting('scenemode', scenemode, camera=camera) + return self._change_setting('scenemode', scenemode, camera=camera) # vim:sw=4:ts=4:et: diff --git a/platypush/schemas/camera/android/__init__.py b/platypush/schemas/camera/android/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/platypush/schemas/camera/android/ipcam.py b/platypush/schemas/camera/android/ipcam.py new file mode 100644 index 000000000..a85f7ae16 --- /dev/null +++ b/platypush/schemas/camera/android/ipcam.py @@ -0,0 +1,332 @@ +from marshmallow import EXCLUDE, fields, pre_dump +from marshmallow.schema import Schema +from marshmallow.validate import OneOf + +from platypush.schemas import StrippedString + + +class CameraStatusSchema(Schema): + """ + Schema for the camera status. + """ + + class Meta: # type: ignore + """ + Exclude unknown fields. + """ + + unknown = EXCLUDE + + name = StrippedString( + required=True, + metadata={ + 'description': 'Name or IP of the camera', + 'example': 'Front Door', + }, + ) + + stream_url = fields.Url( + required=True, + metadata={ + 'description': 'URL to the video stream', + 'example': 'http://192.168.1.10:8080/video', + }, + ) + + image_url = fields.Url( + required=True, + metadata={ + 'description': 'URL to get a snapshot from the camera', + 'example': 'http://192.168.1.10:8080/photo.jpg', + }, + ) + + audio_url = fields.Url( + required=True, + metadata={ + 'description': 'URL to get audio from the camera', + 'example': 'http://192.168.1.10:8080/audio.wav', + }, + ) + + orientation = fields.Str( + required=True, + metadata={ + 'description': 'Orientation of the camera', + 'example': 'landscape', + 'validate': OneOf(['landscape', 'portrait']), + }, + ) + + idle = fields.Bool( + required=True, + metadata={ + 'description': 'Idle status of the camera', + 'example': False, + }, + ) + + audio_only = fields.Bool( + required=True, + metadata={ + 'description': 'Whether the camera is in audio-only mode', + 'example': False, + }, + ) + + overlay = fields.Bool( + required=True, + metadata={ + 'description': 'Whether the camera is in overlay mode', + 'example': False, + }, + ) + + quality = fields.Int( + required=True, + metadata={ + 'description': 'Quality of the video stream, in percent', + 'example': 49, + }, + ) + + night_vision = fields.Bool( + required=True, + metadata={ + 'description': 'Whether night vision is enabled', + 'example': False, + }, + ) + + night_vision_average = fields.Int( + required=True, + metadata={ + 'description': 'Average brightness for night vision', + 'example': 2, + }, + ) + + night_vision_gain = fields.Float( + required=True, + metadata={ + 'description': 'Brightness gain for night vision', + 'example': 1.0, + }, + ) + + ip_address = fields.Str( + required=True, + metadata={ + 'description': 'IP address of the camera', + 'example': '192.168.1.10', + }, + ) + + motion_limit = fields.Int( + required=True, + metadata={ + 'description': 'Motion limit', + 'example': 250, + }, + ) + + motion_detect = fields.Bool( + required=True, + metadata={ + 'description': 'Whether motion detection is enabled', + 'example': False, + }, + ) + + motion_display = fields.Bool( + required=True, + metadata={ + 'description': 'Whether motion display is enabled', + 'example': False, + }, + ) + + gps_active = fields.Bool( + required=True, + metadata={ + 'description': 'Whether GPS is active', + 'example': False, + }, + ) + + video_size = fields.Str( + required=True, + metadata={ + 'description': 'Size of the video stream', + 'example': '1920x1080', + }, + ) + + photo_size = fields.Str( + required=True, + metadata={ + 'description': 'Size of the photo', + 'example': '1920x1080', + }, + ) + + mirror_flip = fields.Str( + required=True, + metadata={ + 'description': 'Mirror/flip mode', + 'example': 'none', + 'validate': OneOf(['none', 'horizontal', 'vertical', 'both']), + }, + ) + + video_connections = fields.Int( + required=True, + metadata={ + 'description': 'Number of active video connections', + 'example': 0, + }, + ) + + audio_connections = fields.Int( + required=True, + metadata={ + 'description': 'Number of active audio connections', + 'example': 0, + }, + ) + + zoom = fields.Int( + required=True, + metadata={ + 'description': 'Zoom level, as a percentage', + 'example': 100, + }, + ) + + crop_x = fields.Int( + required=True, + metadata={ + 'description': 'Crop X, as a percentage', + 'example': 50, + }, + ) + + crop_y = fields.Int( + required=True, + metadata={ + 'description': 'Crop Y, as a percentage', + 'example': 50, + }, + ) + + coloreffect = fields.Str( + required=True, + metadata={ + 'description': 'Color effect', + 'example': 'none', + }, + ) + + scenemode = fields.Str( + required=True, + metadata={ + 'description': 'Scene mode', + 'example': 'auto', + }, + ) + + focusmode = fields.Str( + required=True, + metadata={ + 'description': 'Focus mode', + 'example': 'continuous-video', + }, + ) + + whitebalance = fields.Str( + required=True, + metadata={ + 'description': 'White balance', + 'example': 'auto', + }, + ) + + flashmode = fields.Str( + required=True, + validate=OneOf(['off', 'on', 'auto']), + metadata={ + 'description': 'Flash mode', + 'example': 'off', + }, + ) + + torch = fields.Bool( + required=True, + metadata={ + 'description': 'Whether the torch is enabled', + 'example': False, + }, + ) + + focus_distance = fields.Float( + required=True, + metadata={ + 'description': 'Focus distance', + 'example': 0.0, + }, + ) + + focal_length = fields.Float( + required=True, + metadata={ + 'description': 'Focal length', + 'example': 4.25, + }, + ) + + aperture = fields.Float( + required=True, + metadata={ + 'description': 'Aperture', + 'example': 1.7, + }, + ) + + filter_density = fields.Float( + required=True, + metadata={ + 'description': 'Filter density', + 'example': 0.0, + }, + ) + + exposure_ns = fields.Int( + required=True, + metadata={ + 'description': 'Exposure time in nanoseconds', + 'example': 9384, + }, + ) + + iso = fields.Int( + required=True, + metadata={ + 'description': 'ISO', + 'example': 100, + }, + ) + + manual_sensor = fields.Bool( + required=True, + metadata={ + 'description': 'Whether the sensor is in manual mode', + 'example': False, + }, + ) + + @pre_dump + def normalize_bools(self, data, **_): + for k, v in data.items(): + if k != 'flashmode' and isinstance(v, str) and v.lower() in ['on', 'off']: + data[k] = v.lower() == 'on' + return data