Plugin: cozmo (bridge through PyCozmo)

ddiff --git a/plugins/cozmo/README.md b/plugins/cozmo/README.md
nnew file mode 100644
iindex 0000000..fa4871f
---- /dev/null
++++ b/plugins/cozmo/README.md
@@@ -0,0 +1,89 @@
++# openvibe-cozmo
++
++OpenVibe Node plugin for Anki / Digital Dream Labs **Cozmo**, built on
++[PyCozmo](https://github.com/zayfod/pycozmo) (MIT). It is a *bridge* plugin: Cozmo runs closed firmware and cannot run
++the Node itself, so the Node runs on a computer next to the robot and drives it over Wi-Fi.
++
++## Bridge setup
++
++Cozmo is its own Wi-Fi access point (network `Cozmo_XXXXXX`, robot at `172.31.1.1`) and has no internet. The bridge
++computer (a Raspberry Pi or a laptop) therefore needs **two network interfaces**:
++
++| interface | joined to | used for |
++|-----------|-----------|----------|
++| `wlan0` (or the built-in Wi-Fi) | Cozmo's Wi-Fi | PyCozmo's UDP link to the robot |
++| Ethernet, or a second Wi-Fi (USB dongle) | your home network | the Node's link to OpenVibe.Bot and video |
++
++1. Put Cozmo on its charger and turn it on. Raise and lower its lift: its face shows the network name and password.
++2. Join one interface to that network. Make sure the *default route stays on the internet interface*: Cozmo's network
++   has no internet, and some systems prefer the newest Wi-Fi. With NetworkManager:
++   `nmcli connection modify Cozmo_XXXXXX ipv4.never-default yes ipv6.never-default yes`.
++3. Close the Cozmo app on any phone: the robot accepts one controller at a time.
++4. Install the plugin with the robot extra and a speech engine:
++   ```sh
++   pip install 'openvibe-cozmo[robot]'   # pulls pycozmo
++   sudo apt install espeak-ng            # or libttspico-utils (pico2wave)
++   ```
++5. Add the plugin to the Node's config (`driver: cozmo`) and start the Node. Setup waits `connect_timeout_s` for the
++   robot; if it is unreachable the plugin reports fault `not_connected`.
++
++## Firmware
++
++PyCozmo 0.8 speaks the protocol of Cozmo firmware **2381** (the last release of the Cozmo app). At connect the plugin
++reads the firmware signature. On another version, or on factory/recovery firmware (`build: FACTORY`), it stays up but
++latches fault `firmware_unsupported`: it emits an `event` with the version found and the supported ones, lists the
++fault in telemetry, and answers every command except `halt` with that nack. Update the robot with the Cozmo app. If you
++know a version works, allow it with `allow_firmware: [2381, <version>]`.
++
++## Commands
++
++| kind | value | effect |
++|------|-------|--------|
++| `drive` | `{"throttle": -1..1, "steer": -1..1}` | left = throttle + steer, right = throttle - steer (steer > 0 turns right), normalised, times `max_wheel_mmps`; stops at the deadline |
++| `ptz` | `{"tilt": -1..1}` | head angle, -1 = -25°, 1 = 44.5° (no pan: turn with `drive`) |
++| `actuator` | `{"name": "head", "value": -1..1}` | same as tilt |
++| `actuator` | `{"name": "lift", "value": 0..1}` | lift height, 0 = 32 mm (down), 1 = 92 mm (up) |
++| `actuator` | `{"name": "backpack_lights", "value": {"r","g","b"}}` | all five backpack LEDs (0..255 each; `null` = off) |
++| `actuator` | `{"name": "head_light", "value": true}` | the infrared head light |
++| `actuator` | `{"name": "cube_lights", "value": {"r","g","b"}, "cube": id?}` | every connected cube, or one object id |
++| `say` | `{"text": "..."}` | espeak-ng / pico2wave, played on Cozmo's speaker (ack = queued) |
++| `display` | `{"text"}` / `{"face": "neutral"\|"happy"\|"sad"\|"surprised"\|"sleepy"\|"angry"}` / `{"image_png_b64"}` | 128x32 1-bit image on the face screen |
++
++Cliff and pick-up: PyCozmo reports them on its own thread; the plugin stops the wheels in that callback, then stops
++every motor through the runtime and emits `cliff` / `picked_up` (and `cliff_cleared` / `picked_up_cleared`). While a
++cliff is seen, drive commands that would turn either wheel forward get nack `cliff` (backing away is allowed); while the
++robot is picked up every drive gets nack `picked_up`. Below `low_battery_v` the plugin emits `low_battery` once.
++
++Camera frames (320x240, grey unless `color: true`) are written as JPEG to `frame_dir` (a temp dir by default) through
++a temp file and `os.replace`, and announced with `{"op":"video","format":"jpeg","path":...,"seq":n}`.
++
++Telemetry: `battery` (`volts`, `percent` between 3.5 V and 4.2 V), `sensors` (`head_deg`, `lift_mm`, `cliff`,
++`picked_up`, `cubes`), `faults`.
++
++## Config
++
++| key | default | |
++|-----|---------|---|
++| `connect_timeout_s` | 10 | how long setup waits for the robot |
++| `robot_addr` | PyCozmo's (`["172.31.1.1", 5551]`) | |
++| `allow_firmware` | `[2381]` | firmware versions accepted |
++| `max_wheel_mmps` | 150 | capped at 200 (Cozmo's maximum) |
++| `wheel_accel_mmps2` | 0 | 0 = firmware default |
++| `camera` / `color` / `fps` / `jpeg_quality` / `frame_dir` | true / false / 10 / 70 / temp dir | |
++| `max_say_chars` | 200 | longer text is cut |
++| `voice`, `speech_rate`, `pico_lang` | -, 150, `en-US` | TTS options |
++| `volume` | robot's | 0..1 |
++| `low_battery_v` | 3.6 | |
++| `connect_cubes` | true | connect light cubes PyCozmo hears |
++| `procedural_face` | false | let PyCozmo animate the face (it overrides `display`) |
++
++## Tests
++
++The tests use a fake PyCozmo client (`tests/fake_pycozmo.py`) and need neither PyCozmo nor a robot:
++
++```sh
++python3 -m pytest -q plugins/cozmo/tests      # from the repository root; needs pytest and pillow
++```
++
++To try the plugin by hand against a robot, run `python3 -m openvibe_cozmo` and type JSON lines, starting with
++`{"op":"hello","config":{}}` and sending `{"op":"heartbeat"}` at least every second.
ddiff --git a/plugins/cozmo/openvibe_cozmo/__init__.py b/plugins/cozmo/openvibe_cozmo/__init__.py
nnew file mode 100644
iindex 0000000..925993a
---- /dev/null
++++ b/plugins/cozmo/openvibe_cozmo/__init__.py
@@@ -0,0 +1,508 @@
++"""Cozmo driver: a bridge plugin for Anki/Digital Dream Labs Cozmo through PyCozmo.
++
++Cozmo runs closed firmware and is its own Wi-Fi access point (no internet). The Node runs on a computer next to it with
++one network interface joined to Cozmo's Wi-Fi and another on the internet; this plugin talks to the robot with PyCozmo
++(https://github.com/zayfod/pycozmo, MIT) over the first interface.
++
++Command mapping:
++
++* ``drive`` ``{"throttle": -1..1, "steer": -1..1}`` -> ``drive_wheels(left, right)`` in mm/s, where left = throttle +
++  steer and right = throttle - steer (steer > 0 turns right), normalised so no wheel exceeds 1, times
++  ``max_wheel_mmps``. ``duration`` is never passed (PyCozmo would sleep in it); the runtime's deadline stops the wheels.
++* ``ptz`` ``{"tilt": -1..1}`` -> head angle, linear from MIN_HEAD_ANGLE (-25 deg) to MAX_HEAD_ANGLE (44.5 deg).
++* ``actuator``: ``head`` (-1..1, as tilt), ``lift`` (0..1 = 32..92 mm), ``backpack_lights`` / ``cube_lights``
++  (``{"r","g","b"}`` 0..255, or null for off), ``head_light`` (bool).
++* ``say`` ``{"text"}`` -> espeak-ng or pico2wave on a worker thread, then ``play_audio``.
++* ``display`` ``{"text"} | {"face"} | {"image_png_b64"}`` -> 128x32 1-bit image -> ``display_image``.
++
++Safety beyond the runtime's (deadline, heartbeat, EOF, estop): PyCozmo reports cliff and pick-up on its own threads;
++those callbacks stop the wheels themselves at once and then go through ``runtime.safe_stop``. While a cliff is
++detected, drive commands that would turn either wheel forward are refused (backing away is allowed); while the robot is
++picked up every drive command is refused. Firmware PyCozmo does not support latches a ``firmware_unsupported`` fault:
++setup still succeeds so the Node stays up and reports it, and every command except ``halt`` is nacked.
++
++PyCozmo is imported lazily in setup, so describe and probe work without it installed.
++"""
++
++import os
++import queue
++import shutil
++import tempfile
++import threading
++import time
++
++from openvibe_plugin import Fault, Plugin, clamp
++
++from . import render, tts
++
++__version__ = "0.1.0"
++
++# Cozmo's single-cell LiPo: about 3.5 V empty, 4.2 V full (under light load).
++BATTERY_EMPTY_V = 3.5
++BATTERY_FULL_V = 4.2
++FRAME_RING = 4  # camera frames rotate through this many files so a slow reader keeps the frame it was told about
++SAY_QUEUE = 4
++
++
++def _default_client_factory(pycozmo, config):
++    addr = config.get("robot_addr")  # e.g. ["172.31.1.1", 5551]; None -> PyCozmo's default
++    return pycozmo.Client(
++        robot_addr=tuple(addr) if addr else None,
++        auto_initialize=True,
++        # The animation controller is what streams audio and screen images to the robot, so it must run.
++        enable_animations=True,
++        # PyCozmo's procedural face would overwrite our display images; opt in with config.
++        enable_procedural_face=bool(config.get("procedural_face", False)),
++    )
++
++
++class Cozmo(Plugin):
++    driver = "cozmo"
++    version = __version__
++    motion_kinds = frozenset({"drive"})
++    telemetry_interval_s = 0.5
++
++    def __init__(self, client_factory=None, pycozmo_module=None, synthesizer=None, clock=time.monotonic,
++                 which=shutil.which):
++        """All arguments are for tests: ``client_factory(pycozmo, config)`` returns a Client-like object,
++        ``pycozmo_module`` replaces ``import pycozmo``, ``synthesizer(text, wav_path)`` replaces the TTS binary
++        lookup, ``clock`` drives the camera throttle."""
++        self.client_factory = client_factory or _default_client_factory
++        self.pc = pycozmo_module
++        self.synthesizer = synthesizer
++        self.clock = clock
++        self.which = which
++        self.cfg = {}
++        self.client = None  # set only once connected; stop() is a no-op until then
++        self.fault = None  # (code, message) latched at connect, e.g. firmware_unsupported
++        self.cliff = False
++        self.picked_up = False
++        self.low_battery = False
++        self.limits = {}
++        self.frame_dir = None
++        self._own_frame_dir = False
++        self._frame_lock = threading.Lock()
++        self._next_frame = 0.0
++        self._frame_seq = 0
++        self._say_queue = None
++        self._say_thread = None
++        self._work_dir = None
++        self._cubes_requested = set()
++        self._camera_on = False
++
++    # Description (no hardware, no pycozmo).
++    def describe(self, config):
++        synth = self.synthesizer or tts.find_synthesizer(config, self.which)
++        return {
++            "drive": {"type": "differential", "max_mmps": self._max_wheel_mmps(config, 200.0)},
++            "ptz": {"tilt": {"min": -1, "max": 1, "maps_to": "head"}},
++            "actuator": {"names": ["head", "lift", "backpack_lights", "head_light", "cube_lights"],
++                         "ranges": {"head": [-1, 1], "lift": [0, 1]}},
++            "say": {"max_chars": int(config.get("max_say_chars", 200)),
++                    "engine": getattr(synth, "engine", "custom") if synth else None},
++            "display": {"width": render.WIDTH, "height": render.HEIGHT, "text": True, "faces": list(render.FACES),
++                        "image": True},
++            "camera": {"jpeg": True, "width": 320, "height": 240,
++                       "color": bool(config.get("color", False)), "fps": self._fps(config)},
++            "battery": {},
++            "sensors": ["cliff", "picked_up", "head_deg", "lift_mm"],
++            "bridge": True,
++        }
++
++    @staticmethod
++    def _max_wheel_mmps(config, hw_max):
++        return max(0.0, min(float(config.get("max_wheel_mmps", 150.0)), hw_max))
++
++    @staticmethod
++    def _fps(config):
++        return max(0.1, min(float(config.get("fps", 10)), 30.0))
++
++    # Setup and teardown.
++    def setup(self, config):
++        self.cfg = config
++        if self.pc is None:
++            try:
++                import pycozmo  # noqa: F401 (lazy: describe/probe must work without it)
++            except ImportError:
++                raise Fault("unsupported", "PyCozmo is not installed (pip install 'openvibe-cozmo[robot]')")
++            self.pc = pycozmo
++        pc = self.pc
++        r = pc.robot
++        self.limits = {
++            "head_min": r.MIN_HEAD_ANGLE.radians, "head_max": r.MAX_HEAD_ANGLE.radians,
++            "lift_min": r.MIN_LIFT_HEIGHT.mm, "lift_max": r.MAX_LIFT_HEIGHT.mm,
++            "wheel_mmps": self._max_wheel_mmps(config, r.MAX_WHEEL_SPEED.mmps),
++            "accel": float(config.get("wheel_accel_mmps2", 0.0)),  # 0 = firmware default
++        }
++
++        cli = self.client_factory(pc, config)
++        timeout = float(config.get("connect_timeout_s", 10))
++        try:
++            cli.start()
++            cli.connect()
++            cli.wait_for_robot(timeout=timeout)
++        except Exception as e:
++            self._teardown(cli)
++            raise Fault("not_connected",
++                        "could not reach Cozmo within %.0f s (%s). Join this computer to Cozmo's Wi-Fi (the network "
++                        "named 'Cozmo_...'; the password is shown on Cozmo's face when you raise and lower the lift) "
++                        "and check nothing else (the Cozmo app) is connected to it." % (timeout, e))
++        self.client = cli
++        self._check_firmware(cli, config)
++
++        # Safety callbacks come first and are registered even when the firmware is unsupported.
++        ev = pc.event
++        cli.add_handler(ev.EvtCliffDetectedChange, self._on_cliff)
++        cli.add_handler(ev.EvtRobotPickedUpChange, self._on_picked_up)
++        self.picked_up = bool(getattr(cli, "robot_picked_up", False))
++        cliff_flag = getattr(getattr(r, "RobotStatusFlag", None), "CLIFF_DETECTED", 0)
++        self.cliff = bool(cliff_flag and (getattr(cli, "robot_status", 0) & cliff_flag))
++        if self.fault:
++            return
++
++        self._start_speaker(config)
++        if config.get("camera", True):
++            self._setup_frame_dir(config)
++            cli.add_handler(ev.EvtNewRawCameraImage, self._on_camera_image)
++            cli.enable_camera(True, color=bool(config.get("color", False)))
++            self._camera_on = True
++        if "volume" in config:
++            cli.set_volume(int(clamp(config["volume"], 0, 1) * 65535))
++        self.log("cozmo: connected, firmware %s", (cli.robot_fw_sig or {}).get("version"))
++
++    def _check_firmware(self, cli, config):
++        sig = cli.robot_fw_sig or {}
++        supported = self.pc.protocol_declaration.FIRMWARE_VERSION
++        allowed = config.get("allow_firmware") or [supported]
++        try:
++            allowed = [int(v) for v in allowed]
++        except (TypeError, ValueError):
++            allowed = [supported]
++        version = sig.get("version")
++        build = sig.get("build")
++        if build == "FACTORY":
++            msg = ("Cozmo is running factory/recovery firmware (version %s); PyCozmo supports %s. Update the robot "
++                   "with the Cozmo app, then reconnect." % (version, supported))
++        elif version not in allowed:
++            msg = ("Cozmo firmware %s is not supported (supported: %s). Update the robot with the Cozmo app, or add "
++                   "the version to allow_firmware at your own risk." % (version, ", ".join(str(v) for v in allowed)))
++        else:
++            return
++        self.fault = ("firmware_unsupported", msg)
++        self.log("cozmo: %s", msg)
++        self.emit_event("firmware_unsupported", version=version, build=build, supported=allowed, message=msg)
++
++    def _setup_frame_dir(self, config):
++        d = config.get("frame_dir")
++        if d:
++            os.makedirs(d, exist_ok=True)
++            self.frame_dir, self._own_frame_dir = d, False
++        else:
++            self.frame_dir, self._own_frame_dir = tempfile.mkdtemp(prefix="openvibe-cozmo-frames-"), True
++
++    def close(self):
++        cli, self.client = self.client, None
++        if self._say_queue is not None:
++            self._say_queue.put(None)
++            if self._say_thread is not None:
++                self._say_thread.join(timeout=2)
++            self._say_queue = self._say_thread = None
++        if cli is not None:
++            if self._camera_on:
++                self._camera_on = False
++                try:
++                    cli.enable_camera(False)
++                except Exception:
++                    pass
++            self._teardown(cli)
++        for d, own in ((self.frame_dir, self._own_frame_dir), (self._work_dir, True)):
++            if d and own:
++                shutil.rmtree(d, ignore_errors=True)
++        self._work_dir = None
++
++    def _teardown(self, cli):
++        for name in ("disconnect", "stop"):
++            try:
++                getattr(cli, name)()
++            except Exception as e:
++                self.log("cozmo: %s failed: %r", name, e)
++
++    # Stopping. Called under the runtime lock from the main thread or from a PyCozmo thread via safe_stop.
++    def _halt_wheels(self, cli):
++        cli.drive_wheels(0.0, 0.0, 0.0, 0.0)
++
++    def stop_motion(self):
++        self._stop(motion_only=True)
++
++    def stop(self):
++        self._stop(motion_only=False)
++
++    def _stop(self, motion_only):
++        cli = self.client
++        if cli is None:
++            return
++        err = None
++        # Both steps always run; the first failure is re-raised so the runtime reports stop_failed.
++        try:
++            self._halt_wheels(cli)
++        except Exception as e:
++            err = e
++        try:
++            cli.stop_all_motors()  # wheels, head and lift
++        except Exception as e:
++            err = err or e
++        if err is not None:
++            raise err
++
++    # Commands.
++    def handle(self, cmd):
++        if self.fault:
++            raise Fault(*self.fault)
++        if self.client is None:
++            raise Fault("not_connected", "Cozmo is not connected")
++        v = cmd.value if isinstance(cmd.value, dict) else {}
++        k = cmd.kind
++        if k == "drive":
++            self._drive(v)
++        elif k == "ptz":
++            if "pan" in v and v["pan"] not in (0, None):
++                raise Fault("unsupported", "Cozmo has no pan axis; turn with drive")
++            if "tilt" in v:
++                self._head(v["tilt"])
++        elif k == "actuator":
++            self._actuator(v)
++        elif k == "say":
++            self._say(v)
++        elif k == "display":
++            self._display(v)
++        else:
++            raise Fault("unsupported", "kind %s" % k)
++
++    def wheel_speeds(self, throttle, steer):
++        """(left, right) in mm/s for a throttle/steer pair; public for tests and docs."""
++        t = clamp(throttle, -1, 1)
++        s = clamp(steer, -1, 1)
++        left, right = t + s, t - s
++        m = max(1.0, abs(left), abs(right))
++        vmax = self.limits.get("wheel_mmps", 150.0)
++        return left / m * vmax, right / m * vmax
++
++    def _drive(self, v):
++        if "throttle" in v or "steer" in v:
++            left, right = self.wheel_speeds(v.get("throttle", 0), v.get("steer", 0))
++        else:  # the dryrun's holonomic form; Cozmo ignores y
++            left, right = self.wheel_speeds(v.get("x", 0), v.get("rotation", 0))
++        if self.picked_up:
++            raise Fault("picked_up", "Cozmo is picked up; put it down on a flat surface")
++        if self.cliff and (left > 0 or right > 0):
++            raise Fault("cliff", "cliff detected; only backing away is allowed")
++        acc = self.limits.get("accel", 0.0)
++        # Never pass duration: PyCozmo sleeps in drive_wheels when it is set.
++        self.client.drive_wheels(left, right, acc, acc)
++
++    def head_radians(self, value):
++        lo, hi = self.limits["head_min"], self.limits["head_max"]
++        return lo + (clamp(value, -1, 1) + 1.0) / 2.0 * (hi - lo)
++
++    def lift_mm(self, value):
++        lo, hi = self.limits["lift_min"], self.limits["lift_max"]
++        return lo + clamp(value, 0, 1) * (hi - lo)
++
++    def _head(self, value):
++        self.client.set_head_angle(self.head_radians(value))
++
++    def _light_state(self, value):
++        pc = self.pc
++        if value is None or value is False:
++            rgb = (0, 0, 0)
++        elif isinstance(value, dict):
++            rgb = tuple(int(clamp(value.get(c, 0), 0, 255)) for c in ("r", "g", "b"))
++        elif isinstance(value, (list, tuple)) and len(value) == 3:
++            rgb = tuple(int(clamp(c, 0, 255)) for c in value)
++        else:
++            raise Fault("bad_value", "light value must be {r,g,b} (0..255) or null")
++        c = pc.lights.Color(rgb=rgb)
++        return pc.protocol_encoder.LightState(on_color=c.to_int16(), off_color=c.to_int16())
++
++    def _actuator(self, v):
++        name = v.get("name")
++        value = v.get("value")
++        cli = self.client
++        if name == "head":
++            self._head(value)
++        elif name == "lift":
++            cli.set_lift_height(self.lift_mm(value))
++        elif name == "backpack_lights":
++            cli.set_all_backpack_lights(self._light_state(value))
++        elif name == "head_light":
++            if not isinstance(value, bool):
++                raise Fault("bad_value", "head_light value must be true or false")
++            cli.set_head_light(value)
++        elif name == "cube_lights":
++            ls = self._light_state(value)
++            cubes = sorted(dict(cli.connected_objects))
++            if v.get("cube") is not None:
++                try:
++                    want = int(v["cube"])
++                except (TypeError, ValueError):
++                    raise Fault("bad_value", "cube must be an object id")
++                if want not in cubes:
++                    raise Fault("unavailable", "cube %s is not connected (connected: %s)" % (want, cubes))
++                cubes = [want]
++            if not cubes:
++                raise Fault("unavailable", "no cube connected")
++            pe = self.pc.protocol_encoder
++            for oid in cubes:
++                # CubeId selects the cube the following CubeLights applies to.
++                cli.conn.send(pe.CubeId(object_id=oid))
++                cli.conn.send(pe.CubeLights(states=(ls, ls, ls, ls)))
++        else:
++            raise Fault("bad_value", "unknown actuator %r" % (name,))
++
++    # Speech: synthesis and playback on a worker thread; handle() only queues.
++    def _start_speaker(self, config):
++        if self.synthesizer is None:
++            self.synthesizer = tts.find_synthesizer(config, self.which)
++        self._work_dir = tempfile.mkdtemp(prefix="openvibe-cozmo-tts-")
++        self._say_queue = queue.Queue(maxsize=SAY_QUEUE)
++        self._say_thread = threading.Thread(target=self._speaker, name="cozmo-say", daemon=True)
++        self._say_thread.start()
++
++    def _say(self, v):
++        if self.synthesizer is None:
++            raise Fault("unsupported", "no espeak-ng or pico2wave on this computer (apt install espeak-ng)")
++        text = tts.clean_text(v.get("text", ""), int(self.cfg.get("max_say_chars", 200)))
++        if not text:
++            raise Fault("bad_value", "say needs text")
++        try:
++            self._say_queue.put_nowait(text)
++        except queue.Full:
++            raise Fault("busy", "speech queue full")
++
++    def _speaker(self):
++        n = 0
++        while True:
++            text = self._say_queue.get()
++            if text is None:
++                return
++            n += 1
++            raw = os.path.join(self._work_dir, "say-%d-raw.wav" % n)
++            out = os.path.join(self._work_dir, "say-%d.wav" % n)
++            try:
++                self.synthesizer(text, raw)
++                tts.to_cozmo_wav(raw, out)
++                cli = self.client
++                if cli is not None:
++                    cli.play_audio(out)  # loads the whole file into packets before returning
++            except Exception as e:
++                self.log("cozmo: say failed: %r", e)
++                self.emit_event("say_failed", error=str(e))
++            finally:
++                for p in (raw, out):
++                    try:
++                        os.remove(p)
++                    except OSError:
++                        pass
++
++    def _display(self, v):
++        try:
++            if "image_png_b64" in v:
++                img = render.render_png_b64(v["image_png_b64"])
++            elif "face" in v:
++                img = render.render_face(v["face"])
++            elif "text" in v:
++                img = render.render_text(str(v["text"])[:200])
++            else:
++                raise Fault("bad_value", "display needs text, face or image_png_b64")
++        except ValueError as e:
++            raise Fault("bad_value", str(e))
++        self.client.display_image(img)  # no duration: PyCozmo would sleep
++
++    # PyCozmo callbacks (its threads). They must never raise: an exception would kill PyCozmo's dispatch thread.
++    def _on_cliff(self, cli, state):
++        self._on_hazard("cliff", "cliff", state)
++
++    def _on_picked_up(self, cli, state):
++        self._on_hazard("picked_up", "picked_up", state)
++
++    def _on_hazard(self, attr, name, state):
++        state = bool(state)
++        setattr(self, attr, state)
++        if not state:
++            self.emit_event(name + "_cleared")
++            return
++        # Stop the wheels right here, before waiting for the runtime lock that the main thread may hold.
++        c = self.client
++        if c is not None:
++            try:
++                self._halt_wheels(c)
++                c.stop_all_motors()
++            except Exception as e:
++                self.log("cozmo: %s stop failed: %r", name, e)
++        self.emit_event(name, state=True)
++        try:
++            if self.runtime is not None:
++                self.runtime.safe_stop(name)
++        except Exception as e:
++            self.log("cozmo: safe_stop(%s) failed: %r", name, e)
++
++    def _on_camera_image(self, cli, image):
++        try:
++            now = self.clock()
++            with self._frame_lock:
++                if now < self._next_frame or self.frame_dir is None:
++                    return
++                self._next_frame = now + 1.0 / self._fps(self.cfg)
++                self._frame_seq += 1
++                seq = self._frame_seq
++            path = os.path.join(self.frame_dir, "frame-%d.jpg" % (seq % FRAME_RING))
++            tmp = path + ".tmp"
++            image.convert("RGB").save(tmp, "JPEG", quality=int(self.cfg.get("jpeg_quality", 70)))
++            os.replace(tmp, path)  # readers never see a partial file
++            self.emit_video(path, format="jpeg", width=image.width, height=image.height, seq=seq)
++        except Exception as e:
++            self.log("cozmo: camera frame failed: %r", e)
++
++    # Telemetry.
++    def poll(self):
++        cli = self.client
++        faults = [self.fault[0]] if self.fault else []
++        if cli is None:
++            return {"faults": faults} if faults else None
++        self._connect_cubes(cli)
++        t = {"sensors": {"cliff": self.cliff, "picked_up": self.picked_up,
++                         "cubes": len(getattr(cli, "connected_objects", {}) or {})}}
++        try:
++            t["sensors"]["head_deg"] = round(cli.head_angle.degrees, 1)
++            t["sensors"]["lift_mm"] = round(cli.lift_position.height.mm, 1)
++        except AttributeError:
++            pass
++        v = float(getattr(cli, "battery_voltage", 0.0) or 0.0)
++        if v > 0:  # 0 until the first RobotState arrives
++            pct = (v - BATTERY_EMPTY_V) / (BATTERY_FULL_V - BATTERY_EMPTY_V) * 100.0
++            t["battery"] = {"volts": round(v, 2), "percent": int(max(0.0, min(100.0, pct)))}
++            threshold = float(self.cfg.get("low_battery_v", 3.6))
++            if v < threshold and not self.low_battery:
++                self.low_battery = True
++                self.emit_event("low_battery", volts=round(v, 2), threshold=threshold)
++            elif v > threshold + 0.1:  # hysteresis so a sagging cell does not flap
++                self.low_battery = False
++        if self.low_battery:
++            faults.append("low_battery")
++        t["faults"] = faults
++        return t
++
++    def _connect_cubes(self, cli):
++        """PyCozmo lists cubes it hears (available_objects) but does not connect them; ask once per cube."""
++        if self.fault or not self.cfg.get("connect_cubes", True):
++            return
++        try:
++            for fid, obj in list(cli.available_objects.items()):
++                if fid in self._cubes_requested:
++                    continue
++                kind = getattr(obj.object_type, "name", str(obj.object_type))
++                if kind.startswith("Block_LIGHTCUBE"):
++                    self._cubes_requested.add(fid)
++                    cli.conn.send(self.pc.protocol_encoder.ObjectConnect(factory_id=fid, connect=True))
++        except Exception as e:
++            self.log("cozmo: cube connect failed: %r", e)
ddiff --git a/plugins/cozmo/openvibe_cozmo/__main__.py b/plugins/cozmo/openvibe_cozmo/__main__.py
nnew file mode 100644
iindex 0000000..2410712
---- /dev/null
++++ b/plugins/cozmo/openvibe_cozmo/__main__.py
@@@ -0,0 +1,6 @@
++import sys
++
++from openvibe_cozmo import Cozmo
++from openvibe_plugin import run
++
++sys.exit(run(Cozmo()))
ddiff --git a/plugins/cozmo/openvibe_cozmo/render.py b/plugins/cozmo/openvibe_cozmo/render.py
nnew file mode 100644
iindex 0000000..c92682f
---- /dev/null
++++ b/plugins/cozmo/openvibe_cozmo/render.py
@@@ -0,0 +1,130 @@
++"""Rendering for Cozmo's face screen.
++
++PyCozmo's ``Client.display_image`` takes a 128x32 image in mode "1" (its ImageEncoder rejects anything else); the robot
++doubles it vertically on its 128x64 OLED. Everything here is pure Pillow so it can be tested without a robot.
++"""
++
++import base64
++import io
++
++from PIL import Image, ImageDraw, ImageFont, ImageOps
++
++WIDTH = 128
++HEIGHT = 32
++MAX_PNG_B64 = 256 * 1024  # a 128x32 picture never needs more; refuse bigger payloads before decoding
++
++FACES = ("neutral", "happy", "sad", "surprised", "sleepy", "angry")
++
++
++def blank():
++    return Image.new("1", (WIDTH, HEIGHT), 0)
++
++
++def _font():
++    try:
++        return ImageFont.load_default(size=11)  # Pillow >= 10.1 (FreeType build)
++    except TypeError:
++        return ImageFont.load_default()
++
++
++def _wrap(draw, text, font, width):
++    """Greedy word wrap by pixel width. Words longer than a line are cut by characters."""
++    lines = []
++    for para in text.splitlines() or [""]:
++        line = ""
++        for word in para.split():
++            cand = word if not line else line + " " + word
++            if draw.textlength(cand, font=font) <= width:
++                line = cand
++                continue
++            if line:
++                lines.append(line)
++            while draw.textlength(word, font=font) > width and len(word) > 1:
++                cut = len(word)
++                while cut > 1 and draw.textlength(word[:cut], font=font) > width:
++                    cut -= 1
++                lines.append(word[:cut])
++                word = word[cut:]
++            line = word
++        lines.append(line)
++    return lines
++
++
++def render_text(text):
++    """Centered text, wrapped to as many lines as fit; the rest is cut with an ellipsis."""
++    img = blank()
++    draw = ImageDraw.Draw(img)
++    font = _font()
++    text = " ".join(str(text).split()) if "\n" not in str(text) else str(text)
++    lines = _wrap(draw, text, font, WIDTH - 2)
++    box = draw.textbbox((0, 0), "Ag", font=font)
++    line_h = max(1, box[3] - box[1] + 1)
++    max_lines = max(1, HEIGHT // line_h)
++    if len(lines) > max_lines:
++        lines = lines[:max_lines]
++        last = lines[-1]
++        while last and draw.textlength(last + "...", font=font) > WIDTH - 2:
++            last = last[:-1]
++        lines[-1] = last + "..."
++    top = (HEIGHT - line_h * len(lines)) // 2 - box[1]
++    for i, line in enumerate(lines):
++        w = draw.textlength(line, font=font)
++        draw.text(((WIDTH - w) // 2, top + i * line_h), line, fill=1, font=font)
++    return img
++
++
++def render_face(name):
++    """Two simple eyes. Unknown names raise ValueError."""
++    name = str(name).lower()
++    if name not in FACES:
++        raise ValueError("unknown face %r (known: %s)" % (name, ", ".join(FACES)))
++    img = blank()
++    d = ImageDraw.Draw(img)
++    for cx in (40, 88):
++        x0, x1 = cx - 14, cx + 14
++        if name == "neutral":
++            d.rounded_rectangle((x0, 6, x1, 26), radius=5, fill=1)
++        elif name == "happy":
++            # Upward arcs (closed, smiling eyes).
++            d.pieslice((x0, 8, x1, 36), 180, 360, fill=1)
++            d.pieslice((x0 + 5, 14, x1 - 5, 40), 180, 360, fill=0)
++        elif name == "sad":
++            # Eyes with the outer top corner cut away, drooping outwards.
++            d.rounded_rectangle((x0, 10, x1, 26), radius=4, fill=1)
++            if cx < 64:
++                d.polygon([(x0 - 1, 9), (x0 - 1, 18), (x1 + 1, 9)], fill=0)
++            else:
++                d.polygon([(x1 + 1, 9), (x1 + 1, 18), (x0 - 1, 9)], fill=0)
++        elif name == "surprised":
++            d.ellipse((cx - 12, 3, cx + 12, 29), fill=1)
++            d.ellipse((cx - 5, 10, cx + 5, 22), fill=0)
++        elif name == "sleepy":
++            d.rectangle((x0, 19, x1, 23), fill=1)
++        elif name == "angry":
++            d.rounded_rectangle((x0, 8, x1, 26), radius=4, fill=1)
++            if cx < 64:
++                d.polygon([(x0 - 1, 7), (x1 + 1, 7), (x1 + 1, 16)], fill=0)
++            else:
++                d.polygon([(x0 - 1, 7), (x1 + 1, 7), (x0 - 1, 16)], fill=0)
++    return img
++
++
++def render_png_b64(data):
++    """Decode a base64 image (PNG or anything Pillow reads), fit it into 128x32 keeping aspect, dither to 1 bit."""
++    if not isinstance(data, str) or not data:
++        raise ValueError("image_png_b64 must be a non-empty string")
++    if len(data) > MAX_PNG_B64:
++        raise ValueError("image too large (%d bytes of base64, max %d)" % (len(data), MAX_PNG_B64))
++    try:
++        raw = base64.b64decode(data, validate=False)
++        src = Image.open(io.BytesIO(raw))  # lazy: reads the header only
++        if src.width * src.height > 4096 * 4096:
++            raise ValueError("image dimensions too large")
++        src.load()
++    except ValueError:
++        raise
++    except Exception as e:
++        raise ValueError("not a readable image: %s" % e)
++    grey = src.convert("L")
++    fitted = ImageOps.pad(grey, (WIDTH, HEIGHT), color=0)
++    return fitted.convert("1")
ddiff --git a/plugins/cozmo/openvibe_cozmo/tts.py b/plugins/cozmo/openvibe_cozmo/tts.py
nnew file mode 100644
iindex 0000000..e686449
---- /dev/null
++++ b/plugins/cozmo/openvibe_cozmo/tts.py
@@@ -0,0 +1,104 @@
++"""Text to speech for Cozmo.
++
++PyCozmo's ``audio.load_wav`` accepts 16-bit WAV at 22050 Hz or 48000 Hz. ``espeak-ng`` writes 22050 Hz mono 16-bit;
++``pico2wave`` writes 16000 Hz mono 16-bit, so every synthesized file goes through ``to_cozmo_wav`` which converts to
++22050 Hz mono 16-bit with a small pure-Python linear resampler (``audioop`` is gone in Python 3.13).
++
++A synthesizer is any callable ``synth(text, wav_path)`` that writes a WAV file or raises. ``find_synthesizer`` picks one
++from the binaries on PATH; tests inject their own.
++"""
++
++import shutil
++import subprocess
++import sys
++import wave
++from array import array
++
++COZMO_RATE = 22050
++SYNTH_TIMEOUT_S = 30
++
++
++def clean_text(text, max_chars):
++    """Printable, single-spaced, length-limited text. A leading '-' is dropped so no TTS binary reads it as a flag."""
++    text = "".join(ch if ch.isprintable() else " " for ch in str(text))
++    text = " ".join(text.split())[:max_chars]
++    return text.lstrip("-").strip()
++
++
++def espeak_ng(binary="espeak-ng", voice=None, speed=150):
++    def synth(text, wav_path):
++        cmd = [binary, "-w", wav_path, "-s", str(int(speed)), "--stdin"]
++        if voice:
++            cmd[1:1] = ["-v", str(voice)]
++        subprocess.run(cmd, input=text.encode("utf-8"), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
++                       timeout=SYNTH_TIMEOUT_S, check=True)
++    synth.engine = "espeak-ng"
++    return synth
++
++
++def pico2wave(binary="pico2wave", lang="en-US"):
++    def synth(text, wav_path):
++        # pico2wave insists on a .wav suffix, which our paths have.
++        subprocess.run([binary, "-l", str(lang), "-w", wav_path, text], stdout=subprocess.DEVNULL,
++                       stderr=subprocess.PIPE, timeout=SYNTH_TIMEOUT_S, check=True)
++    synth.engine = "pico2wave"
++    return synth
++
++
++def find_synthesizer(config, which=shutil.which):
++    """Return a synthesizer for the first TTS binary found (espeak-ng, then pico2wave), or None."""
++    path = which("espeak-ng")
++    if path:
++        return espeak_ng(path, voice=config.get("voice"), speed=config.get("speech_rate", 150))
++    path = which("pico2wave")
++    if path:
++        return pico2wave(path, lang=config.get("pico_lang", "en-US"))
++    return None
++
++
++def resample_linear(samples, src_rate, dst_rate):
++    """Linear-interpolation resampler for a sequence of ints. Good enough for speech."""
++    n = len(samples)
++    if src_rate == dst_rate or n == 0:
++        return array("h", samples)
++    n_out = max(1, int(n * dst_rate / src_rate))
++    step = src_rate / dst_rate
++    last = n - 1
++    out = array("h", bytes(2 * n_out))
++    for i in range(n_out):
++        pos = i * step
++        j = int(pos)
++        if j >= last:
++            out[i] = samples[last]
++            continue
++        f = pos - j
++        a = samples[j]
++        out[i] = int(round(a + (samples[j + 1] - a) * f))
++    return out
++
++
++def to_cozmo_wav(src_path, dst_path, rate=COZMO_RATE):
++    """Convert a 16-bit PCM WAV of any rate and channel count to mono 16-bit at ``rate``."""
++    with wave.open(src_path, "rb") as w:
++        channels = w.getnchannels()
++        width = w.getsampwidth()
++        src_rate = w.getframerate()
++        data = w.readframes(w.getnframes())
++    if width != 2:
++        raise ValueError("TTS produced %d-bit audio; only 16-bit PCM is supported" % (8 * width))
++    samples = array("h")
++    samples.frombytes(data[:len(data) - len(data) % 2])
++    if sys.byteorder == "big":
++        samples.byteswap()
++    if channels > 1:
++        samples = array("h", (sum(samples[i:i + channels]) // channels
++                              for i in range(0, len(samples) - channels + 1, channels)))
++    out = resample_linear(samples, src_rate, rate)
++    if sys.byteorder == "big":
++        out.byteswap()
++    with wave.open(dst_path, "wb") as w:
++        w.setnchannels(1)
++        w.setsampwidth(2)
++        w.setframerate(rate)
++        w.writeframes(out.tobytes())
++    return len(out)
ddiff --git a/plugins/cozmo/pyproject.toml b/plugins/cozmo/pyproject.toml
nnew file mode 100644
iindex 0000000..6852af0
---- /dev/null
++++ b/plugins/cozmo/pyproject.toml
@@@ -0,0 +1,17 @@
++[build-system]
++requires = ["setuptools>=61"]
++build-backend = "setuptools.build_meta"
++
++[project]
++name = "openvibe-cozmo"
++version = "0.1.0"
++description = "OpenVibe Node bridge plugin for Anki/DDL Cozmo (via PyCozmo)"
++requires-python = ">=3.9"
++license = { text = "AGPL-3.0-or-later" }
++dependencies = ["openvibe-plugin", "pillow"]
++
++[project.optional-dependencies]
++robot = ["pycozmo>=0.8.0,<0.9"]
++
++[tool.setuptools]
++packages = ["openvibe_cozmo"]
ddiff --git a/plugins/cozmo/tests/fake_pycozmo.py b/plugins/cozmo/tests/fake_pycozmo.py
nnew file mode 100644
iindex 0000000..666e8a6
---- /dev/null
++++ b/plugins/cozmo/tests/fake_pycozmo.py
@@@ -0,0 +1,239 @@
++"""A stand-in for the parts of PyCozmo 0.8.0 the plugin uses, so tests run without it (or a robot).
++
++Names, signatures and constants mirror pycozmo.Client, pycozmo.robot, pycozmo.event, pycozmo.lights and
++pycozmo.protocol_encoder. Every Client call is recorded in ``calls`` as (name, args, kwargs, thread).
++"""
++
++import collections
++import math
++import threading
++import types
++import wave
++
++
++class Angle:
++    def __init__(self, radians=None, degrees=None):
++        self.radians = radians if radians is not None else math.radians(degrees)
++
++    @property
++    def degrees(self):
++        return math.degrees(self.radians)
++
++
++class Distance:
++    def __init__(self, mm):
++        self.mm = mm
++
++
++class Speed:
++    def __init__(self, mmps):
++        self.mmps = mmps
++
++
++class LiftPosition:
++    def __init__(self, height):
++        self.height = height
++
++
++class RobotStatusFlag:
++    IS_PICKED_UP = 0x8
++    CLIFF_DETECTED = 0x10000
++
++
++robot = types.SimpleNamespace(
++    MIN_HEAD_ANGLE=Angle(degrees=-25), MAX_HEAD_ANGLE=Angle(degrees=44.5),
++    MIN_LIFT_HEIGHT=Distance(mm=32.0), MAX_LIFT_HEIGHT=Distance(mm=92.0),
++    MAX_WHEEL_SPEED=Speed(mmps=200.0), RobotStatusFlag=RobotStatusFlag)
++
++
++class Event:
++    pass
++
++
++class EvtCliffDetectedChange(Event):
++    pass
++
++
++class EvtRobotPickedUpChange(Event):
++    pass
++
++
++class EvtNewRawCameraImage(Event):
++    pass
++
++
++event = types.SimpleNamespace(EvtCliffDetectedChange=EvtCliffDetectedChange,
++                              EvtRobotPickedUpChange=EvtRobotPickedUpChange,
++                              EvtNewRawCameraImage=EvtNewRawCameraImage)
++
++
++class Color:
++    def __init__(self, int_color=None, rgb=None, name=None):
++        self.rgb = rgb
++
++    def to_int16(self):
++        r, g, b = self.rgb
++        return (r * 31 // 255) << 10 | (g * 31 // 255) << 5 | (b * 31 // 255)
++
++
++lights = types.SimpleNamespace(Color=Color)
++
++
++class Packet:
++    def __init__(self, **kw):
++        self.__dict__.update(kw)
++
++    def __repr__(self):
++        return "%s(%r)" % (type(self).__name__, self.__dict__)
++
++
++class LightState(Packet):
++    def __init__(self, on_color=0, off_color=0, on_frames=0, off_frames=0, transition_on_frames=0,
++                 transition_off_frames=0, offset=0):
++        super().__init__(on_color=on_color, off_color=off_color)
++
++
++class CubeId(Packet):
++    def __init__(self, object_id=0, rotation_period_frames=0):
++        super().__init__(object_id=object_id, rotation_period_frames=rotation_period_frames)
++
++
++class CubeLights(Packet):
++    def __init__(self, states=()):
++        assert len(states) == 4
++        super().__init__(states=states)
++
++
++class ObjectConnect(Packet):
++    def __init__(self, factory_id=0, connect=False):
++        super().__init__(factory_id=factory_id, connect=connect)
++
++
++protocol_encoder = types.SimpleNamespace(LightState=LightState, CubeId=CubeId, CubeLights=CubeLights,
++                                         ObjectConnect=ObjectConnect)
++protocol_declaration = types.SimpleNamespace(FIRMWARE_VERSION=2381)
++
++
++class ConnectionTimeout(Exception):
++    pass
++
++
++class Conn:
++    def __init__(self, client):
++        self.client = client
++        self.sent = []
++
++    def send(self, pkt):
++        self.sent.append(pkt)
++
++
++class ObjectType:
++    def __init__(self, name):
++        self.name = name
++
++
++class FakeObject:
++    def __init__(self, factory_id, object_type):
++        self.factory_id = factory_id
++        self.object_type = ObjectType(object_type)
++
++
++class Client:
++    """Records calls; ``reachable=False`` makes wait_for_robot time out."""
++
++    def __init__(self, robot_addr=None, protocol_log_messages=None, auto_initialize=True, enable_animations=True,
++                 enable_procedural_face=True, fw_sig=None, reachable=True):
++        self.calls = []
++        self.conn = Conn(self)
++        self.reachable = reachable
++        self._fw = fw_sig if fw_sig is not None else {"version": 2381, "build": "DEVELOPMENT"}
++        self.robot_fw_sig = None
++        self.dispatch_handlers = collections.defaultdict(list)
++        self.head_angle = Angle(radians=robot.MIN_HEAD_ANGLE.radians)
++        self.lift_position = LiftPosition(height=Distance(mm=32.0))
++        self.battery_voltage = 0.0
++        self.robot_status = 0
++        self.robot_picked_up = False
++        self.available_objects = {}
++        self.connected_objects = {}
++        self.played = []  # (path, frames, rate, channels, thread) for play_audio
++
++    def _rec(self, name, *args, **kwargs):
++        self.calls.append((name, args, kwargs, threading.current_thread()))
++
++    def names(self):
++        return [c[0] for c in self.calls]
++
++    def calls_to(self, name):
++        return [c for c in self.calls if c[0] == name]
++
++    # Lifecycle.
++    def start(self):
++        self._rec("start")
++
++    def stop(self):
++        self._rec("stop")
++        self.dispatch_handlers = collections.defaultdict(list)
++
++    def connect(self):
++        self._rec("connect")
++
++    def disconnect(self):
++        self._rec("disconnect")
++
++    def wait_for_robot(self, timeout=5.0):
++        self._rec("wait_for_robot", timeout=timeout)
++        if not self.reachable:
++            raise ConnectionTimeout("Failed to connect to Cozmo.")
++        self.robot_fw_sig = self._fw
++
++    # Events (pycozmo.event.Dispatcher).
++    def add_handler(self, event, f, one_shot=False):
++        self.dispatch_handlers[event].append(f)
++
++    def dispatch(self, event, *args, **kwargs):
++        for f in list(self.dispatch_handlers[event]):
++            f(*args, **kwargs)
++
++    # Motion.
++    def drive_wheels(self, lwheel_speed, rwheel_speed, lwheel_acc=0.0, rwheel_acc=0.0, duration=None):
++        self._rec("drive_wheels", lwheel_speed, rwheel_speed, lwheel_acc, rwheel_acc, duration=duration)
++
++    def stop_all_motors(self):
++        self._rec("stop_all_motors")
++
++    def set_head_angle(self, angle, accel=10.0, max_speed=10.0, duration=0.0):
++        self._rec("set_head_angle", angle)
++
++    def set_lift_height(self, height, accel=10.0, max_speed=10.0, duration=0.0):
++        self._rec("set_lift_height", height)
++
++    # Lights, screen, sound, camera.
++    def set_all_backpack_lights(self, light):
++        self._rec("set_all_backpack_lights", light)
++
++    def set_head_light(self, enable):
++        self._rec("set_head_light", enable)
++
++    def display_image(self, im, duration=None):
++        if im.size != (128, 32) or im.mode != "1":  # what pycozmo.image_encoder.ImageEncoder enforces
++            raise ValueError("bad image %s %s" % (im.size, im.mode))
++        self._rec("display_image", im, duration=duration)
++
++    def play_audio(self, fspec):
++        with wave.open(fspec, "rb") as w:  # what pycozmo.audio.load_wav enforces
++            if w.getsampwidth() != 2 or w.getframerate() not in (22050, 48000):
++                raise ValueError("Invalid audio format")
++            self.played.append((fspec, w.getnframes(), w.getframerate(), w.getnchannels(),
++                                threading.current_thread()))
++        self._rec("play_audio", fspec)
++
++    def enable_camera(self, enable=True, color=False):
++        self._rec("enable_camera", enable, color=color)
++
++    def set_volume(self, level):
++        self._rec("set_volume", level)
++
++
++module = types.SimpleNamespace(Client=Client, robot=robot, event=event, lights=lights,
++                               protocol_encoder=protocol_encoder, protocol_declaration=protocol_declaration)
ddiff --git a/plugins/cozmo/tests/test_cozmo.py b/plugins/cozmo/tests/test_cozmo.py
nnew file mode 100644
iindex 0000000..1bd5665
---- /dev/null
++++ b/plugins/cozmo/tests/test_cozmo.py
@@@ -0,0 +1,546 @@
++import base64
++import io
++import json
++import math
++import os
++import sys
++import tempfile
++import threading
++import time
++import unittest
++import wave
++
++HERE = os.path.dirname(os.path.abspath(__file__))
++sys.path.insert(0, HERE)
++sys.path.insert(0, os.path.dirname(HERE))
++sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))
++
++from PIL import Image  # noqa: E402
++
++import fake_pycozmo as fpc  # noqa: E402
++from openvibe_cozmo import Cozmo, render, tts  # noqa: E402
++from openvibe_plugin import Runtime  # noqa: E402
++
++
++class Clock:
++    t = 10.0
++
++    def __call__(self):
++        return self.t
++
++
++def write_wav(path, rate=16000, n=1600, channels=1):
++    with wave.open(path, "wb") as w:
++        w.setnchannels(channels)
++        w.setsampwidth(2)
++        w.setframerate(rate)
++        frames = bytearray()
++        for i in range(n):
++            s = int(8000 * math.sin(i / 10.0))
++            frames += s.to_bytes(2, "little", signed=True) * channels
++        w.writeframes(bytes(frames))
++
++
++class FakeTTS:
++    engine = "fake"
++
++    def __init__(self):
++        self.texts = []
++        self.threads = []
++
++    def __call__(self, text, wav_path):
++        self.texts.append(text)
++        self.threads.append(threading.current_thread())
++        write_wav(wav_path, rate=16000)  # like pico2wave: must be resampled to 22050
++
++
++class Base(unittest.TestCase):
++    fw_sig = None
++    config = {}
++
++    def setUp(self):
++        self.out = io.StringIO()
++        self.clock = Clock()
++        self.clients = []
++        self.tts = FakeTTS()
++        self.frames = tempfile.mkdtemp()
++
++        def factory(pc, config):
++            c = fpc.Client(fw_sig=self.fw_sig, reachable=config.get("_reachable", True))
++            self.clients.append(c)
++            return c
++
++        self.p = Cozmo(client_factory=factory, pycozmo_module=fpc.module, synthesizer=self.tts, clock=self.clock)
++        self.p.log = lambda *a: None
++        self.rt = Runtime(self.p, infile=io.StringIO(""), outfile=self.out, clock=self.clock)
++
++    def tearDown(self):
++        self.p.close()
++
++    def hello(self, **config):
++        cfg = {"frame_dir": self.frames}
++        cfg.update(self.config)
++        cfg.update(config)
++        self.send({"op": "hello", "config": cfg})
++
++    @property
++    def cli(self):
++        return self.clients[-1]
++
++    def send(self, m):
++        self.rt.feed_line(json.dumps(m))
++
++    def msgs(self):
++        return [json.loads(line) for line in self.out.getvalue().splitlines()]
++
++    def last(self):
++        return self.msgs()[-1]
++
++    def cmd(self, kind, value, id="c", deadline_ms=300):
++        self.send({"op": "command", "id": id, "kind": kind, "value": value, "deadline_ms": deadline_ms})
++        return self.last()
++
++    def events(self, name):
++        return [m for m in self.msgs() if m.get("op") == "event" and m.get("name") == name]
++
++
++class ConnectTest(Base):
++    def test_ready_and_describe(self):
++        self.hello()
++        ms = self.msgs()
++        self.assertEqual(ms[0]["op"], "describe")
++        self.assertEqual(ms[0]["driver"], "cozmo")
++        self.assertEqual(ms[0]["motion_kinds"], ["drive"])
++        caps = ms[0]["capabilities"]
++        self.assertTrue(caps["bridge"])
++        self.assertEqual(caps["camera"], {"jpeg": True, "width": 320, "height": 240, "color": False, "fps": 10.0})
++        self.assertEqual(caps["display"]["width"], 128)
++        self.assertEqual(caps["display"]["height"], 32)
++        self.assertIn("cube_lights", caps["actuator"]["names"])
++        self.assertEqual(ms[-1], {"op": "ready"})
++        self.assertEqual(self.cli.names()[:3], ["start", "connect", "wait_for_robot"])
++        self.assertEqual(self.cli.calls_to("wait_for_robot")[0][2], {"timeout": 10.0})
++        self.assertIn(("enable_camera", (True,), {"color": False}), [c[:3] for c in self.cli.calls])
++        # The runtime stops everything right after setup.
++        self.assertIn("stop_all_motors", self.cli.names())
++
++    def test_probe_never_creates_client(self):
++        self.send({"op": "hello", "config": {}, "probe": True})
++        self.assertEqual(self.clients, [])
++        self.assertEqual([m["op"] for m in self.msgs()], ["describe"])
++
++    def test_describe_without_pycozmo(self):
++        p = Cozmo(which=lambda name: None)
++        caps = p.describe({})
++        self.assertIsNone(caps["say"]["engine"])
++        self.assertIsNone(p.pc)
++
++    def test_unreachable_faults_and_cleans_up(self):
++        self.hello(_reachable=False, connect_timeout_s=2)
++        f = self.last()
++        self.assertEqual(f["op"], "fault")
++        self.assertEqual(f["fault_code"], "not_connected")
++        self.assertIn("Wi-Fi", f["message"])
++        self.assertEqual(self.cli.names()[-2:], ["disconnect", "stop"])
++        self.assertIsNone(self.p.client)
++        self.p.stop()  # safe while unconnected
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["fault_code"], "not_ready")
++
++    def test_stop_is_safe_before_setup(self):
++        self.p.stop()
++        self.p.stop_motion()
++        self.assertIsNone(self.p.poll())
++
++
++class DriveTest(Base):
++    def setUp(self):
++        super().setUp()
++        self.hello()
++        self.cli.calls.clear()
++
++    def last_drive(self):
++        return self.cli.calls_to("drive_wheels")[-1]
++
++    def test_drive_maps_to_mmps_without_duration(self):
++        for value, (l, r) in [
++            ({"throttle": 1, "steer": 0}, (150, 150)),
++            ({"throttle": -0.5, "steer": 0}, (-75, -75)),
++            ({"throttle": 0, "steer": 1}, (150, -150)),       # steer > 0 turns right: left wheel forward
++            ({"throttle": 1, "steer": 1}, (150, 0)),          # normalised: (2, 0) / 2
++            ({"throttle": 0.5, "steer": 0.25}, (112.5, 37.5)),
++            ({"throttle": 5, "steer": 0}, (150, 150)),        # clamped
++        ]:
++            self.assertEqual(self.cmd("drive", value)["op"], "ack", value)
++            name, args, kwargs, _ = self.last_drive()
++            self.assertAlmostEqual(args[0], l, msg=value)
++            self.assertAlmostEqual(args[1], r, msg=value)
++            self.assertLessEqual(len(args), 4)
++            self.assertIsNone(kwargs.get("duration"), "duration must never be passed")
++
++    def test_max_wheel_capped_by_hardware(self):
++        p = Cozmo(client_factory=lambda pc, c: fpc.Client(), pycozmo_module=fpc.module, synthesizer=self.tts)
++        p.log = lambda *a: None
++        rt = Runtime(p, infile=io.StringIO(""), outfile=io.StringIO(), clock=self.clock)
++        rt.feed_line(json.dumps({"op": "hello", "config": {"max_wheel_mmps": 500, "camera": False}}))
++        self.assertEqual(p.wheel_speeds(1, 0), (200.0, 200.0))
++        p.close()
++
++    def test_bad_value(self):
++        self.assertEqual(self.cmd("drive", {"throttle": "fast"})["fault_code"], "bad_value")
++        self.assertEqual(self.cli.calls_to("drive_wheels"), [])
++
++    def test_deadline_stops(self):
++        self.cmd("drive", {"throttle": 1}, deadline_ms=300)
++        self.clock.t += 0.2
++        self.rt.tick()
++        self.assertNotIn("stop_all_motors", self.cli.names())
++        self.clock.t += 0.11
++        self.rt.tick()
++        self.assertEqual(self.last_drive()[1][:2], (0.0, 0.0))
++        self.assertEqual(self.cli.names()[-1], "stop_all_motors")
++        self.assertEqual(self.rt.stop_reasons[-1], "deadline")
++
++    def test_heartbeat_loss_stops(self):
++        self.cmd("drive", {"throttle": 1}, deadline_ms=5000)
++        self.clock.t += 1.1
++        self.rt.tick()
++        self.assertEqual(self.rt.stop_reasons[-1], "heartbeat")
++        self.assertEqual(self.cli.names()[-2:], ["drive_wheels", "stop_all_motors"])
++        self.assertEqual(self.last_drive()[1][:2], (0.0, 0.0))
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["fault_code"], "no_heartbeat")
++
++    def test_estop_stops_and_refuses(self):
++        self.cmd("drive", {"throttle": 1})
++        self.send({"op": "estop"})
++        self.assertEqual(self.cli.names()[-1], "stop_all_motors")
++        n = len(self.cli.calls_to("drive_wheels"))
++        for kind, value in [("drive", {"throttle": 1}), ("ptz", {"tilt": 1}), ("actuator", {"name": "lift",
++                                                                                         "value": 1})]:
++            self.assertEqual(self.cmd(kind, value)["fault_code"], "estopped")
++        self.assertEqual(len(self.cli.calls_to("drive_wheels")), n)
++        self.assertEqual(self.cmd("halt", {}, id="h"), {"op": "ack", "id": "h"})
++        self.send({"op": "resume"})
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["op"], "ack")
++
++    def test_stop_failure_still_calls_stop_all_motors(self):
++        def boom(*a, **k):
++            raise OSError("socket gone")
++        self.cli.drive_wheels = boom
++        self.send({"op": "stop"})
++        self.assertEqual(self.cli.names()[-1], "stop_all_motors")
++        self.assertEqual(self.events("stop_failed")[0]["reason"], "stop")
++
++
++class HazardTest(Base):
++    def setUp(self):
++        super().setUp()
++        self.hello()
++
++    def fire(self, event, state):
++        t = threading.Thread(target=self.cli.dispatch, args=(event, self.cli, state), name="pycozmo-conn")
++        t.start()
++        t.join(2)
++
++    def test_cliff_stops_from_other_thread_and_blocks_forward(self):
++        self.cmd("drive", {"throttle": 1}, deadline_ms=5000)
++        self.cli.calls.clear()
++        self.fire(fpc.EvtCliffDetectedChange, True)
++        # First two calls happen on the event thread, before the runtime lock.
++        first = self.cli.calls[:2]
++        self.assertEqual([c[0] for c in first], ["drive_wheels", "stop_all_motors"])
++        self.assertEqual(first[0][1][:2], (0.0, 0.0))
++        self.assertEqual(first[0][3].name, "pycozmo-conn")
++        self.assertIn("cliff", self.rt.stop_reasons)
++        self.assertIsNone(self.rt.motion_deadline)
++        self.assertEqual(self.events("cliff")[0]["state"], True)
++
++        self.assertEqual(self.cmd("drive", {"throttle": 0.5})["fault_code"], "cliff")
++        self.assertEqual(self.cmd("drive", {"throttle": 0, "steer": 0.5})["fault_code"], "cliff")
++        n = len(self.cli.calls_to("drive_wheels"))
++        self.assertEqual(self.cmd("drive", {"throttle": -0.5})["op"], "ack")
++        self.assertEqual(self.cli.calls_to("drive_wheels")[n][1][:2], (-75.0, -75.0))
++
++        self.fire(fpc.EvtCliffDetectedChange, False)
++        self.assertEqual(len(self.events("cliff_cleared")), 1)
++        self.assertEqual(self.cmd("drive", {"throttle": 0.5})["op"], "ack")
++
++    def test_picked_up_stops_and_refuses_drive(self):
++        self.cmd("drive", {"throttle": 1}, deadline_ms=5000)
++        self.cli.calls.clear()
++        self.fire(fpc.EvtRobotPickedUpChange, True)
++        self.assertEqual([c[0] for c in self.cli.calls[:2]], ["drive_wheels", "stop_all_motors"])
++        self.assertEqual(self.cli.calls[0][1][:2], (0.0, 0.0))
++        self.assertIn("picked_up", self.rt.stop_reasons)
++        self.assertEqual(len(self.events("picked_up")), 1)
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["fault_code"], "picked_up")
++        self.assertEqual(self.cmd("drive", {"throttle": -1})["fault_code"], "picked_up")
++        self.assertEqual(self.cmd("ptz", {"tilt": 0})["op"], "ack")
++        self.fire(fpc.EvtRobotPickedUpChange, False)
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["op"], "ack")
++
++    def test_callback_never_raises_into_pycozmo(self):
++        def boom(*a, **k):
++            raise OSError("gone")
++        self.cli.drive_wheels = boom
++        self.cli.stop_all_motors = boom
++        self.p._on_cliff(self.cli, True)  # must not raise
++        self.assertTrue(self.p.cliff)
++
++
++class FirmwareTest(Base):
++    fw_sig = {"version": 2315, "build": "DEVELOPMENT"}
++
++    def test_mismatch_nacks_everything_but_halt(self):
++        self.hello()
++        self.assertEqual(self.last(), {"op": "ready"})  # the Node stays up and reports the fault
++        ev = self.events("firmware_unsupported")[0]
++        self.assertEqual(ev["version"], 2315)
++        self.assertEqual(ev["supported"], [2381])
++        for kind, value in [("drive", {"throttle": 1}), ("ptz", {"tilt": 1}), ("say", {"text": "hi"}),
++                            ("display", {"text": "hi"}), ("actuator", {"name": "lift", "value": 1})]:
++            r = self.cmd(kind, value)
++            self.assertEqual(r["fault_code"], "firmware_unsupported", kind)
++            self.assertIn("2315", r["message"])
++            self.assertIn("2381", r["message"])
++        self.assertEqual(self.cli.calls_to("drive_wheels")[-1][1][:2], (0.0, 0.0))  # only stops were sent
++        self.assertEqual(self.cmd("halt", {}, id="h"), {"op": "ack", "id": "h"})
++        self.assertNotIn("enable_camera", self.cli.names())
++        self.clock.t += 1
++        self.send({"op": "heartbeat"})
++        self.rt.tick()
++        self.assertEqual(self.last()["faults"], ["firmware_unsupported"])
++
++    def test_allow_firmware(self):
++        self.hello(allow_firmware=[2315, 2381])
++        self.assertEqual(self.events("firmware_unsupported"), [])
++        self.assertEqual(self.cmd("drive", {"throttle": 1})["op"], "ack")
++
++
++class FactoryFirmwareTest(Base):
++    fw_sig = {"version": 2381, "build": "FACTORY"}
++
++    def test_factory_build(self):
++        self.hello()
++        r = self.cmd("drive", {"throttle": 1})
++        self.assertEqual(r["fault_code"], "firmware_unsupported")
++        self.assertIn("factory", r["message"])
++
++
++class HeadLiftLightsTest(Base):
++    def setUp(self):
++        super().setUp()
++        self.hello()
++
++    def test_head_tilt_bounds(self):
++        for tilt, deg in [(-1, -25.0), (1, 44.5), (0, 9.75), (-3, -25.0), (3, 44.5)]:
++            self.assertEqual(self.cmd("ptz", {"tilt": tilt})["op"], "ack")
++            self.assertAlmostEqual(self.cli.calls_to("set_head_angle")[-1][1][0], math.radians(deg))
++        self.cmd("actuator", {"name": "head", "value": 1})
++        self.assertAlmostEqual(self.cli.calls_to("set_head_angle")[-1][1][0], math.radians(44.5))
++        self.assertEqual(self.cmd("ptz", {"pan": 0.5})["fault_code"], "unsupported")
++
++    def test_lift_bounds(self):
++        for value, mm in [(0, 32.0), (1, 92.0), (0.5, 62.0), (-1, 32.0), (2, 92.0)]:
++            self.assertEqual(self.cmd("actuator", {"name": "lift", "value": value})["op"], "ack")
++            self.assertAlmostEqual(self.cli.calls_to("set_lift_height")[-1][1][0], mm)
++
++    def test_lights(self):
++        self.assertEqual(self.cmd("actuator", {"name": "backpack_lights", "value": {"r": 255, "g": 0,
++                                                                                    "b": 0}})["op"], "ack")
++        ls = self.cli.calls_to("set_all_backpack_lights")[-1][1][0]
++        self.assertEqual((ls.on_color, ls.off_color), (31 << 10, 31 << 10))
++        self.assertEqual(self.cmd("actuator", {"name": "head_light", "value": True})["op"], "ack")
++        self.assertEqual(self.cli.calls_to("set_head_light")[-1][1], (True,))
++        self.assertEqual(self.cmd("actuator", {"name": "head_light", "value": "on"})["fault_code"], "bad_value")
++        self.assertEqual(self.cmd("actuator", {"name": "wings", "value": 1})["fault_code"], "bad_value")
++
++    def test_cube_lights(self):
++        self.assertEqual(self.cmd("actuator", {"name": "cube_lights", "value": {"g": 255}})["fault_code"],
++                         "unavailable")
++        self.cli.connected_objects = {1: {}, 2: {}}
++        self.cli.conn.sent.clear()
++        self.assertEqual(self.cmd("actuator", {"name": "cube_lights", "value": {"g": 255}})["op"], "ack")
++        sent = self.cli.conn.sent
++        self.assertEqual([type(p).__name__ for p in sent], ["CubeId", "CubeLights"] * 2)
++        self.assertEqual([p.object_id for p in sent[::2]], [1, 2])
++        self.assertEqual(sent[1].states[0].on_color, 31 << 5)
++        sent.clear()
++        self.assertEqual(self.cmd("actuator", {"name": "cube_lights", "value": None, "cube": 2})["op"], "ack")
++        self.assertEqual(sent[0].object_id, 2)
++        self.assertEqual(self.cmd("actuator", {"name": "cube_lights", "value": {}, "cube": 9})["fault_code"],
++                         "unavailable")
++
++    def test_cubes_connected_once(self):
++        self.cli.available_objects = {0xAB: fpc.FakeObject(0xAB, "Block_LIGHTCUBE1"),
++                                      0xCD: fpc.FakeObject(0xCD, "Charger_Basic")}
++        self.cli.conn.sent.clear()
++        self.p.poll()
++        self.p.poll()
++        self.assertEqual([(type(p).__name__, p.factory_id) for p in self.cli.conn.sent], [("ObjectConnect", 0xAB)])
++
++
++class SayTest(Base):
++    def test_say_plays_off_thread(self):
++        self.hello()
++        self.assertEqual(self.cmd("say", {"text": "hello  there\n" + "x" * 500})["op"], "ack")
++        deadline = time.time() + 5
++        while not self.cli.played and time.time() < deadline:
++            time.sleep(0.01)
++        self.assertEqual(len(self.cli.played), 1)
++        path, frames, rate, channels, thread = self.cli.played[0]
++        self.assertEqual((rate, channels), (22050, 1))
++        self.assertAlmostEqual(frames, 1600 * 22050 / 16000, delta=2)
++        self.assertIsNot(thread, threading.main_thread())
++        self.assertIsNot(self.tts.threads[0], threading.main_thread())
++        self.assertTrue(self.tts.texts[0].startswith("hello there x"))
++        self.assertEqual(len(self.tts.texts[0]), 200)
++
++    def test_empty_text(self):
++        self.hello()
++        self.assertEqual(self.cmd("say", {"text": "  "})["fault_code"], "bad_value")
++
++    def test_missing_tts_is_unsupported(self):
++        p = Cozmo(client_factory=lambda pc, c: fpc.Client(), pycozmo_module=fpc.module, which=lambda n: None)
++        p.log = lambda *a: None
++        out = io.StringIO()
++        rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=self.clock)
++        rt.feed_line(json.dumps({"op": "hello", "config": {"camera": False}}))
++        rt.feed_line(json.dumps({"op": "command", "id": "s", "kind": "say", "value": {"text": "hi"}}))
++        r = json.loads(out.getvalue().splitlines()[-1])
++        self.assertEqual(r["fault_code"], "unsupported")
++        self.assertIn("espeak-ng", r["message"])
++        p.close()
++
++    def test_find_synthesizer_prefers_espeak(self):
++        self.assertEqual(tts.find_synthesizer({}, lambda n: "/usr/bin/" + n).engine, "espeak-ng")
++        self.assertEqual(tts.find_synthesizer({}, lambda n: "/usr/bin/pico2wave" if n == "pico2wave" else None)
++                         .engine, "pico2wave")
++        self.assertIsNone(tts.find_synthesizer({}, lambda n: None))
++
++    def test_resample_stereo_48k(self):
++        d = tempfile.mkdtemp()
++        src, dst = os.path.join(d, "a.wav"), os.path.join(d, "b.wav")
++        write_wav(src, rate=48000, n=4800, channels=2)
++        tts.to_cozmo_wav(src, dst)
++        with wave.open(dst) as w:
++            self.assertEqual((w.getframerate(), w.getnchannels(), w.getsampwidth()), (22050, 1, 2))
++            self.assertAlmostEqual(w.getnframes(), 2205, delta=1)
++
++    def test_clean_text(self):
++        self.assertEqual(tts.clean_text("--help me\x00now", 200), "help me now")
++
++
++class DisplayTest(Base):
++    def setUp(self):
++        super().setUp()
++        self.hello()
++
++    def shown(self):
++        name, args, kwargs, _ = self.cli.calls_to("display_image")[-1]
++        self.assertIsNone(kwargs.get("duration"))
++        return args[0]
++
++    def test_text(self):
++        self.assertEqual(self.cmd("display", {"text": "Hello Cozmo"})["op"], "ack")
++        img = self.shown()
++        self.assertEqual((img.size, img.mode), ((128, 32), "1"))
++        self.assertGreater(img.convert("L").histogram()[255], 20)
++
++    def test_long_text_fits(self):
++        img = render.render_text("the quick brown fox jumps over the lazy dog " * 10)
++        self.assertEqual((img.size, img.mode), ((128, 32), "1"))
++
++    def test_faces(self):
++        for face in render.FACES:
++            self.assertEqual(self.cmd("display", {"face": face})["op"], "ack", face)
++            self.assertEqual(self.shown().size, (128, 32))
++        self.assertEqual(self.cmd("display", {"face": "smug"})["fault_code"], "bad_value")
++
++    def test_png(self):
++        src = Image.new("L", (64, 64), 255)
++        buf = io.BytesIO()
++        src.save(buf, "PNG")
++        b64 = base64.b64encode(buf.getvalue()).decode()
++        self.assertEqual(self.cmd("display", {"image_png_b64": b64})["op"], "ack")
++        img = self.shown()
++        self.assertEqual((img.size, img.mode), ((128, 32), "1"))
++        self.assertTrue(img.getpixel((64, 16)))
++        self.assertFalse(img.getpixel((0, 16)))  # padded
++        self.assertEqual(self.cmd("display", {"image_png_b64": "bm90IGFuIGltYWdl"})["fault_code"], "bad_value")
++        self.assertEqual(self.cmd("display", {})["fault_code"], "bad_value")
++
++
++class CameraTest(Base):
++    def test_throttle_and_jpeg(self):
++        self.hello(fps=10)
++        img = Image.new("RGB", (320, 240), (200, 10, 10))
++        for _ in range(3):  # same instant: only the first passes
++            self.cli.dispatch(fpc.EvtNewRawCameraImage, self.cli, img)
++        self.clock.t += 0.05
++        self.cli.dispatch(fpc.EvtNewRawCameraImage, self.cli, img)
++        self.clock.t += 0.06
++        self.cli.dispatch(fpc.EvtNewRawCameraImage, self.cli, img)
++        vids = [m for m in self.msgs() if m.get("op") == "video"]
++        self.assertEqual([v["seq"] for v in vids], [1, 2])
++        v = vids[0]
++        self.assertEqual((v["format"], v["width"], v["height"]), ("jpeg", 320, 240))
++        self.assertTrue(v["path"].startswith(self.frames))
++        with open(v["path"], "rb") as f:
++            self.assertEqual(f.read(2), b"\xff\xd8")
++        self.assertEqual(Image.open(v["path"]).size, (320, 240))
++        self.assertEqual([n for n in os.listdir(self.frames) if n.endswith(".tmp")], [])
++
++    def test_camera_disabled(self):
++        self.hello(camera=False)
++        self.assertNotIn("enable_camera", self.cli.names())
++
++
++class TelemetryTest(Base):
++    def test_poll_and_low_battery_once(self):
++        self.hello()
++        self.cli.battery_voltage = 3.85
++        self.cli.head_angle = fpc.Angle(degrees=10)
++        self.cli.lift_position = fpc.LiftPosition(fpc.Distance(50.0))
++        t = self.p.poll()
++        self.assertEqual(t["battery"], {"volts": 3.85, "percent": 50})
++        self.assertEqual(t["sensors"], {"cliff": False, "picked_up": False, "cubes": 0, "head_deg": 10.0,
++                                        "lift_mm": 50.0})
++        self.assertEqual(t["faults"], [])
++        self.cli.battery_voltage = 3.55
++        self.p.poll()
++        t = self.p.poll()
++        self.assertEqual(len(self.events("low_battery")), 1)
++        self.assertEqual(t["faults"], ["low_battery"])
++
++    def test_no_battery_before_first_state(self):
++        self.hello()
++        self.assertNotIn("battery", self.p.poll())
++
++
++class EOFTest(unittest.TestCase):
++    def test_run_stops_and_disconnects_at_eof(self):
++        clients = []
++
++        def factory(pc, config):
++            clients.append(fpc.Client())
++            return clients[-1]
++
++        p = Cozmo(client_factory=factory, pycozmo_module=fpc.module, synthesizer=FakeTTS())
++        p.log = lambda *a: None
++        lines = [{"op": "hello", "config": {"camera": False}},
++                 {"op": "command", "id": "d", "kind": "drive", "value": {"throttle": 1}, "deadline_ms": 5000}]
++        infile = io.StringIO("".join(json.dumps(m) + "\n" for m in lines))
++        out = io.StringIO()
++        rt = Runtime(p, infile=infile, outfile=out)
++        self.assertEqual(rt.run(), 0)
++        names = clients[0].names()
++        self.assertEqual(rt.stop_reasons[-1], "eof")
++        i = names.index("drive_wheels", names.index("stop_all_motors") + 1)  # the drive after setup's stop
++        self.assertEqual(clients[0].calls[i][1][:2], (150.0, 150.0))
++        self.assertEqual(names[-4:], ["drive_wheels", "stop_all_motors", "disconnect", "stop"])
++        self.assertEqual(clients[0].calls[-4][1][:2], (0.0, 0.0))
++        self.assertIsNone(p.client)
++
++
++if __name__ == "__main__":
++    unittest.main()