feat: support new remote-entity

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 _Changes in the next release_
 
+### Added
+- New remote-entity type. Requires remote-core / Core Simulator version 0.43.0 or newer.
+
 ---
 
 ## v0.1.7 - 2024-03-13

examples/remote.json

@@ -0,0 +1,18 @@
+{
+  "driver_id": "remote_test",
+  "version": "0.0.1",
+  "min_core_api": "0.20.0",
+  "name": { "en": "Remote test" },
+  "icon": "uc:integration",
+  "description": {
+    "en": "Minimal Python integration driver example with a remote entity."
+  },
+  "port": 9084,
+  "developer": {
+    "name": "Unfolded Circle ApS",
+    "email": "[email protected]",
+    "url": "https://www.unfoldedcircle.com"
+  },
+  "home_page": "https://www.unfoldedcircle.com",
+  "release_date": "2024-04-08"
+}

examples/remote.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Remote entity integration example. Bare minimum of an integration driver."""
+import asyncio
+import logging
+import sys
+from typing import Any
+
+import ucapi
+from ucapi import remote
+from ucapi.remote import *
+
+loop = asyncio.get_event_loop()
+api = ucapi.IntegrationAPI(loop)
+
+supported_commands = [
+    "VOLUME_UP",
+    "VOLUME_DOWN",
+    "HOME",
+    "GUIDE",
+    "CONTEXT_MENU",
+    "CURSOR_UP",
+    "CURSOR_DOWN",
+    "CURSOR_LEFT",
+    "CURSOR_RIGHT",
+    "CURSOR_ENTER",
+]
+
+
+async def cmd_handler(
+    entity: ucapi.Remote, cmd_id: str, params: dict[str, Any] | None
+) -> ucapi.StatusCodes:
+    """
+    Remote command handler.
+
+    Called by the integration-API if a command is sent to a configured remote-entity.
+
+    :param entity: remote entity
+    :param cmd_id: command
+    :param params: optional command parameters
+    :return: status of the command
+    """
+    print(f"Got {entity.id} command request: {cmd_id}")
+
+    state = None
+    match cmd_id:
+        case remote.Commands.ON:
+            state = remote.States.ON
+        case remote.Commands.OFF:
+            state = remote.States.OFF
+        case remote.Commands.TOGGLE:
+            if entity.attributes[remote.Attributes.STATE] == remote.States.OFF:
+                state = remote.States.ON
+            else:
+                state = remote.States.OFF
+        case remote.Commands.SEND_CMD:
+            command = params.get("command")
+            if command not in supported_commands:
+                print(f"Unknown command: {command}", file=sys.stderr)
+                return ucapi.StatusCodes.BAD_REQUEST
+
+            repeat = params.get("repeat", 1)
+            delay = params.get("delay", 0)
+            hold = params.get("hold", 0)
+            print(f"Command: {command} (repeat={repeat}, delay={delay}, hold={hold})")
+        case remote.Commands.SEND_CMD_SEQUENCE:
+            sequence = params.get("sequence")
+            repeat = params.get("repeat", 1)
+            delay = params.get("delay", 0)
+            hold = params.get("hold", 0)
+            print(
+                f"Command sequence: {sequence} (repeat={repeat}, delay={delay}, hold={hold})"
+            )
+        case _:
+            print(f"Unsupported command: {cmd_id}", file=sys.stderr)
+            return ucapi.StatusCodes.BAD_REQUEST
+
+    if state:
+        api.configured_entities.update_attributes(
+            entity.id, {remote.Attributes.STATE: state}
+        )
+
+    return ucapi.StatusCodes.OK
+
+
+@api.listens_to(ucapi.Events.CONNECT)
+async def on_connect() -> None:
+    # When the remote connects, we just set the device state. We are ready all the time!
+    await api.set_device_state(ucapi.DeviceStates.CONNECTED)
+
+
+def create_button_mappings() -> list[DeviceButtonMapping]:
+    return [
+        # simple short- and long-press mapping
+        create_btn_mapping(Buttons.HOME, "HOME", "GUIDE"),
+        # use channel buttons for volume control
+        create_btn_mapping(Buttons.CHANNEL_DOWN, "VOLUME_DOWN"),
+        create_btn_mapping(Buttons.CHANNEL_UP, "VOLUME_UP"),
+        create_btn_mapping(Buttons.DPAD_UP, "CURSOR_UP"),
+        create_btn_mapping(Buttons.DPAD_DOWN, "CURSOR_DOWN"),
+        create_btn_mapping(Buttons.DPAD_LEFT, "CURSOR_LEFT"),
+        create_btn_mapping(Buttons.DPAD_RIGHT, "CURSOR_RIGHT"),
+        # use a send command
+        create_btn_mapping(
+            Buttons.DPAD_MIDDLE, create_send_cmd("CONTEXT_MENU", hold=1000)
+        ),
+        # use a sequence command
+        create_btn_mapping(
+            Buttons.BLUE,
+            create_sequence_cmd(
+                [
+                    "CURSOR_UP",
+                    "CURSOR_RIGHT",
+                    "CURSOR_DOWN",
+                    "CURSOR_LEFT",
+                ],
+                delay=200,
+            ),
+        ),
+    ]
+
+
+def create_ui() -> list[UiPage]:
+    ui_page1 = UiPage("page1", "Main")
+    ui_page1.add(create_ui_text("Hello remote entity", 0, 0, size=Size(4, 1)))
+    ui_page1.add(create_ui_icon("uc:home", 0, 2, cmd="HOME"))
+    ui_page1.add(create_ui_icon("uc:up-arrow-bold", 2, 2, cmd="CURSOR_UP"))
+    ui_page1.add(create_ui_icon("uc:down-arrow-bold", 2, 4, cmd="CURSOR_DOWN"))
+    ui_page1.add(create_ui_icon("uc:left-arrow", 1, 3, cmd="CURSOR_LEFT"))
+    ui_page1.add(create_ui_icon("uc:right-arrow", 3, 3, cmd="CURSOR_RIGHT"))
+    ui_page1.add(create_ui_text("Ok", 2, 3, cmd="CURSOR_ENTER"))
+
+    ui_page2 = UiPage("page2", "Page 2")
+    ui_page2.add(
+        create_ui_text(
+            "Pump up the volume!",
+            0,
+            0,
+            size=Size(4, 2),
+            cmd=create_send_cmd("VOLUME_UP", repeat=5),
+        )
+    )
+    ui_page2.add(
+        create_ui_text(
+            "Test sequence",
+            0,
+            4,
+            size=Size(4, 1),
+            cmd=create_sequence_cmd(
+                [
+                    "CURSOR_UP",
+                    "CURSOR_RIGHT",
+                    "CURSOR_DOWN",
+                    "CURSOR_LEFT",
+                ],
+                delay=200,
+            ),
+        )
+    )
+    ui_page2.add(create_ui_text("On", 0, 5, cmd="on"))
+    ui_page2.add(create_ui_text("Off", 1, 5, cmd="off"))
+
+    return [ui_page1, ui_page2]
+
+
+if __name__ == "__main__":
+    logging.basicConfig()
+
+    entity = ucapi.Remote(
+        "remote1",
+        "Demo remote",
+        [remote.Features.ON_OFF, remote.Features.TOGGLE],
+        {remote.Attributes.STATE: remote.States.OFF},
+        simple_commands=supported_commands,
+        button_mapping=create_button_mappings(),
+        ui_pages=create_ui(),
+        cmd_handler=cmd_handler,
+    )
+    api.available_entities.add(entity)
+
+    loop.run_until_complete(api.init("remote.json"))
+    loop.run_forever()

ucapi/__init__.py

@@ -35,6 +35,7 @@
 from .cover import Cover  # noqa: F401
 from .light import Light  # noqa: F401
 from .media_player import MediaPlayer  # noqa: F401
+from .remote import Remote  # noqa: F401
 from .sensor import Sensor  # noqa: F401
 from .switch import Switch  # noqa: F401
 

ucapi/api.py

@@ -6,6 +6,7 @@
 """
 
 import asyncio
