Support for camera snapshot and stream endpoints and for disabling logging on response and event messages

This commit is contained in:
Fabio Manganiello 2019-03-06 02:01:17 +01:00
parent dcb0555571
commit 0596d77403
9 changed files with 371 additions and 66 deletions

View file

@ -143,6 +143,7 @@ class Daemon:
LOGGER.info('Received STOP event: {}'.format(msg))
self.stop_app()
elif isinstance(msg, Event):
if not msg.disable_logging:
LOGGER.info('Received event: {}'.format(msg))
self.event_processor.process_event(msg)

View file

@ -0,0 +1,112 @@
import json
import os
import shutil
import tempfile
import time
from flask import abort, jsonify, request, Response, Blueprint, \
send_from_directory
from platypush.backend.http.app import template_folder
from platypush.backend.http.app.utils import logger, get_remote_base_url, \
authenticate_user, send_request
camera = Blueprint('camera', __name__, template_folder=template_folder)
# Declare routes list
__routes__ = [
camera,
]
def get_device_id(device_id=None):
if device_id is None:
device_id = str(send_request(action='camera.get_default_device_id').output)
return device_id
def get_frame_file(device_id=None):
device_id = get_device_id(device_id)
was_recording = True
frame_file = None
status = send_request(action='camera.status', device_id=device_id).output
if device_id not in status:
was_recording = False
response = send_request(action='camera.start_recording',
device_id=device_id)
while not frame_file:
frame_file = send_request(action='camera.status', device_id=device_id). \
output.get(device_id, {}).get('image_file')
if not frame_file:
time.sleep(0.1)
if not was_recording:
# stop_recording will delete the temporary frames. Copy the image file
# to a temporary file before stopping recording
tmp_file = None
with tempfile.NamedTemporaryFile(prefix='camera_capture_', suffix='.jpg',
delete=False) as f:
tmp_file = f.name
shutil.copyfile(frame_file, tmp_file)
frame_file = tmp_file
send_request(action='camera.stop_recording', device_id=device_id)
return frame_file
def video_feed(device_id=None):
device_id = get_device_id(device_id)
send_request(action='camera.start_recording', device_id=device_id)
last_frame_file = None
last_frame = None
try:
while True:
frame_file = get_frame_file(device_id)
if frame_file == last_frame_file:
continue
with open(frame_file, 'rb') as f:
frame = f.read()
yield (b'--frame\r\n'
b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
last_frame_file = frame_file
last_frame = frame
finally:
send_request(action='camera.stop_recording', device_id=device_id)
@camera.route('/camera/<device_id>/frame', methods=['GET'])
@authenticate_user
def get_camera_frame(device_id):
frame_file = get_frame_file(device_id)
return send_from_directory(os.path.dirname(frame_file),
os.path.basename(frame_file))
@camera.route('/camera/frame', methods=['GET'])
@authenticate_user
def get_default_camera_frame():
frame_file = get_frame_file()
return send_from_directory(os.path.dirname(frame_file),
os.path.basename(frame_file))
@camera.route('/camera/stream', methods=['GET'])
@authenticate_user
def get_default_stream_feed():
return Response(video_feed(),
mimetype='multipart/x-mixed-replace; boundary=frame')
@camera.route('/camera/<device_id>/stream', methods=['GET'])
@authenticate_user
def get_stream_feed(device_id):
return Response(video_feed(device_id),
mimetype='multipart/x-mixed-replace; boundary=frame')
# vim:sw=4:ts=4:et:

View file

@ -4,7 +4,8 @@ import logging
import os
import sys
from flask import Response
from functools import wraps
from flask import request, Response
from redis import Redis
# NOTE: The HTTP service will *only* work on top of a Redis bus. The default
@ -82,11 +83,21 @@ def send_message(msg):
if isinstance(msg, Request):
response = get_message_response(msg)
logger().info('Processing response on the HTTP backend: {}'.
logger().debug('Processing response on the HTTP backend: {}'.
format(response))
return response
def send_request(action, **kwargs):
msg = {
'type': 'request',
'action': action
}
if kwargs:
msg['args'] = kwargs
return send_message(msg)
def authenticate():
return Response('Authentication required', 401,
@ -123,13 +134,19 @@ def get_routes():
routes_dir = os.path.join(
os.path.dirname(os.path.abspath(__file__)), 'routes')
routes = []
base_module = '.'.join(__name__.split('.')[:-1])
for path, dirs, files in os.walk(routes_dir):
for f in files:
if f.endswith('.py'):
sys.path.insert(0, path)
mod_name = '.'.join(
(base_module + '.' + os.path.join(path, f).replace(
os.path.dirname(__file__), '')[1:] \
.replace(os.sep, '.')).split('.') \
[:(-2 if f == '__init__.py' else -1)])
try:
mod = importlib.import_module('.'.join(f.split('.')[:-1]))
mod = importlib.import_module(mod_name)
if hasattr(mod, '__routes__'):
routes.extend(mod.__routes__)
except Exception as e:
@ -152,4 +169,12 @@ def get_remote_base_url():
host=get_ip_or_hostname(), port=get_http_port())
def authenticate_user(route):
@wraps(route)
def authenticated_route(*args, **kwargs):
if not authentication_ok(request): return authenticate()
return route(*args, **kwargs)
return authenticated_route
# vim:sw=4:ts=4:et:

View file

@ -14,7 +14,14 @@ from platypush.utils import get_event_class_by_type
class Event(Message):
""" Event message class """
def __init__(self, target=None, origin=None, id=None, timestamp=None, **kwargs):
# If this class property is set to false then the logging of these events
# will be disabled. Logging is usually disabled for events with a very
# high frequency that would otherwise pollute the logs e.g. camera capture
# events
disable_logging = False
def __init__(self, target=None, origin=None, id=None, timestamp=None,
disable_logging=disable_logging, **kwargs):
"""
Params:
target -- Target node [String]
@ -30,6 +37,7 @@ class Event(Message):
self.type = '{}.{}'.format(self.__class__.__module__,
self.__class__.__name__)
self.args = kwargs
self.disable_logging = disable_logging
@classmethod
def build(cls, msg):

View file

@ -44,4 +44,18 @@ class CameraPictureTakenEvent(CameraEvent):
super().__init__(*args, filename=filename, **kwargs)
class CameraFrameCapturedEvent(CameraEvent):
"""
Event triggered when a camera frame has been captured
"""
disable_logging = True
def __init__(self, filename=None, *args, **kwargs):
super().__init__(*args, filename=filename,
disable_logging=kwargs.pop('disable_logging')
if 'disable_logging' in kwargs else self.disable_logging,
**kwargs)
# vim:sw=4:ts=4:et:

View file

@ -210,7 +210,7 @@ class Request(Message):
logger.warning(('Response processed with errors from ' +
'action {}: {}').format(
self.action, str(response)))
else:
elif not response.disable_logging:
logger.info('Processed response from action {}: {}'.
format(self.action, str(response)))
except Exception as e:

View file

@ -7,7 +7,7 @@ class Response(Message):
""" Response message class """
def __init__(self, target=None, origin=None, id=None, output=None, errors=[],
timestamp=None):
timestamp=None, disable_logging=False):
"""
Params:
target -- Target [String]
@ -24,6 +24,7 @@ class Response(Message):
self.errors = self._parse_msg(errors)
self.origin = origin
self.id = id
self.disable_logging = disable_logging
def is_error(self):
""" Returns True if the respopnse has errors """
@ -50,6 +51,7 @@ class Response(Message):
}
args['timestamp'] = msg['_timestamp'] if '_timestamp' in msg else time.time()
args['disable_logging'] = msg.get('_disable_logging', False)
if 'id' in msg: args['id'] = msg['id']
if 'origin' in msg: args['origin'] = msg['origin']
return cls(**args)
@ -67,11 +69,12 @@ class Response(Message):
'target' : self.target if hasattr(self, 'target') else None,
'origin' : self.origin if hasattr(self, 'origin') else None,
'_timestamp' : self.timestamp,
'_disable_logging' : self.disable_logging,
'response' : {
'output' : self.output,
'errors' : self.errors,
},
})
# vim:sw=4:ts=4:et:
# vim:sw=4:ts=4:et:

