Stawamus Software Map

    n_ticks = int(math.floor(duration / dt + _TIME_EPS)) if duration > 0 else 0
    for _ in range(n_ticks):
        t = deck.now()
        if on_tick is not None:
            on_tick(deck, planner, t)
        parcels = deck.get_parcels()
        goals = planner.update(t, parcels)
        if recorder is not None:
            recorder.capture(goals)
        out = controller.compute(t, parcels, goals)
        deck.set_heights(out.heights, t)
        deck.step(dt)
        log.record(deck, out, parcels, goals, planner)
        if until is not None and until(deck, planner, log):
            log.stopped_by_until = True
            break
    return log.finalize(planner)

@dataclass(frozen=True)
class Measurement:
    """One planar pose observation of a parcel, stamped with its capture time.

    Optional: the observed footprint (length along the measured yaw, width across it; None = the sensor does not
    see the outline, e.g. Mode A / AprilTags) and quality flags (FLAG_*)."""

    id: int
    t_capture: float  # s, deck time at which the image/sample was taken
    x: float
    y: float
    yaw: float  # rad, wrapped to (-pi, pi]
    length: float | None = None  # m, observed footprint along yaw
    width: float | None = None  # m, observed footprint across yaw
    flags: int = 0

    def process(self, image: np.ndarray, t_capture: float, tracks: Iterable = (),
                hints: Iterable[Expect] = ()) -> list[Measurement]:
        """segment -> blobs -> associate on one camera image (`timing` accumulates the wall time per stage)."""
        t0 = time.perf_counter()
        mask = self.segment(image)
        t1 = time.perf_counter()
        self.last_blobs = self.blobs(mask)
        t2 = time.perf_counter()
        meas = self.associate(self.last_blobs, self.candidates(t_capture, tracks, hints), t_capture)
        t3 = time.perf_counter()
        tm = self.timing
        tm["frames"] += 1
        tm["segment_s"] += t1 - t0
        tm["blobs_s"] += t2 - t1
        tm["associate_s"] += t3 - t2
        return meas

"""Per-parcel constant-velocity Kalman filter with timestamped latency compensation.

State s = [x, y, yaw, vx, vy, omega]. Process model: constant velocity driven by
piecewise-constant white acceleration (sigma `accel_noise` m/s² on x/y,
`yaw_accel_noise` rad/s² on yaw). Measurements observe (x, y, yaw) at `t_capture`,
so a delayed measurement corrects the state at the time it was taken and
`estimate(t)` predicts forward to any later time.

    def update(self, measurements: list[Measurement]) -> None:
        """Fuse measurements (any order within the batch); stale or non-finite ones are skipped."""
        valid = [m for m in measurements if m.id in self.registry
                 and all(map(math.isfinite, (m.t_capture, m.x, m.y, m.yaw)))]
        for m in sorted(valid, key=lambda m: m.t_capture):
            tr = self._tracks.get(m.id)
            if tr is None or m.t_capture - tr.t > self.timeout:
                self._tracks[m.id] = self._init_track(m)
                continue
            if m.t_capture - tr.t < 0:
                continue
            if self.fp is not None and m.flags:
                self.stats["partial"] += bool(m.flags & FLAG_PARTIAL)
                self.stats["merged"] += bool(m.flags & FLAG_MERGED)
            new = self._fuse(tr, m)
            if new is None:
                self.stats["reinit"] += 1
            self._tracks[m.id] = self._init_track(m, tr) if new is None else new

@dataclass(frozen=True)
class ParcelInfo:
    """Parcel registry entry: dimensions known from the induction scan."""

    id: int
    length: float
    width: float
    height: float
    mass: float

GoalMode = Literal["velocity", "goto", "hold", "deliver", "twist", "idle"]


@dataclass(frozen=True)
class Goal:
    mode: GoalMode
    # velocity: desired planar velocity (m/s), deck frame
    velocity: tuple[float, float] = (0.0, 0.0)
    # goto / hold: target position (m), deck frame
    target: tuple[float, float] | None = None
    # deliver: destination chute id (DeckInfo.chutes)
    chute_id: int | None = None
    # twist: +1 = counter-clockwise, -1 = clockwise
    twist_dir: int = 1
    # optional per-goal speed cap (m/s); None = control.max_speed
    max_speed: float | None = None

"""High-density flow orchestrator for singulated belt infeeds (stage B, reports/DESIGN/BRIEF.md 4.2).