+import dataclasses
 import json
 import logging
 import os
@@ -32,6 +33,15 @@
 _LOG.setLevel(logging.DEBUG)
 
 
+class _EnhancedJSONEncoder(json.JSONEncoder):
+    """Python dataclass json encoder."""
+
+    def default(self, o):
+        if dataclasses.is_dataclass(o):
+            return dataclasses.asdict(o)
+        return super().default(o)
+
+
 class IntegrationAPI:
     """Integration API to communicate with Remote Two."""
 
@@ -239,7 +249,7 @@ async def _send_ws_response(
         }
 
         if websocket in self._clients:
-            data_dump = json.dumps(data)
+            data_dump = json.dumps(data, cls=_EnhancedJSONEncoder)
             _LOG.debug("[%s] ->: %s", websocket.remote_address, data_dump)
             await websocket.send(data_dump)
         else:
@@ -259,10 +269,14 @@ async def _broadcast_ws_event(
         :param category: event category
         """
         data = {"kind": "event", "msg": msg, "msg_data": msg_data, "cat": category}
-        data_dump = json.dumps(data)
+        data_dump = json.dumps(data, cls=_EnhancedJSONEncoder)
         # filter fields
         if _LOG.isEnabledFor(logging.DEBUG):
-            data_log = json.dumps(data) if filter_log_msg_data(data) else data_dump
+            data_log = (
+                json.dumps(data, cls=_EnhancedJSONEncoder)
+                if filter_log_msg_data(data)
+                else data_dump
+            )
 
         for websocket in self._clients:
             if _LOG.isEnabledFor(logging.DEBUG):
@@ -287,11 +301,15 @@ async def _send_ws_event(
             websockets.ConnectionClosed: When the connection is closed.
         """
         data = {"kind": "event", "msg": msg, "msg_data": msg_data, "cat": category}
-        data_dump = json.dumps(data)
+        data_dump = json.dumps(data, cls=_EnhancedJSONEncoder)
 
         if websocket in self._clients:
             if _LOG.isEnabledFor(logging.DEBUG):
-                data_log = json.dumps(data) if filter_log_msg_data(data) else data_dump
+                data_log = (
+                    json.dumps(data, cls=_EnhancedJSONEncoder)
+                    if filter_log_msg_data(data)
+                    else data_dump
+                )
                 _LOG.debug("[%s] ->: %s", websocket.remote_address, data_log)
             await websocket.send(data_dump)
         else:

ucapi/button.py

@@ -59,8 +59,6 @@ def __init__(
             EntityTypes.BUTTON,
             ["press"],
             {Attributes.STATE: States.AVAILABLE},
-            None,
-            None,
-            area,
-            cmd_handler,
+            area=area,
+            cmd_handler=cmd_handler,
         )

ucapi/entity.py

@@ -23,6 +23,7 @@ class EntityTypes(str, Enum):
     CLIMATE = "climate"
     LIGHT = "light"
     MEDIA_PLAYER = "media_player"
+    REMOTE = "remote"
     SENSOR = "sensor"
     SWITCH = "switch"
 
@@ -42,8 +43,8 @@ def __init__(
         entity_type: EntityTypes,
         features: list[str],
         attributes: dict[str, Any],
-        device_class: str | None,
-        options: dict[str, Any] | None,
+        device_class: str | None = None,
+        options: dict[str, Any] | None = None,
         area: str | None = None,
         cmd_handler: CommandHandler = None,
     ):