diff --git a/docs/source/platypush/responses/qrcode.rst b/docs/source/platypush/responses/qrcode.rst deleted file mode 100644 index 16699a69c..000000000 --- a/docs/source/platypush/responses/qrcode.rst +++ /dev/null @@ -1,5 +0,0 @@ -``qrcode`` -===================================== - -.. automodule:: platypush.message.response.qrcode - :members: diff --git a/docs/source/responses.rst b/docs/source/responses.rst index 925f56c82..211449122 100644 --- a/docs/source/responses.rst +++ b/docs/source/responses.rst @@ -8,7 +8,6 @@ Responses platypush/responses/google.drive.rst platypush/responses/printer.cups.rst - platypush/responses/qrcode.rst platypush/responses/ssh.rst platypush/responses/tensorflow.rst platypush/responses/translate.rst diff --git a/platypush/message/event/qrcode.py b/platypush/message/event/qrcode.py index 50260afff..46eb66d25 100644 --- a/platypush/message/event/qrcode.py +++ b/platypush/message/event/qrcode.py @@ -1,18 +1,19 @@ from typing import List from platypush.message.event import Event -from platypush.message.response.qrcode import ResultModel - - -class QrcodeEvent(Event): - pass class QrcodeScannedEvent(Event): """ Event triggered when a QR-code or bar code is scanned. """ - def __init__(self, results: List[ResultModel], *args, **kwargs): + + def __init__(self, results: List[dict], *args, **kwargs): + """ + :param results: List of decoded QR code results: + + .. schema:: qrcode.QrcodeDecodedResultSchema(many=True) + """ super().__init__(*args, results=results, **kwargs) diff --git a/platypush/message/response/qrcode.py b/platypush/message/response/qrcode.py deleted file mode 100644 index c31d0f47e..000000000 --- a/platypush/message/response/qrcode.py +++ /dev/null @@ -1,61 +0,0 @@ -import base64 -from typing import Optional, List - -from pyzbar.pyzbar import Decoded -from pyzbar.locations import Rect - -from platypush.message import Mapping -from platypush.message.response import Response - - -class QrcodeResponse(Response): - pass - - -class QrcodeGeneratedResponse(QrcodeResponse): - # noinspection PyShadowingBuiltins - def __init__(self, - content: str, - format: str, - data: Optional[str] = None, - image_file: Optional[str] = None, - *args, **kwargs): - super().__init__(*args, output={ - 'text': content, - 'data': data, - 'format': format, - 'image_file': image_file, - }, **kwargs) - - -class RectModel(Mapping): - def __init__(self, rect: Rect): - super().__init__() - self.left = rect.left - self.top = rect.top - self.width = rect.width - self.height = rect.height - - -class ResultModel(Mapping): - def __init__(self, result: Decoded, *args, **kwargs): - super().__init__(*args, **kwargs) - try: - data = result.data.decode() - except (ValueError, TypeError): - data = base64.encodebytes(result.data).decode() - - self.data = data - self.type = result.type - self.rect = dict(RectModel(result.rect)) if result.rect else {} - - -class QrcodeDecodedResponse(QrcodeResponse): - def __init__(self, results: List[Decoded], image_file: Optional[str] = None, *args, **kwargs): - super().__init__(*args, output={ - 'image_file': image_file, - 'results': [dict(ResultModel(result)) for result in results], - }, **kwargs) - - -# vim:sw=4:ts=4:et: diff --git a/platypush/plugins/qrcode/__init__.py b/platypush/plugins/qrcode/__init__.py index b211cb23c..eb9e75837 100644 --- a/platypush/plugins/qrcode/__init__.py +++ b/platypush/plugins/qrcode/__init__.py @@ -5,16 +5,20 @@ import threading import time from typing import Optional, List +import qrcode +from pyzbar import pyzbar +from PIL import Image + from platypush import Config from platypush.context import get_bus from platypush.message.event.qrcode import QrcodeScannedEvent -from platypush.message.response.qrcode import ( - QrcodeGeneratedResponse, - QrcodeDecodedResponse, - ResultModel, -) from platypush.plugins import Plugin, action from platypush.plugins.camera import CameraPlugin +from platypush.schemas.qrcode import ( + QrcodeDecodedSchema, + QrcodeDecodedResultSchema, + QrcodeGeneratedSchema, +) from platypush.utils import get_plugin_class_by_name @@ -25,8 +29,10 @@ class QrcodePlugin(Plugin): def __init__(self, camera_plugin: Optional[str] = None, **kwargs): """ - :param camera_plugin: Name of the plugin that will be used as a camera to capture images (e.g. - ``camera.cv`` or ``camera.pi``). + :param camera_plugin: Name of the plugin that will be used as a camera + to capture images (e.g. ``camera.cv`` or ``camera.pi``). This is + required if you want to use the ``start_scanning`` action to scan + QR codes from a camera. """ super().__init__(**kwargs) self.camera_plugin = camera_plugin @@ -36,6 +42,7 @@ class QrcodePlugin(Plugin): self, camera_plugin: Optional[str] = None, **config ) -> CameraPlugin: camera_plugin = camera_plugin or self.camera_plugin + assert camera_plugin, 'No camera plugin specified' if not config: config = Config.get(camera_plugin) or {} config['stream_raw_frames'] = True @@ -43,7 +50,7 @@ class QrcodePlugin(Plugin): cls = get_plugin_class_by_name(camera_plugin) assert cls and issubclass( cls, CameraPlugin - ), '{} is not a valid camera plugin'.format(camera_plugin) + ), f'{camera_plugin} is not a valid camera plugin' return cls(**config) @action @@ -53,23 +60,24 @@ class QrcodePlugin(Plugin): output_file: Optional[str] = None, show: bool = False, format: str = 'png', - ) -> QrcodeGeneratedResponse: + ) -> dict: """ Generate a QR code. - If you configured the :class:`platypush.backend.http.HttpBackend` then you can also generate - codes directly from the browser through ``http://:/qrcode?content=...``. + + If you configured the :class:`platypush.backend.http.HttpBackend` then + you can also generate codes directly from the browser through + ``http://:/qrcode?content=...``. :param content: Text, URL or content of the QR code. - :param output_file: If set then the QR code will be exported in the specified image file. - Otherwise, a base64-encoded representation of its binary content will be returned in - the response as ``data``. - :param show: If True, and if the device where the application runs has an active display, - then the generated QR code will be shown on display. + :param output_file: If set then the QR code will be exported in the + specified image file. Otherwise, a base64-encoded representation of + its binary content will be returned in the response as ``data``. + :param show: If True, and if the device where the application runs has + an active display, then the generated QR code will be shown on + display. :param format: Output image format (default: ``png``). - :return: :class:`platypush.message.response.qrcode.QrcodeGeneratedResponse`. + :return: .. schema:: qrcode.QrcodeGeneratedSchema """ - import qrcode - qr = qrcode.make(content) img = qr.get_image() ret = { @@ -79,6 +87,7 @@ class QrcodePlugin(Plugin): if show: img.show() + if output_file: output_file = os.path.abspath(os.path.expanduser(output_file)) img.save(output_file, format=format) @@ -88,40 +97,28 @@ class QrcodePlugin(Plugin): img.save(f, format=format) ret['data'] = base64.encodebytes(f.getvalue()).decode() - return QrcodeGeneratedResponse(**ret) + return dict(QrcodeGeneratedSchema().dump(ret)) @action - def decode(self, image_file: str) -> QrcodeDecodedResponse: + def decode(self, image_file: str) -> dict: """ Decode a QR code from an image file. :param image_file: Path of the image file. + :return: .. schema:: qrcode.QrcodeDecodedSchema """ - from pyzbar import pyzbar - from PIL import Image - image_file = os.path.abspath(os.path.expanduser(image_file)) - img = Image.open(image_file) - results = pyzbar.decode(img) - return QrcodeDecodedResponse(results) - - @staticmethod - def _convert_frame(frame): - import numpy as np - from PIL import Image - - assert isinstance( - frame, np.ndarray - ), 'Image conversion only works with numpy arrays for now (got {})'.format( - type(frame) - ) - mode = 'RGB' - if len(frame.shape) > 2 and frame.shape[2] == 4: - mode = 'RGBA' - - return Image.frombuffer( - mode, (frame.shape[1], frame.shape[0]), frame, 'raw', mode, 0, 1 - ) + with open(image_file, 'rb') as f: + img = Image.open(f) + results = pyzbar.decode(img) + return dict( + QrcodeDecodedSchema().dump( + { + 'results': results, + 'image_file': image_file, + } + ) + ) @action def start_scanning( @@ -129,29 +126,31 @@ class QrcodePlugin(Plugin): camera_plugin: Optional[str] = None, duration: Optional[float] = None, n_codes: Optional[int] = None, - ) -> Optional[List[ResultModel]]: + ) -> Optional[List[dict]]: """ Decode QR-codes and bar codes using a camera. :param camera_plugin: Camera plugin (overrides default ``camera_plugin``). - :param duration: How long the capturing phase should run (default: until ``stop_scanning`` or app termination). + :param duration: How long the capturing phase should run (default: + until ``stop_scanning`` or app termination). :param n_codes: Stop after decoding this number of codes (default: None). - :return: When ``duration`` or ``n_codes`` are specified or ``stop_scanning`` is called, it will return a list of - :class:`platypush.message.response.qrcode.ResultModel` instances with the scanned results, + :return: .. schema:: qrcode.QrcodeDecodedResultSchema(many=True) """ - from pyzbar import pyzbar - assert not self._capturing.is_set(), 'A capturing process is already running' camera = self._get_camera(camera_plugin) codes = [] last_results = {} - last_results_timeout = 10.0 + last_results_timeout = 5.0 last_results_time = 0 self._capturing.set() try: - with camera: + with camera.open( + stream=True, + frames_dir=None, + ) as session: + camera.start_camera(session) start_time = time.time() while ( @@ -159,29 +158,24 @@ class QrcodePlugin(Plugin): and (not duration or time.time() < start_time + duration) and (not n_codes or len(codes) < n_codes) ): - output = camera.get_stream() - with output.ready: - output.ready.wait() - img = self._convert_frame(output.raw_frame) - results = pyzbar.decode(img) - if results: - results = [ - result - for result in QrcodeDecodedResponse(results).output[ - 'results' - ] - if result['data'] not in last_results - or time.time() - >= last_results_time + last_results_timeout - ] + img = camera.capture_frame(session) + results = pyzbar.decode(img) - if results: - codes.extend(results) - get_bus().post(QrcodeScannedEvent(results=results)) - last_results = { - result['data']: result for result in results - } - last_results_time = time.time() + if results: + results = [ + result + for result in QrcodeDecodedResultSchema().dump( + results, many=True + ) + if result['data'] not in last_results + or time.time() >= last_results_time + last_results_timeout + ] + + if results: + codes.extend(results) + get_bus().post(QrcodeScannedEvent(results=results)) + last_results = {result['data']: result for result in results} + last_results_time = time.time() finally: self._capturing.clear() diff --git a/platypush/plugins/qrcode/manifest.yaml b/platypush/plugins/qrcode/manifest.yaml index faee6bfbf..a88230ac7 100644 --- a/platypush/plugins/qrcode/manifest.yaml +++ b/platypush/plugins/qrcode/manifest.yaml @@ -21,7 +21,7 @@ manifest: - python-numpy - python-pillow - python-qrcode - - pyzbar + # - pyzbar # Only available via yay for now pip: - numpy - qrcode diff --git a/platypush/schemas/qrcode.py b/platypush/schemas/qrcode.py new file mode 100644 index 000000000..1a4a755de --- /dev/null +++ b/platypush/schemas/qrcode.py @@ -0,0 +1,158 @@ +import base64 + +from marshmallow import EXCLUDE, fields, pre_dump +from marshmallow.schema import Schema + + +class QrcodeGeneratedSchema(Schema): + """ + Schema for a QR code generation response. + """ + + class Meta: # type: ignore + """ + Exclude unknown fields. + """ + + unknown = EXCLUDE + + text = fields.String( + required=True, + metadata={ + 'description': 'Text content of the QR code', + 'example': 'https://platypush.tech', + }, + ) + + data = fields.String( + metadata={ + 'description': 'Base64-encoded content of the QR code', + 'example': 'iVBORw0KGgoAAAANSUhEUgAAAXIAAAFyAQAAAADAX2yk', + } + ) + + format = fields.String( + metadata={ + 'description': 'Format of the QR code image', + 'example': 'png', + }, + ) + + image_file = fields.String( + metadata={ + 'description': 'Path to the generated QR code image file', + 'example': '/tmp/qr_code.png', + }, + ) + + +class QrcodeDecodedRectSchema(Schema): + """ + Schema for a single QR code decoding result rectangle. + """ + + x = fields.Integer( + required=True, + metadata={ + 'description': 'X coordinate of the rectangle in the image', + 'example': 0, + }, + ) + + y = fields.Integer( + required=True, + metadata={ + 'description': 'Y coordinate of the rectangle in the image', + 'example': 0, + }, + ) + + width = fields.Integer( + required=True, + metadata={ + 'description': 'Width of the rectangle', + 'example': 100, + }, + ) + + height = fields.Integer( + required=True, + metadata={ + 'description': 'Height of the rectangle', + 'example': 100, + }, + ) + + +class QrcodeDecodedResultSchema(Schema): + """ + Schema for a single QR code decoding result. + """ + + data = fields.String( + required=True, + metadata={ + 'description': 'Decoded QRcode data, as a base64-encoded string if binary', + 'example': 'https://platypush.tech', + }, + ) + + type = fields.String( + required=True, + metadata={ + 'description': ( + 'Type of code that was decoded. Supports the types available under the ' + '`pyzbar.ZBarSymbol` class: ' + 'https://github.com/NaturalHistoryMuseum/pyzbar/blob/master/pyzbar/wrapper.py#L43' + ), + 'example': 'QRCODE', + }, + ) + + rect = fields.Nested( + QrcodeDecodedRectSchema, + required=True, + metadata={ + 'description': 'Rectangle in the image where the QR code was found', + }, + ) + + @pre_dump + def pre_dump(self, data, **_): + if hasattr(data, '_asdict'): + data = data._asdict() + + try: + data['data'] = data['data'].decode() + except (ValueError, TypeError): + data['data'] = base64.b64encode(data['data']).decode() + + return data + + +class QrcodeDecodedSchema(Schema): + """ + Schema for a QR code decoding response. + """ + + class Meta: # type: ignore + """ + Exclude unknown fields. + """ + + unknown = EXCLUDE + + results = fields.List( + fields.Nested(QrcodeDecodedResultSchema), + required=True, + metadata={ + 'description': 'Decoded QR code results', + }, + ) + + image_file = fields.String( + metadata={ + 'description': 'Path to the image file that was decoded', + 'example': '/tmp/qr_code.png', + }, + ) diff --git a/platypush/utils/mock/modules.py b/platypush/utils/mock/modules.py index e83fd377e..5426abe11 100644 --- a/platypush/utils/mock/modules.py +++ b/platypush/utils/mock/modules.py @@ -97,6 +97,7 @@ mock_imports = [ "pyotp", "pysmartthings", "pyzbar", + "qrcode", "rtmidi", "samsungtvws", "serial",