driver.py

1# OpenAss: nur privater, experimenteller Gebrauch; auf eigene Verantwortung.2# Niemals Unternehmensrechner, Produktivsysteme, Industrieanlagen oder3# sicherheitskritische Systeme anschließen oder bedienen.4# Haftungshinweis und gesetzliche Grenzen: ../../../DISCLAIMER.txt.5"""Fail-closed serial HID driver and a reviewable, output-free dry-run driver.67Pico UART0 uses GP0/GP1; Pi UART uses GPIO14/15 (header pins 8/10), at 1152008baud with 3.3 V logic and shared ground. The Pico's9USB goes only to the target; it is not the serial control channel. ARM also10requires the firmware's physical local grant. No action is automatically11retried after an uncertain acknowledgement. A failed link ceases PING so the12firmware's 750 ms lease releases all keys/buttons. Recovery is an explicit13start() followed by a new physical grant and arm().14"""1516import asyncio17from collections.abc import Callable, Mapping18import logging19import threading20import time21from typing import Any2223from .keyboard import KeyboardLayout, KeyStroke24from .mapping import Geometry25from .protocol import MAX_FRAME_BYTES, MAX_SEQUENCE, ProtocolError, decode_frame, encode_frame, require_ok2627_LOG = logging.getLogger(__name__)28_BUTTONS = {"left": 1, "right": 2, "middle": 4}29_POINTER_TYPES = {"move", "click", "double_click", "drag", "scroll"}30_ARMED_COMMANDS = {"ABS", "REL", "KEY", "RELEASE", "PING"}313233class HIDError(RuntimeError):34    pass353637def _int(value: object, name: str, minimum: int, maximum: int) -> int:38    if type(value) is not int or not minimum <= value <= maximum:39        raise ValueError(f"{name} must be an integer in {minimum}..{maximum}")40    return value414243def validate_action(action: Mapping[str, Any], geometry: Geometry | None, keyboard_layout: str = "us") -> dict[str, Any]:44    """Pure full-action preflight; call before approvals and before any output.4546    Policy on sensitive operations belongs to the safety controller. These47    limits additionally bound individual HID operations and reject unsupported48    text even if its valid prefix could otherwise have already been typed.49    """50    if not isinstance(action, Mapping) or not isinstance(action.get("type"), str):51        raise ValueError("action must contain a string type")52    kind = action["type"]53    if kind == "doubleclick":54        kind = "double_click"55    result: dict[str, Any] = {"type": kind}56    keys_by_type = {57        "move": {"x", "y"}, "click": {"x", "y", "button"},58        "double_click": {"x", "y", "button"},59        "drag": {"path", "x1", "y1", "x2", "y2"},60        "scroll": {"x", "y", "scroll_x", "scroll_y"},61        "type": {"text"}, "keypress": {"keys", "key"},62        "wait": {"milliseconds"}, "screenshot": set(),63    }64    if kind not in keys_by_type:65        raise ValueError(f"unsupported action: {kind}")66    if set(action) - ({"type"} | keys_by_type[kind]):67        raise ValueError("unknown action fields")68    if kind in _POINTER_TYPES:69        if not isinstance(geometry, Geometry):70            raise ValueError("pointer actions require calibrated capture geometry")7172    def point(x: object, y: object) -> dict[str, int]:73        # Geometry also rejects non-int, non-finite, out-of-image and bar points.74        assert geometry is not None75        geometry.to_hid(x, y)76        return {"x": x, "y": y}  # type: ignore[dict-item]7778    if kind in {"move", "click", "double_click", "scroll"}:79        result.update(point(action.get("x"), action.get("y")))80    if kind in {"click", "double_click"}:81        button = action.get("button", "left")82        if not isinstance(button, str) or button not in _BUTTONS:83            raise ValueError("button must be left, right or middle")84        result["button"] = button85    if kind == "drag":86        if "path" in action and any(k in action for k in ("x1", "y1", "x2", "y2")):87            raise ValueError("drag must use either path or endpoint coordinates")88        path = action.get("path")89        if path is None:90            path = [point(action.get("x1"), action.get("y1")), point(action.get("x2"), action.get("y2"))]91        if not isinstance(path, (list, tuple)) or not 2 <= len(path) <= 256:92            raise ValueError("drag path must have 2..256 points")93        points = []94        for item in path:95            if not isinstance(item, Mapping) or set(item) != {"x", "y"}:96                raise ValueError("drag points must contain only x and y")97            points.append(point(item["x"], item["y"]))98        result["path"] = points99    if kind == "scroll":100        result["scroll_x"] = _int(action.get("scroll_x", 0), "scroll_x", -12000, 12000)101        result["scroll_y"] = _int(action.get("scroll_y", 0), "scroll_y", -12000, 12000)102    layout = KeyboardLayout(keyboard_layout)103    if kind == "type":104        text = action.get("text")105        if not isinstance(text, str) or len(text) > 4096:106            raise ValueError("text must contain at most 4096 characters")107        layout.preflight_text(text)108        result["text"] = text109    if kind == "keypress":110        if "key" in action and "keys" in action:111            raise ValueError("keypress must use either key or keys")112        keys = action.get("keys", action.get("key"))113        layout.keypress(keys)114        result["keys"] = keys if isinstance(keys, str) else list(keys)115    if kind == "wait":116        result["milliseconds"] = _int(action.get("milliseconds", 1000), "milliseconds", 0, 5000)117    return result118119120class _SerialLink:121    """One outstanding request at a time, including heartbeat and emergency stop."""122123    def __init__(self, serial: Any, acknowledgement_timeout: float) -> None:124        self.serial = serial125        self.timeout = acknowledgement_timeout126        self.lock = threading.Lock()127        self.sequence = 0128129    def request(self, operation: str, arguments: tuple[int, ...] = ()) -> tuple[str, ...]:130        with self.lock:131            self.sequence = self.sequence % MAX_SEQUENCE + 1132            sequence = self.sequence133            message = encode_frame(sequence, operation, arguments)134            deadline = time.monotonic() + self.timeout135            try:136                written = self.serial.write(message)137                if written != len(message):138                    raise ProtocolError("incomplete serial write")139                # No flush(): some platforms can block indefinitely draining a140                # disconnected UART. write_timeout already bounds submission.141                remaining = deadline - time.monotonic()142                if remaining <= 0:143                    raise ProtocolError("acknowledgement timed out")144                self.serial.timeout = remaining145                raw = self.serial.read_until(b"\n", MAX_FRAME_BYTES + 1)146                if time.monotonic() > deadline:147                    raise ProtocolError("acknowledgement timed out")148                return require_ok(decode_frame(raw), sequence)149            except ProtocolError:150                raise151            except Exception as exc:152                raise ProtocolError(f"serial request failed: {type(exc).__name__}") from exc153154155class _Actions:156    """Common absolute-pointer executor; relative positioning is not exposed."""157158    keyboard: KeyboardLayout159    _armed: bool160    _fault: str | None161    _scroll_residual_x: int162    _scroll_residual_y: int163164    def preflight(self, action: Mapping[str, Any], geometry: Geometry | None = None) -> dict[str, Any]:165        return validate_action(action, geometry, self.keyboard.name)166167    async def _request(self, operation: str, arguments: tuple[int, ...] = (), *, expected_generation: int | None = None) -> tuple[str, ...]:168        raise NotImplementedError169170    async def _report(self, operation: str, arguments: tuple[int, ...] = ()) -> None:171        if not self._armed or self._fault or self._execution_generation != self._stop_generation:172            raise HIDError("HID is disarmed, faulted or this action was stopped")173        await self._request(operation, arguments, expected_generation=self._execution_generation)174175    async def _release(self) -> None:176        # Faults disarm and STOP handles release; no action may resume after it.177        if self._armed and not self._fault and self._execution_generation == self._stop_generation:178            await self._report("RELEASE")179180    async def _stroke(self, stroke: KeyStroke) -> None:181        try:182            await self._report("KEY", stroke.report_arguments())183            await asyncio.sleep(0.015)184        finally:185            await self._release()186187    async def execute(self, action: Mapping[str, Any], geometry: Geometry | None = None) -> None:188        canonical = self.preflight(action, geometry)189        submitted_generation = self._stop_generation190        async with self._action_lock:191            if not self._armed or self._fault or submitted_generation != self._stop_generation:192                raise HIDError("HID is disarmed, faulted or the queued action was stopped")193            self._execution_generation = self._stop_generation194            kind = canonical["type"]195            if kind in {"wait", "screenshot"}:196                if kind == "wait":197                    await asyncio.sleep(canonical["milliseconds"] / 1000)198                return199            if kind in {"move", "click", "double_click", "scroll"}:200                assert geometry is not None201                hx, hy = geometry.to_hid(canonical["x"], canonical["y"])202                await self._report("ABS", (hx, hy, 0))203            if kind == "move":204                return205            if kind in {"click", "double_click"}:206                for index in range(2 if kind == "double_click" else 1):207                    if index:208                        await asyncio.sleep(0.12)209                    try:210                        await self._report("ABS", (hx, hy, _BUTTONS[canonical["button"]]))211                        await asyncio.sleep(0.02)212                    finally:213                        await self._release()214            elif kind == "drag":215                assert geometry is not None216                points = [geometry.to_hid(p["x"], p["y"]) for p in canonical["path"]]217                await self._report("ABS", (*points[0], 0))218                try:219                    for point in points:220                        await self._report("ABS", (*point, 1))221                        await asyncio.sleep(0.015)222                finally:223                    await self._release()224            elif kind == "scroll":225                # Computer Use scroll units are pixels; HID wheel is detents.226                # Preserve small remainders instead of converting 1 px to 1227                # notch. 120 px/notch is a conservative configurable-by-code228                # baseline, not an OS-independent pixel-distance guarantee.229                sx = self._scroll_residual_x + canonical["scroll_x"]230                sy = self._scroll_residual_y + canonical["scroll_y"]231                pan = abs(sx) // 120 * (1 if sx >= 0 else -1)232                down = abs(sy) // 120 * (1 if sy >= 0 else -1)233                self._scroll_residual_x, self._scroll_residual_y = sx - pan * 120, sy - down * 120234                if pan or down:235                    await self._report("REL", (0, 0, -down, pan, 0))236            elif kind == "type":237                for stroke in self.keyboard.preflight_text(canonical["text"]):238                    await self._stroke(stroke)239            elif kind == "keypress":240                await self._stroke(self.keyboard.keypress(canonical["keys"]))241242243class RealHID(_Actions):244    def __init__(self, serial_port: str = "/dev/serial0", keyboard_layout: str = "us", baudrate: int = 115200,245                 acknowledgement_timeout: float = 0.35, *, serial_factory: Callable[..., Any] | None = None) -> None:246        if not isinstance(serial_port, str) or not serial_port:247            raise ValueError("serial_port is required")248        if baudrate != 115200:249            raise ValueError("HB1 firmware requires 115200 baud")250        if type(acknowledgement_timeout) not in {int, float} or not 0 < acknowledgement_timeout <= 0.35:251            raise ValueError("acknowledgement_timeout must be in (0, 0.35]")252        self.serial_port, self.baudrate = serial_port, baudrate253        self.acknowledgement_timeout = acknowledgement_timeout254        self.keyboard = KeyboardLayout(keyboard_layout)255        self._serial_factory = serial_factory256        self._link: _SerialLink | None = None257        self._armed = False258        self._fault: str | None = None259        self._heartbeat: asyncio.Task | None = None260        self._action_lock = asyncio.Lock()261        self._lifecycle_lock = asyncio.Lock()262        self._wire_lock = asyncio.Lock()263        self._stop_generation = 0264        self._execution_generation = 0265        self._scroll_residual_x = self._scroll_residual_y = 0266        self._last_ack: float | None = None267268    def health_snapshot(self) -> dict[str, Any]:269        return {"connected": self._link is not None, "armed": self._armed, "fault": self._fault,270                "keyboard_layout": self.keyboard.name, "last_ack_monotonic": self._last_ack,271                "dry_run": False, "lock_state_warning": "Target CapsLock and NumLock must be known off"}272273    async def _cancel_heartbeat(self) -> None:274        task, self._heartbeat = self._heartbeat, None275        if task and task is not asyncio.current_task():276            task.cancel()277            try:278                await task279            except asyncio.CancelledError:280                pass281282    async def _exchange(self, operation: str, arguments: tuple[int, ...] = (), *, expected_generation: int | None = None) -> tuple[str, ...]:283        async with self._wire_lock:284            if self._link is None:285                raise HIDError("serial link is not started")286            if operation in _ARMED_COMMANDS and not self._armed:287                raise HIDError("output was stopped while request was waiting")288            if expected_generation is not None and expected_generation != self._stop_generation:289                raise HIDError("this action was stopped while request was waiting")290            task = asyncio.create_task(asyncio.to_thread(self._link.request, operation, arguments))291            try:292                result = await asyncio.shield(task)293                self._last_ack = time.monotonic()294                return result295            except asyncio.CancelledError:296                # A cancelled coroutine does not stop its serial thread. Settle297                # the bounded request before another caller can use the link.298                try:299                    await asyncio.shield(task)300                except Exception:301                    pass302                raise303304    async def _best_effort_stop(self) -> None:305        if self._link:306            try:307                await self._exchange("STOP")308            except (Exception, asyncio.CancelledError) as exc:309                if self._fault is None:310                    self._fault = f"STOP: {type(exc).__name__}: {exc}"311312    async def _request(self, operation: str, arguments: tuple[int, ...] = (), *, expected_generation: int | None = None) -> tuple[str, ...]:313        try:314            return await self._exchange(operation, arguments, expected_generation=expected_generation)315        except (Exception, asyncio.CancelledError) as exc:316            if operation in _ARMED_COMMANDS and (not self._armed or (expected_generation is not None and expected_generation != self._stop_generation)):317                # stop() already owns disarm/release; intentional heartbeat318                # cancellation must not turn a clean manual stop into a fault.319                raise320            self._armed = False321            self._fault = f"{operation}: {type(exc).__name__}: {exc}"322            # Clear the lease-renewal condition before awaiting anything.323            await self._cancel_heartbeat()324            await self._best_effort_stop()325            if isinstance(exc, asyncio.CancelledError):326                raise327            raise HIDError(self._fault) from exc328329    async def start(self) -> None:330        """Explicit reconnect/reset. This always leaves output disarmed."""331        async with self._lifecycle_lock:332            if self._armed:333                raise HIDError("stop the armed driver before reconnecting")334            self._stop_generation += 1335            await self._cancel_heartbeat()336            if self._link:337                old_link, self._link = self._link, None338                await asyncio.to_thread(old_link.serial.close)339            self._fault = None340            factory = self._serial_factory341            if factory is None:342                import serial343                factory = serial.Serial344            try:345                open_task = asyncio.create_task(asyncio.to_thread(346                    factory, self.serial_port, self.baudrate,347                    timeout=self.acknowledgement_timeout, write_timeout=self.acknowledgement_timeout))348                try:349                    port = await asyncio.shield(open_task)350                except asyncio.CancelledError:351                    # An OS serial open still running in a thread must not leak352                    # its eventual port after the caller cancels startup.353                    port = await asyncio.shield(open_task)354                    await asyncio.to_thread(port.close)355                    raise356                self._link = _SerialLink(port, self.acknowledgement_timeout)357                reply = await self._request("HELLO")358                if reply != ("1",):359                    raise HIDError("unsupported firmware protocol version")360            except (Exception, asyncio.CancelledError) as exc:361                self._armed = False362                self._fault = f"start: {type(exc).__name__}: {exc}"363                if self._link:364                    link, self._link = self._link, None365                    await asyncio.to_thread(link.serial.close)366                raise367            self._scroll_residual_x = self._scroll_residual_y = 0368            _LOG.warning("HID target layout %s: CapsLock/NumLock must be known off", self.keyboard.name)369370    async def arm(self) -> None:371        async with self._lifecycle_lock:372            if self._link is None or self._fault:373                raise HIDError("start/restart the driver before arming")374            if self._armed:375                return376            generation = self._stop_generation377            await self._request("ARM")378            if generation != self._stop_generation:379                await self._best_effort_stop()380                raise HIDError("arm was cancelled by stop")381            self._armed = True382            self._heartbeat = asyncio.create_task(self._heartbeat_loop(), name="hid-lease-heartbeat")383384    async def _heartbeat_loop(self) -> None:385        try:386            while self._armed and not self._fault:387                await self._request("PING")388                await asyncio.sleep(0.1)389        except (HIDError, asyncio.CancelledError):390            return391392    async def stop(self) -> None:393        # Do not wait for the action lock: disarm ongoing typing/drag immediately.394        self._stop_generation += 1395        self._armed = False396        self._scroll_residual_x = self._scroll_residual_y = 0397        await self._cancel_heartbeat()398        async with self._lifecycle_lock:399            if self._link:400                await self._best_effort_stop()401402    async def close(self) -> None:403        await self.stop()404        async with self._lifecycle_lock:405            # Another caller may have won the lock to arm after stop released406            # it. Closing is authoritative, so disarm again under this lock.407            self._stop_generation += 1408            self._armed = False409            await self._cancel_heartbeat()410            await self._best_effort_stop()411            if self._link:412                link, self._link = self._link, None413                await asyncio.to_thread(link.serial.close)414415416class DryHID(_Actions):417    """Records primitive commands, produces no serial or target HID output."""418419    def __init__(self, serial_port: str = "dry", keyboard_layout: str = "us", baudrate: int = 115200,420                 acknowledgement_timeout: float = 0.35, **_: Any) -> None:421        self.keyboard = KeyboardLayout(keyboard_layout)422        self.commands: list[tuple[str, tuple[int, ...]]] = []423        self._armed = False424        self._connected = False425        self._fault: str | None = None426        self._action_lock = asyncio.Lock()427        self._stop_generation = 0428        self._execution_generation = 0429        self._scroll_residual_x = self._scroll_residual_y = 0430431    def health_snapshot(self) -> dict[str, Any]:432        return {"connected": self._connected, "armed": self._armed, "fault": self._fault,433                "keyboard_layout": self.keyboard.name, "dry_run": True,434                "lock_state_warning": "Target CapsLock and NumLock must be known off"}435436    async def _request(self, operation: str, arguments: tuple[int, ...] = (), *, expected_generation: int | None = None) -> tuple[str, ...]:437        if expected_generation is not None and expected_generation != self._stop_generation:438            raise HIDError("this action was stopped")439        self.commands.append((operation, arguments))440        return ("1",) if operation == "HELLO" else ()441442    async def start(self) -> None:443        self._stop_generation += 1444        self._connected, self._armed, self._fault = True, False, None445        self._scroll_residual_x = self._scroll_residual_y = 0446        await self._request("HELLO")447448    async def arm(self) -> None:449        if not self._connected:450            raise HIDError("start the driver before arming")451        await self._request("ARM")452        self._armed = True453454    async def stop(self) -> None:455        self._stop_generation += 1456        self._armed = False457        self._scroll_residual_x = self._scroll_residual_y = 0458        if self._connected:459            await self._request("STOP")460461    async def close(self) -> None:462        await self.stop()463        self._connected = False

Projektquellen des Referenzstands. Konfiguration und Gerätezugriff erfolgen lokal auf dem Raspberry Pi.