"""Everything the node reads from the environment, parsed once at startup. Parsing happens here rather than where each value is used, so an unusable setting stops the node before it joins the dataflow instead of surfacing the first time somebody picks up the gamepad. """ import os from dataclasses import dataclass, field def env_float(name: str, default: float) -> float: """Read a number, naming the variable if it cannot be read as one.""" value = os.environ.get(name, "").strip() if not value: return default try: return float(value) except ValueError: raise ValueError(f"{name} must be a number, not {value!r}") from None def env_int(name: str, default: int) -> int: """Read a whole number, naming the variable if it cannot be read as one.""" value = os.environ.get(name, "").strip() if not value: return default try: return int(value) except ValueError: raise ValueError(f"{name} must be a whole number, not {value!r}") from None @dataclass class ConsoleConfig: """Where the built-in web console listens. The node serves the page and holds the browsers' sockets itself, so there is one address and one port rather than a bridge node and two. `0.0.0.0` is the default because the operator is rarely sitting at the machine driving the arm; bind it to `127.0.0.1` where that is not true. """ host: str = "0.0.0.0" port: int = 7000 @property def enabled(self) -> bool: """Whether to open the port at all. Port 0 means a pad-only dataflow.""" return self.port != 0 @classmethod def from_env(cls) -> "ConsoleConfig": """Build the console's address from the environment.""" port = env_int("WEB_PORT", 7000) if not 0 <= port <= 65535: raise ValueError(f"WEB_PORT must be a port number, not {port}") return cls(host=os.environ.get("WEB_HOST", "").strip() or "0.0.0.0", port=port) @dataclass class DeadmanConfig: """How long a source may go quiet before the arm is told to stop. Four layers, outermost last. `joy_ttl_s` is how long a pad reading stays worth acting on; `jog_keepalive_s` is how often an unchanged velocity is repeated; `jog_ttl_s` is how long this node keeps driving without hearing from the source; and behind all of them sits the arm node's own `VELOCITY_COMMAND_TTL_S`, enforced there whether or not this process is alive. Keep them in that order — each shorter than the one behind it — or a backstop fires before the thing it is backing up. """ jog_ttl_s: float = 0.4 # How often to repeat a velocity that has not changed. A velocity is a # set-point, not a sequence: the arm holds the last one until its own expiry # or a new one, so repeating it 20 times a second only makes the controller # re-plan the same motion 20 times, which an operator feels as stepping. # Low enough that the arm's VELOCITY_COMMAND_TTL_S never fires mid-motion, # and no lower. A change — a new direction, or the stop — never waits for it. # # It is checked on the tick, so the period actually achieved is this rounded # *up* to the next tick: 0.2 with a 50 ms tick repeats every 250 ms, because # the tick at 200 ms lands a hair early and does not qualify. Budget against # that number, not this one. jog_keepalive_s: float = 0.2 # How long the `joy` stream may go quiet before the pad reads as gone. The # pad lives in another node now, and that node dying looks like silence # rather than like an unplugged pad, so silence has to mean the same thing. # `dora-joy` publishes every tick, so it means nothing else. joy_ttl_s: float = 0.2 # How long a gamepad may sit untouched before it gives the arm back, so the # browser can take over without anyone having to press anything. gamepad_idle_release_s: float = 2.0 # How often to ask the arm for the things telemetry does not carry: whether # the tool is engaged, whether freedrive is on, whether it is ready. state_poll_s: float = 0.5 @classmethod def from_env(cls) -> "DeadmanConfig": """Build the timings from the environment.""" return cls( jog_ttl_s=env_float("JOG_TTL_S", 0.4), jog_keepalive_s=env_float("JOG_KEEPALIVE_S", 0.2), joy_ttl_s=env_float("JOY_TTL_S", 0.2), gamepad_idle_release_s=env_float("GAMEPAD_IDLE_RELEASE_S", 2.0), state_poll_s=env_float("STATE_POLL_S", 0.5), ) @dataclass class Config: """The node's settings.""" # Full-deflection speeds. Both sources are analog — a gamepad stick and the # console's on-screen ones — so these are the maximum and anything less # moves proportionally slower. speed_mm_s: float speed_deg_s: float # Stick travel to ignore. A worn stick that does not centre would otherwise # drive the arm across the table on its own. deadzone: float # How much of the curve is cubic: 0 is linear, 1 is fully cubic. Fine # positioning happens near centre, and expo is what buys resolution there. expo: float deadman: DeadmanConfig = field(default_factory=DeadmanConfig) console: ConsoleConfig = field(default_factory=ConsoleConfig) @classmethod def from_env(cls) -> "Config": """Build the configuration from the environment, refusing anything unusable.""" deadzone = env_float("DEADZONE", 0.12) if not 0.0 <= deadzone < 1.0: raise ValueError(f"DEADZONE must be at least 0 and below 1, not {deadzone}") expo = env_float("EXPO", 0.4) if not 0.0 <= expo <= 1.0: raise ValueError(f"EXPO must be between 0 and 1, not {expo}") deadman = DeadmanConfig.from_env() # The layers have to fire outermost last, and each one is a backstop for # the one inside it. Behind both of these sits the arm node's own # VELOCITY_COMMAND_TTL_S, which no setting here can see. if deadman.joy_ttl_s >= deadman.jog_ttl_s: raise ValueError( f"JOY_TTL_S must be below JOG_TTL_S ({deadman.jog_ttl_s}), not {deadman.joy_ttl_s}" ) # The repeat has to be quicker than the arm's own velocity expiry or the # arm stops between repeats. That expiry cannot be read from here, but # JOG_TTL_S is already required to sit below it, so staying under # JOG_TTL_S keeps the repeat under both. if not 0.0 < deadman.jog_keepalive_s < deadman.jog_ttl_s: raise ValueError( f"JOG_KEEPALIVE_S must be above 0 and below JOG_TTL_S ({deadman.jog_ttl_s}), " f"not {deadman.jog_keepalive_s}" ) return cls( speed_mm_s=env_float("SPEED_MM_S", 60.0), speed_deg_s=env_float("SPEED_DEG_S", 30.0), deadzone=deadzone, expo=expo, deadman=deadman, console=ConsoleConfig.from_env(), )