Pure numpy (planner/ is checked by tests/test_purity.py). `FlowOrchestrator` is the episode's planner in S10
(experiments/s10.py): the harness calls `update(t, parcels) -> goals` every control tick, exactly as it calls
`Planner.update`. The scenario tells it which parcel is on which belt and where (`add`, `observe_belts`: what a
belt PLC knows from its photo-eyes and encoder), reads its belt commands (`belt_commands`: run or stop each belt
at its stop line), reports chute-mouth photo-eyes (`set_chute_blocked`) and removes parcels once they have left

    def add(self, pid: int, infeed: int, chute: int, length: float, width: float, height: float, mass: float,
            t: float, yaw: float | None = None, kind: str | None = None) -> Route:
        """A parcel was loaded on belt `infeed`, bound for `chute` (L >= W as it lies; yaw = its belt yaw; kind:
        'box' or a bag kind, used by the adaptive tubes' classes only)."""
        lane = self._choose_lane(int(infeed), int(chute))
        route = self.route_for(infeed, chute, length, width, height, yaw, lane=lane)
        f = _Flow(int(pid), int(infeed), int(chute), float(length), float(width), float(height), float(mass),
                  float(t), route, lane=lane, kind=kind)
        f.tube = self.prm.tube_of(kind, length, mass)
        f.heavy = self.is_heavy(kind, length, mass)
        if f.heavy and self.prm.heavy_a_lat is not None:
            set_clock(route, self.prm, corner_radius=self.prm.corner_radius, a_lat=self.prm.heavy_a_lat)
        f.hx, f.hy = half_extents(length, width, self.infeeds[int(infeed)].yaw if yaw is None else yaw,
                                  math.radians(self.prm.yaw_allow_deg))
        f.s = -float(self.infeeds[int(infeed)].length)
        self.flows[int(pid)] = f

    # studies/CURRENT: the FOLLOW rule's liveness fix (studies/RLX FOLLOW, rl_routes mode "follow"; no effect without a
    # FOLLOW pair). A FOLLOW leader is not checked against its own follower's reservation while that reservation lies
    # entirely behind the leader's centre on the leader's first-leg line and the leader's remaining route only moves
    # ahead along that line and away from it: the leader advancing can then only widen the gap. The follower keeps its
    # full-envelope check against the leader (both placement orders), so the pair keeps the full crossing envelope as
    # before. Without it, a follower placed first (jam breaker, priority inversion, heavy first) and closer than the
    # full envelope (e.g. after the leader slid or tipped back) holds its leader for good: a wait-for 2-cycle the jam
    # breaker cannot open (the follower's rear is still on the belt). False (default): unchanged.
    follow_live: bool = False

        fields: list[_Field] = []
        smooth: list[ParcelState] = []
        for p in ps:
            st = self._states.setdefault(p.id, _State())
            pf = self._filter(p, st, dt)
            smooth.append(pf)
            goal = goals.get(p.id) or Goal("idle")
            fields.append(self._window_field(pf, goal, st, dt))
            if self.ball_pred:  # the tilt just commanded acts over the next delay (`_predict`)
                st.pred_hist.append(st.tau.copy())
                while len(st.pred_hist) > self.pred_n:
                    st.pred_hist.popleft()
            st.age += dt
        ps = smooth
        self._touching = None
        heights, owner, conflicts = self._compose(ps, fields)
        if self._lost:
            heights = self._lost_apply(ps, heights)
        if self.lost_hold > 0.0:
            self._last_ps = {p.id: p for p in ps}
        heights = self._transition_slew(ps, heights, owner, dt, self._touching)
        if self.rehome_sink_n > 0:
            heights = self._rehome_sink(t, ps, heights)

    lo = float(base) + floor
    span_h = t * (s_hi - s_lo)
    room = info.stroke - lo
    if front_low:
        mid = lo + span_h / 2.0
    elif span_h > room:
        mid = lo + room / 2.0
    elif low:
        mid = lo + span_h / 2.0
    else:
        mid = float(np.clip(centre, lo + span_h / 2.0, info.stroke - span_h / 2.0))
    h0 = mid + t * (s_hi + s_lo) / 2.0 + float(offset)
    raw = h0 - t * s
    return np.clip(raw, 0.0, info.stroke) if clip else raw

        U = np.array([support_mask(p, info, 0.0) for p in ps])  # caps under each footprint
        self._accommodate(ps, fields, U)
        W = np.array([f.w > 0 for f in fields])  # (P, n)
        D = np.array([point_rect_distance(info.column_xy, p) for p in ps])  # (P, n)
        near = np.array([support_mask(p, info, self.keepout) for p in ps])  # caps within keepout of footprint
        conflicts = int(np.count_nonzero(W.sum(axis=0) > 1))
        cand = W.copy()
        for i in range(len(ps)):
            for j in range(len(ps)):
                if i != j:
                    # a column in j's keep-out ring is never given to a parcel farther from it than j
                    cand[i] &= ~(near[j] & (D[j] < D[i]))
        Dm = np.where(cand, D, np.inf)
        best = np.argmin(Dm, axis=0)  # ties -> lowest parcel id (sorted order)
        has = cand.any(axis=0)
        cols = np.nonzero(has)[0]
        H = np.array([f.h for f in fields])
        heights[cols] = H[best[cols], cols]
        ids = np.array([p.id for p in ps])
        owner[cols] = ids[best[cols]]

