[#398] Replaced `qrcode` response objects with schemas.

docs/source/platypush/responses/qrcode.rst

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

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

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)
 
 

platypush/message/response/qrcode.py

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

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://<host>:<port>/qrcode?content=...``.
+
+        If you configured the :class:`platypush.backend.http.HttpBackend` then
+        you can also generate codes directly from the browser through
+        ``http://<host>:<port>/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()
 

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

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',
+        },
+    )

platypush/utils/mock/modules.py

@@ -97,6 +97,7 @@ mock_imports = [
     "pyotp",
     "pysmartthings",
     "pyzbar",
+    "qrcode",
     "rtmidi",
     "samsungtvws",
     "serial",