import json
import os
import requests
from requests.auth import HTTPBasicAuth
from typing import Optional, Union, Dict, List, Any
from platypush.message.response.camera.android import AndroidCameraStatusResponse, AndroidCameraStatusListResponse, \
AndroidCameraPictureResponse
from platypush.plugins import Plugin, action
class AndroidIpcam:
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):
self.args = {
'name': name,
'host': host,
'port': port,
'username': username,
'password': password,
'timeout': timeout,
'ssl': ssl,
}
self.auth = None
if username:
self.auth = HTTPBasicAuth(self.username, self.password)
def __getattr__(self, item):
if item in self.args:
return super().__getattribute__('args').get(item)
return super().__getattribute__(item)
def __setattr__(self, key, value):
if key not in self.args:
super().__setattr__(key, value)
else:
self.args[key] = value
def __str__(self):
return json.dumps(getattr(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)
@property
def stream_url(self) -> str:
return self.base_url + 'video'
@property
def audio_url(self) -> str:
return self.base_url + 'audio.wav'
@property
def image_url(self) -> str:
return self.base_url + 'photo.jpg'
[docs]class CameraAndroidIpcamPlugin(Plugin):
"""
Plugin to control remote Android cameras over
`IPCam <https://play.google.com/store/apps/details?id=com.pas.webcam>`_.
"""
[docs] 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):
"""
:param host: Camera host name or address
:param port: Camera port
:param username: Camera username, if set
:param password: Camera password, if set
: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.
"""
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)
self.cameras.append(camera)
self._camera_name_to_idx[host] = 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))
self._camera_name_to_idx[name] = len(self.cameras)
self.cameras.append(camera)
def _get_camera(self, camera: Union[int, str] = None) -> AndroidIpcam:
if not camera:
camera = 0
if isinstance(camera, int):
return self.cameras[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]:
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.raise_for_status()
if response.headers.get('content-type') == 'application/json':
return response.json()
return response.text.find('Ok') != -1
[docs] @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
"""
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)
[docs] @action
def status(self, camera: Union[int, str] = None) -> AndroidCameraStatusListResponse:
"""
:param camera: Camera index or name (default: status of all the cameras)
:return: True if the camera is available, False otherwise
"""
cameras = self._camera_name_to_idx.keys() if camera is None else [camera]
statuses = []
for c in cameras:
try:
if isinstance(camera, 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})
statuses.append(status)
except Exception as e:
self.logger.warning('Could not get the status of {}: {}'.format(c, str(e)))
return AndroidCameraStatusListResponse(statuses)
[docs] @action
def set_front_facing_camera(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable the front-facing camera."""
return self.change_setting('ffc', activate, camera=camera)
[docs] @action
def set_torch(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable the torch."""
url = 'enabletorch' if activate else 'disabletorch'
return self._exec(url, camera=camera)
[docs] @action
def set_focus(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable the focus."""
url = 'focus' if activate else 'nofocus'
return self._exec(url, camera=camera)
[docs] @action
def start_recording(self, tag: Optional[str] = None, camera: Union[int, str] = None) -> bool:
"""Start recording."""
params = {'force': 1}
if tag:
params['tag'] = tag
return self._exec('startvideo', params=params, camera=camera)
[docs] @action
def stop_recording(self, camera: Union[int, str] = None) -> bool:
"""Stop recording."""
return self._exec('stopvideo', params={'force': 1}, camera=camera)
[docs] @action
def take_picture(self, image_file: str, camera: Union[int, str] = None) -> AndroidCameraPictureResponse:
"""Take a picture and save it on the local device."""
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)
response = requests.get(cam.image_url, auth=cam.auth, verify=False)
response.raise_for_status()
with open(image_file, 'wb') as f:
f.write(response.content)
return AndroidCameraPictureResponse(image_file=image_file)
[docs] @action
def set_night_vision(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable night vision."""
return self.change_setting('night_vision', activate, camera=camera)
[docs] @action
def set_overlay(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable video overlay."""
return self.change_setting('overlay', activate, camera=camera)
[docs] @action
def set_gps(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable GPS."""
return self.change_setting('gps_active', activate, camera=camera)
[docs] @action
def set_quality(self, quality: int = 100, camera: Union[int, str] = None) -> bool:
"""Set video quality."""
return self.change_setting('quality', int(quality), camera=camera)
[docs] @action
def set_motion_detect(self, activate: bool = True, camera: Union[int, str] = None) -> bool:
"""Enable/disable motion detect."""
return self.change_setting('motion_detect', activate, camera=camera)
[docs] @action
def set_orientation(self, orientation: str = 'landscape', camera: Union[int, str] = None) -> bool:
"""Set video orientation."""
return self.change_setting('orientation', orientation, camera=camera)
[docs] @action
def set_zoom(self, zoom: float, camera: Union[int, str] = None) -> bool:
"""Set the zoom level."""
return self._exec('settings/ptz', params={'zoom': float(zoom)}, camera=camera)
[docs] @action
def set_scenemode(self, scenemode: str = 'auto', camera: Union[int, str] = None) -> bool:
"""Set video orientation."""
return self.change_setting('scenemode', scenemode, camera=camera)
# vim:sw=4:ts=4:et:
