[#398] Replaced camera response objects with schemas.

This commit is contained in:
Fabio Manganiello 2024-05-09 01:40:21 +02:00
parent 579faf63bc
commit 13bde4adba
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
8 changed files with 521 additions and 296 deletions

View file

@ -1,5 +0,0 @@
``camera.android``
=============================================
.. automodule:: platypush.message.response.camera.android
:members:

View file

@ -1,5 +0,0 @@
``camera``
=====================================
.. automodule:: platypush.message.response.camera
:members:

View file

@ -6,8 +6,6 @@ Responses
:maxdepth: 1 :maxdepth: 1
:caption: Responses: :caption: Responses:
platypush/responses/camera.rst
platypush/responses/camera.android.rst
platypush/responses/google.drive.rst platypush/responses/google.drive.rst
platypush/responses/pihole.rst platypush/responses/pihole.rst
platypush/responses/printer.cups.rst platypush/responses/printer.cups.rst

View file

@ -1,8 +0,0 @@
from platypush.message.response import Response
class CameraResponse(Response):
pass
# vim:sw=4:ts=4:et:

View file

@ -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:

View file

@ -3,18 +3,29 @@ import os
import requests import requests
from requests.auth import HTTPBasicAuth 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.plugins import Plugin, action
from platypush.schemas.camera.android.ipcam import CameraStatusSchema
class AndroidIpcam: class AndroidIpcam:
"""
IPCam camera configuration.
"""
args = {} args = {}
def __init__(self, name: str, host: str, port: int = 8080, username: Optional[str] = None, def __init__(
password: Optional[str] = None, timeout: int = 10, ssl: bool = True): self,
name: str,
host: str,
port: int = 8080,
username: Optional[str] = None,
password: Optional[str] = None,
timeout: int = 10,
ssl: bool = True,
):
self.args = { self.args = {
'name': name, 'name': name,
'host': host, 'host': host,
@ -41,12 +52,13 @@ class AndroidIpcam:
self.args[key] = value self.args[key] = value
def __str__(self): def __str__(self):
return json.dumps(getattr(self, 'args') or {}) return json.dumps(self.args or {})
@property @property
def base_url(self) -> str: def base_url(self) -> str:
return 'http{ssl}://{host}:{port}/'.format( 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 @property
def stream_url(self) -> str: def stream_url(self) -> str:
@ -67,16 +79,20 @@ class CameraAndroidIpcamPlugin(Plugin):
`IPCam <https://play.google.com/store/apps/details?id=com.pas.webcam>`_. `IPCam <https://play.google.com/store/apps/details?id=com.pas.webcam>`_.
""" """
def __init__(self, def __init__(
host: Optional[str] = None, self,
port: Optional[int] = 8080, name: Optional[str] = None,
username: Optional[str] = None, host: Optional[str] = None,
password: Optional[str] = None, port: int = 8080,
timeout: int = 10, username: Optional[str] = None,
ssl: bool = True, password: Optional[str] = None,
cameras: Optional[Dict[str, Dict[str, Any]]] = None, timeout: int = 10,
**kwargs): 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 host: Camera host name or address
:param port: Camera port :param port: Camera port
:param username: Camera username, if set :param username: Camera username, if set
@ -84,28 +100,44 @@ class CameraAndroidIpcamPlugin(Plugin):
:param timeout: Connection timeout :param timeout: Connection timeout
:param ssl: Use HTTPS instead of HTTP :param ssl: Use HTTPS instead of HTTP
:param cameras: Alternatively, you can specify a list of IPCam cameras as a :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 list of objects with ``name``, ``host``, ``port``, ``username``,
cameras, the values will contain dictionaries containing `host, `port`, ``password``, ``timeout`` and ``ssl`` attributes.
`username`, `password`, `timeout` and `ssl` attributes for each camera.
""" """
super().__init__(**kwargs) super().__init__(**kwargs)
self.cameras: List[AndroidIpcam] = [] self.cameras: List[AndroidIpcam] = []
self._camera_name_to_idx: Dict[str, int] = {} self._camera_name_to_idx: Dict[str, int] = {}
if not cameras: if not cameras:
camera = AndroidIpcam(name=host, host=host, port=port, username=username, assert host, 'You need to specify at least one camera'
password=password, timeout=timeout, ssl=ssl) 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.cameras.append(camera)
self._camera_name_to_idx[host] = 0 self._camera_name_to_idx[name] = 0
else: else:
for name, camera in cameras.items(): for camera in cameras:
camera = AndroidIpcam(name=name, host=camera['host'], port=camera.get('port', port), assert 'host' in camera, 'You need to specify the host for each camera'
username=camera.get('username'), password=camera.get('password'), name = camera.get('name', camera['host'])
timeout=camera.get('timeout', timeout), ssl=camera.get('ssl', ssl)) 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._camera_name_to_idx[name] = len(self.cameras)
self.cameras.append(camera) 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: if not camera:
camera = 0 camera = 0
@ -114,10 +146,14 @@ class CameraAndroidIpcamPlugin(Plugin):
return self.cameras[self._camera_name_to_idx[camera]] 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) cam = self._get_camera(camera)
url = cam.base_url + url 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() response.raise_for_status()
if response.headers.get('content-type') == 'application/json': if response.headers.get('content-type') == 'application/json':
@ -125,87 +161,138 @@ class CameraAndroidIpcamPlugin(Plugin):
return response.text.find('Ok') != -1 return response.text.find('Ok') != -1
@action def _change_setting(
def change_setting(self, key: str, value: Union[str, int, bool], camera: Union[int, str] = None) -> bool: self,
""" key: str,
Change a setting. value: Union[str, int, bool],
:param key: Setting name camera: Optional[Union[int, str]] = None,
:param value: Setting value ) -> bool:
:param camera: Camera index or configured name
:return: True on success, False otherwise
"""
if isinstance(value, bool): if isinstance(value, bool):
payload = "on" if value else "off" payload = "on" if value else "off"
else: else:
payload = value 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 @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) :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] cameras = self._camera_name_to_idx.keys() if camera is None else [camera]
statuses = [] statuses = []
for c in cameras: for c in cameras:
try: try:
if isinstance(camera, int): print('****** HERE ******')
print(self._camera_name_to_idx)
if isinstance(c, int):
cam = self.cameras[c] cam = self.cameras[c]
else: else:
cam = self.cameras[self._camera_name_to_idx[c]] cam = self.cameras[self._camera_name_to_idx[c]]
status_data = self._exec('status.json', params={'show_avail': 1}, camera=cam.name).get('curvals', {}) response = self._exec(
status = AndroidCameraStatusResponse( 'status.json', params={'show_avail': 1}, camera=cam.name
name=cam.name, )
stream_url=cam.stream_url, assert isinstance(response, dict), f'Invalid response: {response}'
image_url=cam.image_url,
audio_url=cam.audio_url, status_data = response.get('curvals', {})
**{k: v for k, v in status_data.items() status = CameraStatusSchema().dump(
if k in AndroidCameraStatusResponse.attrs}) {
'name': cam.name,
'stream_url': cam.stream_url,
'image_url': cam.image_url,
'audio_url': cam.audio_url,
**status_data,
}
)
statuses.append(status) statuses.append(status)
except Exception as e: 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 @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.""" """Enable/disable the front-facing camera."""
return self.change_setting('ffc', activate, camera=camera) return self._change_setting('ffc', activate, camera=camera)
@action @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.""" """Enable/disable the torch."""
url = 'enabletorch' if activate else 'disabletorch' url = 'enabletorch' if activate else 'disabletorch'
return self._exec(url, camera=camera) return bool(self._exec(url, camera=camera))
@action @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.""" """Enable/disable the focus."""
url = 'focus' if activate else 'nofocus' url = 'focus' if activate else 'nofocus'
return self._exec(url, camera=camera) return bool(self._exec(url, camera=camera))
@action @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.""" """Start recording."""
params = {'force': 1} params = {
if tag: 'force': 1,
params['tag'] = tag **({'tag': tag} if tag else {}),
}
return self._exec('startvideo', params=params, camera=camera) return bool(self._exec('startvideo', params=params, camera=camera))
@action @action
def stop_recording(self, camera: Union[int, str] = None) -> bool: def stop_recording(self, camera: Optional[Union[int, str]] = None) -> bool:
"""Stop recording.""" """Stop recording."""
return self._exec('stopvideo', params={'force': 1}, camera=camera) return bool(self._exec('stopvideo', params={'force': 1}, camera=camera))
@action @action
def take_picture(self, image_file: str, camera: Union[int, str] = None) -> AndroidCameraPictureResponse: def take_picture(
"""Take a picture and save it on the local device.""" 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) cam = self._get_camera(camera)
image_file = os.path.abspath(os.path.expanduser(image_file)) image_file = os.path.abspath(os.path.expanduser(image_file))
os.makedirs(os.path.dirname(image_file), exist_ok=True) os.makedirs(os.path.dirname(image_file), exist_ok=True)
@ -214,47 +301,64 @@ class CameraAndroidIpcamPlugin(Plugin):
with open(image_file, 'wb') as f: with open(image_file, 'wb') as f:
f.write(response.content) f.write(response.content)
return AndroidCameraPictureResponse(image_file=image_file)
return {'image_file': image_file}
@action @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.""" """Enable/disable night vision."""
return self.change_setting('night_vision', activate, camera=camera) return self._change_setting('night_vision', activate, camera=camera)
@action @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.""" """Enable/disable video overlay."""
return self.change_setting('overlay', activate, camera=camera) return self._change_setting('overlay', activate, camera=camera)
@action @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.""" """Enable/disable GPS."""
return self.change_setting('gps_active', activate, camera=camera) return self._change_setting('gps_active', activate, camera=camera)
@action @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.""" """Set video quality."""
return self.change_setting('quality', int(quality), camera=camera) return self._change_setting('quality', int(quality), camera=camera)
@action @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.""" """Enable/disable motion detect."""
return self.change_setting('motion_detect', activate, camera=camera) return self._change_setting('motion_detect', activate, camera=camera)
@action @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.""" """Set video orientation."""
return self.change_setting('orientation', orientation, camera=camera) return self._change_setting('orientation', orientation, camera=camera)
@action @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.""" """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 @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.""" """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: # vim:sw=4:ts=4:et:

View file

@ -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