View file

@ -14,22 +14,24 @@ from platypush.utils import get_decorators
def action(f):
@wraps(f)
def _execute_action(*args, **kwargs):
output = None
errors = []
response = Response()
result = f(*args, **kwargs)
output = f(*args, **kwargs)
if output and isinstance(output, Response):
errors = output.errors \
if isinstance(output.errors, list) else [output.errors]
output = output.output
elif isinstance(output, tuple) and len(output) == 2:
errors = output[1] \
if isinstance(output[1], list) else [output[1]]
if result and isinstance(result, Response):
result.errors = result.errors \
if isinstance(result.errors, list) else [result.errors]
response = result
elif isinstance(result, tuple) and len(result) == 2:
response.errors = result[1] \
if isinstance(result[1], list) else [result[1]]
if len(errors) == 1 and errors[0] is None: errors = []
output = output[0]
if len(response.errors) == 1 and response.errors[0] is None:
response.errors = []
response.output = result[0]
else:
response = Response(output=result, errors=[])
return Response(output=output, errors=errors)
return response
# Propagate the docstring
_execute_action.__doc__ = f.__doc__

View file

@ -9,9 +9,10 @@ import cv2
from datetime import datetime
from platypush.config import Config
from platypush.message.response import Response
from platypush.message.event.camera import CameraRecordingStartedEvent, \
CameraRecordingStoppedEvent, CameraVideoRenderedEvent, \
CameraPictureTakenEvent, CameraEvent
CameraPictureTakenEvent, CameraFrameCapturedEvent, CameraEvent
from platypush.plugins import Plugin, action
@ -44,12 +45,15 @@ class CameraPlugin(Plugin):
_default_color_transform = 'COLOR_BGR2BGRA'
_max_stored_frames = 100
_frame_filename_regex = re.compile('(\d+)-(\d+)-(\d+)_(\d+)-(\d+)-(\d+)-(\d+).jpe?g$')
def __init__(self, device_id=0, frames_dir=_default_frames_dir,
warmup_frames=_default_warmup_frames, video_type=0,
sleep_between_frames=_default_sleep_between_frames,
max_stored_frames=_max_stored_frames,
color_transform=_default_color_transform, *args, **kwargs):
color_transform=_default_color_transform,
scale_x=None, scale_y=None, rotate=None, flip=None,
*args, **kwargs):
"""
:param device_id: Index of the default video device to be used for
capturing (default: 0)
@ -88,6 +92,27 @@ class CameraPlugin(Plugin):
for a full list of supported color transformations.
(default: "``COLOR_BGR2BGRA``")
:type color_transform: str
:param scale_x: If set, the images will be scaled along the x axis by the
specified factor
:type scale_x: float
:param scale_y: If set, the images will be scaled along the y axis by the
specified factor
:type scale_y: float
:param rotate: If set, the images will be rotated by the specified
number of degrees
:type rotate: float
:param flip: If set, the images will be flipped around the specified
axis. Possible values::
- ``0`` - flip along the x axis
- ``1`` - flip along the y axis
- ``-1`` - flip along both the axis
:type flip: int
"""
super().__init__(*args, **kwargs)
@ -101,12 +126,18 @@ class CameraPlugin(Plugin):
self.sleep_between_frames = sleep_between_frames
self.max_stored_frames = max_stored_frames
self.color_transform = color_transform
self.scale_x = scale_x
self.scale_y = scale_y
self.rotate = rotate
self.flip = flip
self._is_recording = {} # device_id => Event map
self._devices = {} # device_id => VideoCapture map
self._recording_threads = {} # device_id => Thread map
self._recording_info = {} # device_id => recording info map
def _init_device(self, device_id):
def _init_device(self, device_id, frames_dir=None, **info):
self._release_device(device_id)
if device_id not in self._devices:
@ -115,6 +146,12 @@ class CameraPlugin(Plugin):
if device_id not in self._is_recording:
self._is_recording[device_id] = threading.Event()
self._recording_info[device_id] = info
if frames_dir:
os.makedirs(frames_dir, exist_ok=True)
self._recording_info[device_id]['frames_dir'] = frames_dir
return self._devices[device_id]
@ -133,12 +170,14 @@ class CameraPlugin(Plugin):
del self._devices[device_id]
self.fire_event(CameraRecordingStoppedEvent(device_id=device_id))
if device_id in self._recording_info:
del self._recording_info[device_id]
def _store_frame_to_file(self, frame, frames_dir, image_file):
if image_file:
filepath = image_file
else:
os.makedirs(frames_dir, exist_ok=True)
filepath = os.path.join(
frames_dir, datetime.now().strftime('%Y-%m-%d_%H-%M-%S-%f.jpg'))
@ -147,21 +186,22 @@ class CameraPlugin(Plugin):
def _get_stored_frames_files(self, frames_dir):
return sorted([
ret = sorted([
os.path.join(frames_dir, f) for f in os.listdir(frames_dir)
if os.path.isfile(os.path.join(frames_dir, f)) and f.endswith('.jpg')
if os.path.isfile(os.path.join(frames_dir, f)) and
re.search(self._frame_filename_regex, f)
])
return ret
def _get_avg_fps(self, frames_dir):
files = self._get_stored_frames_files(frames_dir)
regex = re.compile('(\d+)-(\d+)-(\d+)_(\d+)-(\d+)-(\d+)-(\d+).jpe?g$')
frame_time_diff = 0.0
n_frames = 0
for i in range(1, len(files)):
m1 = re.search(regex, files[i-1])
m2 = re.search(regex, files[i])
m1 = re.search(self._frame_filename_regex, files[i-1])
m2 = re.search(self._frame_filename_regex, files[i])
if not m1 or not m2:
continue
@ -176,7 +216,7 @@ class CameraPlugin(Plugin):
def _remove_expired_frames(self, frames_dir, max_stored_frames):
files = self._get_stored_frames_files(frames_dir)
for f in files[max_stored_frames+1:len(files)]:
for f in files[:len(files)-max_stored_frames]:
os.unlink(f)
@ -199,13 +239,15 @@ class CameraPlugin(Plugin):
shutil.rmtree(frames_dir, ignore_errors=True)
def _recording_thread(self, duration, video_file, image_file, device_id,
def _recording_thread(self):
def thread(duration, video_file, image_file, device_id,
frames_dir, n_frames, sleep_between_frames,
max_stored_frames, color_transform, video_type):
max_stored_frames, color_transform, video_type,
scale_x, scale_y, rotate, flip):
device = self._devices[device_id]
color_transform = getattr(cv2, self.color_transform)
def thread():
rotation_matrix = None
self._is_recording[device_id].wait()
self.logger.info('Starting recording from video device {}'.
format(device_id))
@ -234,9 +276,28 @@ class CameraPlugin(Plugin):
continue
frame = cv2.cvtColor(frame, color_transform)
if rotate:
if not rotation_matrix:
rows, cols = frame.shape
rotation_matrix = cv2.getRotationMatrix2D(
(cols/2, rows/2), rotate, 1)
frame = cv2.warpAffine(frame, rotation_matrix, (cols, rows))
if flip is not None:
frame = cv2.flip(frame, flip)
if scale_x or scale_y:
scale_x = scale_x or 1
scale_y = scale_y or 1
frame = cv2.resize(frame, None, fx=scale_x, fy=scale_y,
interpolation = cv.INTER_CUBIC)
self._store_frame_to_file(frame=frame, frames_dir=frames_dir,
image_file=image_file)
captured_frames += 1
self.fire_event(CameraFrameCapturedEvent(filename=image_file))
if max_stored_frames and not video_file:
self._remove_expired_frames(
@ -269,7 +330,8 @@ class CameraPlugin(Plugin):
def start_recording(self, duration=None, video_file=None, video_type=None,
device_id=None, frames_dir=None,
sleep_between_frames=None, max_stored_frames=None,
color_transform=None):
color_transform=None, scale_x=None, scale_y=None,
rotate=None, flip=None):
"""
Start recording
@ -284,15 +346,22 @@ class CameraPlugin(Plugin):
:param video_type: Overrides the default configured ``video_type``
:type video_file: str
:param device_id, frames_dir, sleep_between_frames, max_stored_frames: Set
:param device_id, frames_dir, sleep_between_frames, max_stored_frames,
color_transform, scale_x, scale_y, rotate, flip: Set
these parameters if you want to override the default configured ones.
"""
device_id = device_id if device_id is not None else self.default_device_id
if device_id in self._is_recording and \
self._is_recording[device_id].is_set():
self.logger.info('A recording on device {} is already in progress'.
format(device_id))
return self.status(device_id=device_id)
recording_started = threading.Event()
def on_recording_started(event):
recording_started.set()
device_id = device_id if device_id is not None else self.default_device_id
frames_dir = os.path.abspath(os.path.expanduser(frames_dir)) \
if frames_dir is not None else self.frames_dir
sleep_between_frames = sleep_between_frames if sleep_between_frames \
@ -301,6 +370,10 @@ class CameraPlugin(Plugin):
is not None else self.max_stored_frames
color_transform = color_transform if color_transform \
is not None else self.color_transform
scale_x = scale_x if scale_x is not None else self.scale_x
scale_y = scale_y if scale_y is not None else self.scale_y
rotate = rotate if rotate is not None else self.rotate
flip = flip if flip is not None else self.flip
if video_type is not None:
video_type = cv2.VideoWriter_fourcc(*video_type.upper()) \
@ -308,31 +381,42 @@ class CameraPlugin(Plugin):
else:
video_type = self.video_type
self._init_device(device_id)
self.register_handler(CameraRecordingStartedEvent, on_recording_started)
frames_dir = os.path.join(frames_dir, str(device_id))
if video_file:
video_file = os.path.abspath(os.path.expanduser(video_file))
frames_dir = os.path.join(frames_dir, 'recording_{}'.format(
datetime.now().strftime('%Y-%m-%d_%H-%M-%S-%f')))
self._recording_threads[device_id] = threading.Thread(
target=self._recording_thread(duration=duration,
video_file=video_file,
self._init_device(device_id, video_file=video_file,
video_type=video_type,
image_file=None, device_id=device_id,
frames_dir=frames_dir, n_frames=None,
frames_dir=frames_dir,
sleep_between_frames=sleep_between_frames,
max_stored_frames=max_stored_frames,
color_transform=color_transform))
color_transform=color_transform, scale_x=scale_x,
scale_y=scale_y, rotate=rotate, flip=flip)
self.register_handler(CameraRecordingStartedEvent, on_recording_started)
self._recording_threads[device_id] = threading.Thread(
target=self._recording_thread(), kwargs = {
'duration':duration,
'video_file':video_file,
'video_type':video_type,
'image_file':None, 'device_id':device_id,
'frames_dir':frames_dir, 'n_frames':None,
'sleep_between_frames':sleep_between_frames,
'max_stored_frames':max_stored_frames,
'color_transform':color_transform,
'scale_x':scale_x, 'scale_y':scale_y,
'rotate':rotate, 'flip':flip
})
self._recording_threads[device_id].start()
self._is_recording[device_id].set()
recording_started.wait()
self.unregister_handler(CameraRecordingStartedEvent, on_recording_started)
return { 'path': video_file if video_file else frames_dir }
return self.status(device_id=device_id)
@action
def stop_recording(self, device_id=None):
@ -341,41 +425,72 @@ class CameraPlugin(Plugin):
"""
device_id = device_id if device_id is not None else self.default_device_id
frames_dir = self._recording_info.get(device_id, {}).get('frames_dir')
self._release_device(device_id)
shutil.rmtree(frames_dir, ignore_errors=True)
@action
def take_picture(self, image_file, device_id=None, warmup_frames=None,
color_transform=None):
color_transform=None, scale_x=None, scale_y=None,
rotate=None, flip=None):
"""
Take a picture.
:param image_file: Path where the output image will be stored.
:type image_file: str
:param device_id, warmup_frames, color_transform: Overrides the configured default parameters
:param device_id, warmup_frames, color_transform, scale_x, scale_y,
rotate, flip: Overrides the configured default parameters
"""
device_id = device_id if device_id is not None else self.default_device_id
image_file = os.path.abspath(os.path.expanduser(image_file))
picture_taken = threading.Event()
def on_picture_taken(event):
picture_taken.set()
image_file = os.path.abspath(os.path.expanduser(image_file))
device_id = device_id if device_id is not None else self.default_device_id
if device_id in self._is_recording and \
self._is_recording[device_id].is_set():
self.logger.info('A recording on device {} is already in progress'.
format(device_id))
status = self.status(device_id=device_id).output.get(device_id)
if 'image_file' in status:
shutil.copyfile(status['image_file'], image_file)
return { 'path': image_file }
raise RuntimeError('Recording already in progress and no images ' +
'have been captured yet')
warmup_frames = warmup_frames if warmup_frames is not None else \
self.warmup_frames
color_transform = color_transform if color_transform \
is not None else self.color_transform
scale_x = scale_x if scale_x is not None else self.scale_x
scale_y = scale_y if scale_y is not None else self.scale_y
rotate = rotate if rotate is not None else self.rotate
flip = flip if flip is not None else self.flip
self._init_device(device_id, image_file=image_file,
warmup_frames=warmup_frames,
color_transform=color_transform,
scale_x=scale_x, scale_y=scale_y, rotate=rotate,
flip=flip)
self._init_device(device_id)
self.register_handler(CameraPictureTakenEvent, on_picture_taken)
self._recording_threads[device_id] = threading.Thread(
target=self._recording_thread(duration=None, video_file=None,
image_file=image_file, video_type=None,
device_id=device_id, frames_dir=None,
n_frames=warmup_frames,
sleep_between_frames=None,
max_stored_frames=None,
color_transform=color_transform))
target=self._recording_thread(), kwargs = {
'duration':None, 'video_file':None,
'image_file':image_file, 'video_type':None,
'device_id':device_id, 'frames_dir':None,
'n_frames':warmup_frames,
'sleep_between_frames':None,
'max_stored_frames':None,
'color_transform':color_transform,
'scale_x':scale_x, 'scale_y':scale_y,
'rotate':rotate, 'flip':flip
})
self._recording_threads[device_id].start()
self._is_recording[device_id].set()
@ -384,5 +499,30 @@ class CameraPlugin(Plugin):
self.unregister_handler(CameraPictureTakenEvent, on_picture_taken)
return { 'path': image_file }
@action
def status(self, device_id=None):
"""
Returns the status of the specified device_id or all the device in a
``{ device_id => device_info }`` map format. Device info includes
``video_file``, ``image_file``, ``frames_dir`` and additional video info
"""
resp = Response(output={
id: {
'image_file': self._get_stored_frames_files(info['frames_dir'])[-2]
if 'frames_dir' in info
and len(self._get_stored_frames_files(info['frames_dir'])) > 1
and 'image_file' not in info else info.get('image_file'), **info
}
for id, info in self._recording_info.items()
if device_id is None or id == device_id
}, disable_logging=True)
return resp
@action
def get_default_device_id(self):
return self.default_device_id
# vim:sw=4:ts=4:et: