Isolate official account RPC capabilities from API key device driver

This commit is contained in:
Pigway
2026-10-03 07:03:35 +08:00
parent 142d771ffa
commit 5aa0e9f762
15 changed files with 434 additions and 126 deletions
+8 -2
View File
@@ -4,9 +4,9 @@ from pigway_frame_flow.drivers.base import FrameDeviceDriver
from pigway_frame_flow.drivers.kokonna.client import KokonnaClient
from pigway_frame_flow.drivers.kokonna.driver import KokonnaDriver
from pigway_frame_flow.drivers.kokonna.mock import MockClient
from pigway_frame_flow.drivers.kokonna.widgets import WidgetReader
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader
from pigway_frame_flow.settings import secret
@@ -26,6 +26,7 @@ class DeviceRegistry:
}
)
self.drivers: dict[str, FrameDeviceDriver] = {}
self.official: dict[str, WidgetReader] = {}
self.keys = []
for id, entry in devices.items():
if not entry.get("enabled", True):
@@ -55,9 +56,14 @@ class DeviceRegistry:
widgets = WidgetReader(
widget_token, entry.get("widget_robot_id"), settings.mock
)
self.drivers[id] = KokonnaDriver(client, id, settings.sanitize, widgets)
self.official[id] = widgets
self.drivers[id] = KokonnaDriver(client, id, settings.sanitize)
def get(self, id: str) -> FrameDeviceDriver:
if id not in self.drivers:
raise FlowError(E.DEVICE_NOT_FOUND, "Device not found", 404)
return self.drivers[id]
def official_widgets(self, id: str):
self.get(id)
return self.official[id].read()