def tracked_footprint(p: ParcelState, k: float) -> tuple[float, float]:
    """(length, width) to plan with: the tracked footprint (fp_length / fp_width; the induction length / width when
    the tracker does not track it) grown by k x fp_margin on every side."""
    g = 2.0 * float(k) * float(p.fp_margin)
    return ((p.length if p.fp_length is None else p.fp_length) + g,
            (p.width if p.fp_width is None else p.fp_width) + g)

        # rehome_sink_n: a column that starts a re-homing run (a known dead column at or below rest height appearing
        # in the dead-column cache: a StallGuard / encoder homing) rehome_sink_n times within rehome_window s while a
        # parcel's footprint covers it is commanded to 0 (dead-low) until no footprint covers it: under a heavy box it
        # otherwise climbs back into the box after every homing, slips again and the box sinks into a pit of
        # re-homing columns (0 = off).
        self.rehome_sink_n = int(g("rehome_sink_n", 0) or 0)

Modes
  off     the controller as it is.
  dither  a random height offset per cap, N(0, bag_motion_sd) (default 6 mm), new every decision (DECIDE_EVERY = 5
          control ticks: 50 ms at the 100 Hz loop), from a generator seeded with bag_motion_seed; on the 9 x 9 cap
          patch around the bag (policies.bagmotion.patch_cells), as the RLBAG "dither" arms.
  rl      the frozen skill pair of studies/RLBAG/FROZEN.json, deployed as trained: the landing skill
          (policies/frozen/rlbag_land.npz) while the bag's centre is less than LAND_IN = 0.35 m inside the edge of the
          belt it comes from, the moving skill (policies/frozen/rlbag.npz) after that; action = mean + learned std x
          N(0, 1) from a generator seeded with bag_motion_seed (policies.bagmotion.BagMotion, stochastic). The default
          files are checked against FROZEN.json's hashes when loaded.

    def from_params(cls, params: dict | None) -> "BagMotionParams":
        p = dict(params or {})
        raw = p.get("bag_motion", "off")
        mode = "off" if raw is None or raw is False else str(raw).strip().lower()
        if mode not in MODES:
            raise ValueError(f"control.bag_motion {raw!r}: one of {MODES}")

