haptic_skin.necklace

  1"""High-level controller for the 8-motor necklace.
  2
  3Maps human directions to motor indices and exposes simple primitives
  4(``set_motor``, ``set_vector``, ``stop``...). All vibration *patterns* (pulse,
  5sweep, ...) live in :mod:`haptic_skin.patterns`; this class just sets PWM.
  6"""
  7
  8from __future__ import annotations
  9
 10from enum import IntEnum
 11from typing import List, Optional, Sequence
 12
 13from . import protocol
 14from .transport import Transport, open_transport
 15
 16
 17class Direction(IntEnum):
 18    """8 cardinal positions, in the wearer's frame. Index 0 = front (12 o'clock),
 19    going clockwise. These ARE the motor indices."""
 20
 21    N = 0   # front
 22    NE = 1
 23    E = 2   # right
 24    SE = 3
 25    S = 4   # back
 26    SW = 5
 27    W = 6   # left
 28    NW = 7
 29
 30
 31# Friendly aliases accepted by :meth:`Necklace.haptic`.
 32_DIRECTION_ALIASES = {
 33    "front": Direction.N, "forward": Direction.N, "straight": Direction.N, "n": Direction.N,
 34    "ne": Direction.NE, "front_right": Direction.NE,
 35    "right": Direction.E, "turn_right": Direction.E, "e": Direction.E,
 36    "se": Direction.SE, "back_right": Direction.SE,
 37    "back": Direction.S, "behind": Direction.S, "u_turn": Direction.S, "s": Direction.S,
 38    "sw": Direction.SW, "back_left": Direction.SW,
 39    "left": Direction.W, "turn_left": Direction.W, "w": Direction.W,
 40    "nw": Direction.NW, "front_left": Direction.NW,
 41}
 42
 43
 44# Wiring permutation of the physical prototype: BUILD_MOTOR_MAP[logical_index] =
 45# the physical wire/motor actually sitting at that orientation. Measured on the
 46# bench 2026-07-02 — the collier was wound as a MIRROR of the canonical order, so
 47# orientation o maps to wire (5 - o) mod 8. Injected by :func:`connect`. The
 48# :class:`Necklace` class itself defaults to identity so the library stays
 49# build-agnostic (unit tests keep asserting logical index == physical index).
 50BUILD_MOTOR_MAP: List[int] = [5, 4, 3, 2, 1, 0, 7, 6]
 51
 52
 53def resolve_direction(d) -> Direction:
 54    """Accept a Direction, an int 0..7, or a name like 'turn_left'."""
 55    if isinstance(d, Direction):
 56        return d
 57    if isinstance(d, int):
 58        return Direction(d % protocol.NUM_MOTORS)
 59    key = str(d).strip().lower()
 60    if key in _DIRECTION_ALIASES:
 61        return _DIRECTION_ALIASES[key]
 62    raise ValueError(f"unknown direction: {d!r}")
 63
 64
 65class Necklace:
 66    """Owns a transport and the current motor state."""
 67
 68    def __init__(self, transport: Optional[Transport] = None,
 69                 port: Optional[str] = None,
 70                 motor_map: Optional[Sequence[int]] = None):
 71        self.transport = transport or open_transport(port)
 72        self.values: List[int] = [0] * protocol.NUM_MOTORS
 73        # motor_map[logical] = physical wire. Defaults to identity; connect()
 74        # passes BUILD_MOTOR_MAP for the mirrored prototype wiring.
 75        if motor_map is None:
 76            self.motor_map: List[int] = list(range(protocol.NUM_MOTORS))
 77        else:
 78            mm = list(motor_map)
 79            if sorted(mm) != list(range(protocol.NUM_MOTORS)):
 80                raise ValueError(
 81                    "motor_map must be a permutation of "
 82                    f"0..{protocol.NUM_MOTORS - 1}, got {motor_map!r}")
 83            self.motor_map = mm
 84
 85    # -- low level ----------------------------------------------------------
 86    # Set commands are fire-and-forget: the transport returns the (usually
 87    # empty) reply frame, which callers may ignore.
 88    def set_motor(self, index: int, value: int) -> bytes:
 89        if not 0 <= index < protocol.NUM_MOTORS:
 90            raise ValueError(
 91                f"motor index {index} out of range (valid 0..{protocol.NUM_MOTORS - 1})")
 92        self.values[index] = protocol.clamp_pwm(value)
 93        return self.transport.send(protocol.set_motor(self.motor_map[index], value))
 94
 95    def set_all(self, value: int) -> bytes:
 96        v = protocol.clamp_pwm(value)
 97        self.values = [v] * protocol.NUM_MOTORS
 98        return self.transport.send(protocol.set_all(value))
 99
100    def set_vector(self, values: Sequence[int]) -> bytes:
101        if len(values) != protocol.NUM_MOTORS:
102            raise ValueError(
103                f"expected {protocol.NUM_MOTORS} values, got {len(values)}")
104        self.values = [protocol.clamp_pwm(v) for v in values]
105        phys = [0] * protocol.NUM_MOTORS
106        for logical, v in enumerate(values):
107            phys[self.motor_map[logical]] = v
108        return self.transport.send(protocol.set_vector(phys))
109
110    def stop(self) -> bytes:
111        self.values = [0] * protocol.NUM_MOTORS
112        return self.transport.send(protocol.stop())
113
114    def ping(self) -> bool:
115        reply = self.transport.send(protocol.ping())
116        frame = protocol.decode_one(reply, start_byte=protocol.START_P2H)
117        return frame is not None and frame[0] == protocol.EVT_PONG
118
119    def version(self) -> str:
120        return f"{protocol.FW_ID} N={protocol.NUM_MOTORS}"
121
122    # -- high level ---------------------------------------------------------
123    def haptic(self, direction, intensity: float = 1.0) -> str:
124        """Vibrate a single direction at 0.0..1.0 intensity (others off)."""
125        d = resolve_direction(direction)
126        vec = [0] * protocol.NUM_MOTORS
127        vec[int(d)] = protocol.intensity_to_pwm(intensity)
128        return self.set_vector(vec)
129
130    def close(self) -> None:
131        try:
132            self.stop()
133        finally:
134            self.transport.close()
135
136    def __enter__(self) -> "Necklace":
137        return self
138
139    def __exit__(self, *exc) -> None:
140        self.close()