def own_mask(info: DeckInfo, p, owner: np.ndarray, ko: dict[int, np.ndarray], tracked: bool) -> np.ndarray:
    """(n,) the columns the motion may move for bag p: the ones the controller assigned to it (a bag not tracked yet,
    still on its belt: the caps under its estimated footprint that nobody owns), minus every other parcel's keep-out."""
    mine = owner == p.id
    if not mine.any() and not tracked:
        mine = support_mask(p, info, 0.0) & (owner < 0)
    return mine & ~others_keepout(info.n, ko, p.id)

def limit_step(target: np.ndarray, prev: np.ndarray, vel: np.ndarray, dt: float, vmax: float, amax: float
               ) -> tuple[np.ndarray, np.ndarray]:
    """One tick of a velocity- and acceleration-limited move from `prev` (moving at `vel`) toward `target`: the speed
    that can still stop at the target (v dt + v^2 / 2a <= distance), capped at vmax, reached within amax dt of the
    previous speed. Returns (new targets, their speeds)."""
    e = np.asarray(target, float) - prev
    d = np.abs(e)
    v_stop = amax * (np.sqrt(dt * dt + 2.0 * d / amax) - dt)
    v_want = np.sign(e) * np.minimum(vmax, v_stop)
    v0 = np.clip(vel, -vmax, vmax)
    v = np.clip(np.clip(v_want, v0 - amax * dt, v0 + amax * dt), -vmax, vmax)
    return prev + v * dt, v

def wrap_controller(ctl: Any, info: DeckInfo, params: dict | None, *, flow: Any = None,
                    measured: Callable[[], np.ndarray] | None = None, kind_of: Callable[[int], str | None] | None = None
                    ) -> Any:
    """The deck's controller with control.bag_motion applied: `ctl` itself when off (nothing constructed), else a
    BagMotionController around it. The real deck and the simulator both build it this way, from the same control
    params; `flow` (the orchestrator) gives the parcel kinds and the belt observations, `measured` the column heights.
    """
    prm = BagMotionParams.from_params(params)
    if prm.mode == "off":
        return ctl
    src = flow_sources(flow, info) if flow is not None else {"kind_of": kind_of or (lambda pid: None)}
    if kind_of is not None:
        src["kind_of"] = kind_of
    rate = float((params or {}).get("rate", 100.0))
    return BagMotionController(ctl, info, prm, src["kind_of"], src.get("belt_items"), src.get("belts", ()),
                               measured, rate)

    def heights(self, t: float, info, p, is_bag: bool, measured: np.ndarray, commanded: np.ndarray,
                goal: Goal) -> np.ndarray:
        """One control tick for parcel p (the deployed loop)."""
        if self.due(p.id):
            obs = self.observe(t, info, p, is_bag, measured, commanded, goal)
            return self.commit(info, p, commanded, goal, t, self.act(obs))
        return self.hold(info, p, commanded, goal, t)

    def act(self, obs: np.ndarray) -> np.ndarray:
        if self.mlp is None:
            return np.zeros(self.act_dim)
        a = self.mlp(obs)
        if self.std is not None:
            a = np.clip(a + self.std * self.rng.standard_normal(self.act_dim), -1.0, 1.0)
        return a

FROZEN_DIR = Path(__file__).resolve().parents[1] / "policies" / "frozen"
REPO = Path(__file__).resolve().parents[2]
# studies/RLBAG/FROZEN.json: the deployed skill pair (file, sha256[:16]); a default file that differs is refused
FROZEN = {"move": ("rlbag.npz", "0306ab8794216c45"), "land": ("rlbag_land.npz", "d667e57598287ea6")}

ACT_DIMS = {"cap": PATCH * PATCH, "struct": 8, "structd": 9}
DITHER_MAX = 0.012  # m: structd's dither amplitude at a = 1 (sd of the per-cap N(0, 1) draw)
STREAM_CH = 5  # structd: patch channels of the stream block
N_STREAM = 8  # structd: scalars of the stream block
NEAR_M = 0.5  # m: "nearby" other parcels (the count; only these are drawn into the patch)
ACT_DIM = ACT_DIMS["cap"]
MEM_S = 1.0  # s of track history for the mean speed
MOVE_TOL = 0.015  # m: the track "moved" (the stream's jam watch)


def obs_dim(mode: str = "cap") -> int:
    """Patch channels, scalars, last action (struct: + the wave phase's cos / sin; structd: the struct observation +
    the stream block)."""
    if mode == "structd":
        return obs_dim("struct") + 1 + STREAM_CH * PATCH * PATCH + N_STREAM
    return 3 * PATCH * PATCH + N_EXTRA + ACT_DIMS[mode] + (2 if mode == "struct" else 0)

Wire format (little-endian)::

    0xAA 0x55 | seq u16 | t_apply_ms u32 | n u8 | heights u16[n] (0.01 mm) | crc16

The CRC is CRC-16/CCITT-FALSE over every byte after the sync, up to the end of the
heights. A 30-column frame is 71 bytes; at 100 Hz with 8N1 framing (10 bits/byte) that
is 71 kbit/s, about 7.7 % of a 921600 baud link.

`FirmwareModel` is the executable reference for the firmware contract: apply at
`t_apply`, rate-limit each column, ignore stale/duplicate seq (uint16 wrap-aware), and go
flat (toward `rest_height`) once no valid frame has arrived for 200 ms.

    def set_heights(self, heights_m: np.ndarray, t_apply: float) -> None:
        h = np.asarray(heights_m, dtype=float).ravel()
        if h.shape[0] != self.info.n:
            raise ValueError(f"expected {self.info.n} heights, got {h.shape[0]}")
        if not np.all(np.isfinite(h)):
            raise ValueError("heights must be finite")
        t_ms = min(max(_ms(t_apply), 0), MAX_T_APPLY_MS)
        frame = encode_frame(self._seq, t_ms, np.clip(h, 0.0, self.info.stroke))
        self.transport.write(frame)
        self._seq = (self._seq + 1) % SEQ_MOD
        self.frames_sent += 1

    def _pump_camera(self) -> None:
        if self.camera is not None:
            self.tracker.update(self.camera.poll(self.now()))

    def get_parcels(self) -> list[ParcelState]:
        self._pump_camera()
        return self.tracker.estimate(self.now() + self.lookahead)

    python -m link ports                       list USB serial ports (best match first)
    python -m link ping                        firmware version, columns, microsteps
    python -m link regs [--col 0]              TMC5240 register dump, decoded, with SPI-OK / SPI-FAIL
    python -m link init                        (re)initialise the drivers (writes and verifies every register)
    python -m link current 0.8 [--hold 0.8]    run / hold current in A RMS (hard cap: param current_cap_ma)
    python -m link enable | disable            driver enable (EN pin)
    python -m link spin [--revs 1 --rps 1]     turn the motor (works unhomed, max 1 rev per command)
    python -m link setpos 60                   declare the carriage's height in mm (bench guide G)
    python -m link move 80 [--speed 10]        move to a height in mm (homed or setpos first)
    python -m link home [--speed 30]           StallGuard homing into the bottom stop
    python -m link status [--watch 1]          state, flags, positions, counters
    python -m link estop-rearm                 re-initialise the drivers and clear the E-stop latch
    python -m link stream --pattern sine       100 Hz height frames (the RealDeck protocol) with clock sync

CAN id (11 bit) = function << 4 | node (node 1..14: the board's slot ID or its can_node parameter):
    0x01 EVENT      board -> host   urgent event at once, 12 bytes: code, column, channel, 0, deck_ms u32, value i32
    0x02 HEIGHT     host -> board   height frames, segmented
    0x04 COMMAND    host -> board   command frames, segmented
    0x05 REPLY      board -> host   reply frames (replies, telemetry, MOVE_STATS), segmented
    0x07 HEARTBEAT  board -> host   every 100 ms, 24 bytes (layout 1), 32 (layout 2, firmware >= 0.3: + the failed
                                    and passable column masks; decode_heartbeat)

// Wire protocol between the Mac and the column firmware (USB CDC; 921600 baud on a UART).
//
// Two frame kinds share the host -> MCU byte stream:
//
//   height frame (binding, deck/real_deck.py, byte-identical):
//     0xAA 0x55 | seq u16 | t_apply_ms u32 | n u8 | heights u16[n] (0.01 mm) | crc16
//   command frame (bring-up / control channel):
//     0xAA 0x5A | cmd u8 | tag u8 | len u8 (<= 64) | payload[len] | crc16
//
// and one kind flows MCU -> host:
//
//   reply frame:
//     0xAA 0x5B | cmd u8 | tag u8 | status u8 | len u16 | payload[len] | crc16

// Frame scheduler: the firmware side of deck.real_deck.FirmwareModel, decision for decision.
//
//   receive(frame, t_rx): accept iff n matches and seq is newer than the last accepted seq
//                         (uint16 serial arithmetic, RFC 1982); an accepted frame feeds the
//                         watchdog, clears the tripped state and is queued for t_apply.
//   update(now):          if no valid frame for watchdog_ms (or none yet): tripped, pending
//                         dropped, last seq forgotten (a restarted host may resume from any seq),
//                         target = rest. Otherwise the due frame (t_apply <= now) with the latest
//                         t_apply becomes the target (ties: the newest received); all due frames
//                         leave the queue.

        for _ in range(int(round(dt / ts))):
            if motor is None:
                d.ctrl[:n] = act.update(d.time, ts)
            else:
                self._motor_substep(act.update(d.time, ts), ts)
            if pusher:
                d.ctrl[n] = self._pusher_tick(ts)
            if belts:
                self._belt_tick()
            if kinetic:  # friction.kinetic: mj_step split around the per-contact friction (`_kinetic_contacts`)
                mujoco.mj_step1(m, d)
                if d.ncon:
                    self._kinetic_contacts()
                mujoco.mj_step2(m, d)
            else:
                mujoco.mj_step(m, d)
            if flex:
                self._flex_sync()
            if d.ncon:
                pen = self._contact_stats(pen, pairs)
            if pa is not None:
                self._liftoff_tick(ts)
            if not timed or d.time + _TIME_EPS >= sensor.next_capture_time:
                sensor.sample(d.time, self._visible_truth())

    # structure (studies/BAGFIX, docs/research/bag_physics.md): "solid" (default: the RLFLEX model, a 2-layer solid
    # lattice whose modulus `young` sets in-plane, compression AND bending stiffness together: a 30 mm, 5 kPa solid bends
    # at ~1e-2 N m, 10-1000 x stiffer than a bag of garments, and stretches like rubber where real film is inextensible)
    # or "shell": ONE sheet of vertices at the bag's bottom face, a <flex dim=2> with bending only (elastic2d bend,
    # stiffness `bend` N m: the contents' bending, folded garments 1e-5 - 1e-3) and an edge-length equality (the
    # film's inextensibility, E t ~ 8-23 kN/m: no stretch); the whole mass on the sheet; the tracker reports the bag's
    # centre H / 2 above the sheet along its mean normal. With `fill` the content flex covers the content lump and the
    # film flaps around it are a second flex of film bending `bend_film` (PE film ~4e-6 N m), hinged to the lump by
    # point constraints at duplicated seam vertices. Contents compression and the bag's own thickness as a collision
    # volume are not modelled (thin tiles at the sheet).
    sphere_frac: float = 0.5  # tile "sphere": bead radius over the vertex spacing (0.5: neighbours touch)
    structure: str = "solid"
    bend: float = 1.0e-4
    bend_film: float = 4.0e-6

"""Simulated sensing. Mode A (SimSensor): ground-truth planar poses + Gaussian noise, sampled at camera rate with latency.
Mode B (RenderedTagSensor): renders a MuJoCo camera and runs the product AprilTag pipeline (perception/) on each frame.
Mode C (OutlineSensor): a fast statistical overhead outline camera (no rendering) feeding the product outline pipeline's
blob and association code (perception.outline_pipeline); presets OUTLINE_PRESETS, `apply_outline_preset`."""

    def sample(self, t: float, truth: list[tuple[int, float, float, float]]) -> None:
        """Capture (id, x, y, yaw) poses if a capture slot is due at time t."""
        if t + _EPS < self._k / self.rate:
            return
        # Next slot strictly after t: a long gap yields one capture, not a burst.
        self._k = math.floor((t + _EPS) * self.rate) + 1
        if not truth:
            return
        noise = self.rng.standard_normal((len(truth), 3))
        for (pid, x, y, yaw), (nx, ny, nyaw) in zip(truth, noise):
            self._queue.append(Measurement(
                id=int(pid),
                t_capture=float(t),
                x=float(x + self.noise_pos * nx),
                y=float(y + self.noise_pos * ny),
                yaw=_wrap(yaw + self.noise_yaw * nyaw),
            ))

    def poll(self, t: float) -> list[Measurement]:
        """Pop every queued measurement with t_capture + latency <= t, oldest first."""
        out: list[Measurement] = []
        while self._queue and self._queue[0].t_capture + self.latency <= t + _EPS:
            out.append(self._queue.popleft())
        return out

name: current_30v
grid: {nx: 44, ny: 50, pitch: 0.048}
column: {cap_size: 0.0455, cap_thickness: 0.006, stroke: 0.12, rest_height: 0.06, mass: 0.0614, joint_damping: 0.0, armature: 1.237,
  cap_edge: round, cap_edge_size: 0.002, cap_crown: 0.001}
actuator:
  kp: 200000.0
  kv: 1019.2
  max_force: 49.9
  max_speed: 0.4
  latency: 0.005
  max_accel: 4.0
  motor: {model: nema17_40d|tmc2240_30v|tr8x12|120, sizing_factor: 1.0, stall: encoder, thermal: true, hold_current: 0.4,
    idle_current: 0.4, power_down_delay: 0.21, sg_latency: 0.005, sg_home_speed: 0.03, sg_home_dwell: 0.33, detect_delay: 0.002,
    detect_error: 0.00012, recover_time: 0.02}

sensing:
  mode: A
  rate: 200.0
  latency: 0.01
  noise_pos: 0.0015
  noise_yaw_deg: 0.75
  tag_size: 0.035
  image_size: [1280, 960]
tracker: {accel_noise: 3.0, yaw_accel_noise: 30.0, accel_noise_low: null, yaw_accel_noise_low: null, maneuver_rate: 0.5, quiet_rate: 10.0,
  min_meas_pos_sigma: 0.0002, min_meas_yaw_sigma_deg: 0.1, lookahead: null, timeout: 0.25}
control: {rate: 100.0, mu_nominal: 0.23, margin_pitches: 1.0, center_on_support: true, max_speed: 0.4, accel_limit: 1.5, kp_vel: 6.0,
  ki_vel: 3.0, pos_tolerance: 0.008, hold_speed: 0.01, brake_gain: 1.0, cup_depth: 0.02, keepout: 0.015, edge_approach: 0.06,
  twist_angle_deg: 6.0, deliver_walls: true, deliver_wall_height: null, deliver_wall_anchor: false, deliver_walls_under: false,
  deliver_entry_speed: null, accommodate: true, corridor_time: null, stuck_escape: true, mu_static: 0.23, mu_kinetic: 0.16}
planner: {gap_threshold: 0.03, gap_margin: 0.005, replan_period: 0.5, fine_res: null, waypoint_tolerance: 0.015}

  s10:
    control_preset: d2
    tube_calibration: sim/config/calibration/tubes_X2S_T99.json
    installed_footprint: true
    routing: nearest
    flow: {gate_slope: 0.19, gate_stall: 0.5, restart: true, giveup_retry: 8.0, stuck_footprint: true, jam_time: 3.0, route_fallback: true,
      gate_stall_belt: 2.0, homing_aware: true, homing_pause_max: 2.0, dead_clearance: 0.1, clear_stuck: true, clear_expire: true}
    control: {mu_anchor: 0.3, dead_raise_max: 0.02, rehome_sink_n: 0, kp_vel: 9.0, ki_vel: 4.5, cross_track_gain: 6.0, cross_track_max: 0.08,
      tilt_slew: 1.5}

    traffic:
      mode: tm
      k_routes: 6
      detour: 0.15
      commit_window: 2.0
      deck_cap: 9
      cycle_check: true
      trunk_groups: true
      fallback: true
      fb_joint: true
      rest_rule: true
      stuck: {detect: true, contain: true, mouth: true, clear_reach: 0.8, max_tries: 3, tight_box: true}
      soft_late: true
      late_relief: 4.0
      trains: false
      dead_obst: {floor: null, clearance: 0.024, dense: 0.096, refresh: true, confirm: 0.5, divert: true}
      follow_live: true

PROJECT="volve-test-01"
REGION="us-central1"
ZONE="${ZONE:-us-central1-a}"
case "$ZONE" in
  "$REGION"-[a-f]) ;;
  *) echo "ZONE must be a zone of $REGION (got $ZONE)" >&2; exit 2 ;;
esac
KEY="$HOME/Documents/keys/volve-test-rl-runner.json"
GCLOUD="${GCLOUD:-gcloud}"
MAX_VMS="${MAX_VMS:-3}"

    if run_gc compute instances create "$n" --project="$PROJECT" --zone="$z" \
        --machine-type="$MACHINE" \
        --provisioning-model="$PROVISIONING" --max-run-duration="${MAX_RUN_H}h" --instance-termination-action=STOP \
        --image-family=ubuntu-2404-lts-amd64 --image-project=ubuntu-os-cloud \
        --boot-disk-size="${DISK_GB}GB" --boot-disk-type=pd-balanced \
        --no-service-account --no-scopes --shielded-secure-boot \
        --labels=purpose="$PREFIX",owner=abyl,wave="$NAME" \
        --metadata="block-project-ssh-keys=TRUE,enable-oslogin=FALSE,stawamus-workers=$WORKERS,stawamus-commit=$COMMIT" \

SOFTWARE = Path(__file__).resolve().parents[2]
SIM_PACKAGES = {"deck", "experiments", "rl", "viz", "tools", "mujoco"}


def test_software_never_imports_sim():
    bad = []
    for f in (p for p in SOFTWARE.rglob("*.py") if p.name != "test_layering.py"):
        for node in ast.walk(ast.parse(f.read_text(), str(f))):
            names = [a.name for a in node.names] if isinstance(node, ast.Import) else \
                [node.module or ""] if isinstance(node, ast.ImportFrom) and node.level == 0 else []
            bad += [f"{f.relative_to(SOFTWARE)}: {n}" for n in names if n.split(".")[0] in SIM_PACKAGES]
        if '"sim"' in f.read_text() and "sys.path" in f.read_text():
            bad.append(f"{f.relative_to(SOFTWARE)}: puts sim/ on sys.path")
    assert not bad, "software/ imports the simulator:\n" + "\n".join(bad)