API Reference — Palmimo
Import
Section titled “Import”from palmimo_sdk import Palmimo
robot = Palmimo()Constructor
Section titled “Constructor”Palmimo(gait_speed=0.012, step_length=30.0, step_height=30.0, on_step=None, *, fps=60, driver=None, display=None, speaker=None, camera=None, mic=None, auto_wake=True, thermal_clock=None, arm_overload_current=ARM_OVERLOAD_CURRENT, leg_overload_current=LEG_OVERLOAD_CURRENT, rail_undervoltage_v=RAIL_UNDERVOLTAGE_V, flight_recorder=True, incident_dir=None)
The first four arguments are positional-or-keyword; everything after the * is keyword-only.
| Argument | Default | Description |
|---|---|---|
gait_speed |
0.012 |
Gait phase increment per step() call |
step_length |
30.0 |
Walking stride length in mm |
step_height |
30.0 |
Swing-leg lift height in mm for the walk and rotate gaits |
on_step |
None |
Callback invoked with the position dict after every step(), after any driver write |
fps |
60 |
Control rate in cycles per second; must be positive (ValueError otherwise). Converts between seconds and steps in run() and play() |
driver |
None |
ServoDriver; None makes the facade compute-only |
display / speaker / camera / mic |
None |
Peripherals — see Peripherals & Connection |
auto_wake |
True |
Whether connect() runs the wake() glide once a driver is attached |
thermal_clock |
None |
Clock injected into the neck thermal and overload guards in place of time.monotonic; for tests |
arm_overload_current / leg_overload_current |
ARM_OVERLOAD_CURRENT / LEG_OVERLOAD_CURRENT |
See Overload Guard |
rail_undervoltage_v |
RAIL_UNDERVOLTAGE_V |
See Rail Undervoltage Guard |
flight_recorder / incident_dir |
True / None |
See Servo Flight Recorder |
Motion Commands
Section titled “Motion Commands”| Method | Motion Enum | String Aliases |
|---|---|---|
robot.forward() |
FORWARD | “forward” |
robot.backward() |
BACKWARD | “backward” |
robot.strafe_left() |
STRAFE_LEFT | “strafe_left” |
robot.strafe_right() |
STRAFE_RIGHT | “strafe_right” |
robot.rotate_left() |
ROTATE_LEFT | “rotate_left” |
robot.rotate_right() |
ROTATE_RIGHT | “rotate_right” |
robot.creep() |
CREEP | “creep” |
robot.dance() |
DANCE | “dance” |
robot.body_tilt() |
BODY_TILT | “body_tilt”, “tilt” |
robot.pushup() |
PUSHUP | “pushup”, “push_up” |
robot.wave() |
WAVE | “wave” (right leg), “wave_right” (FR), “wave_left” (FL) |
robot.nod() |
NOD | “nod” |
robot.head_shake() |
HEAD_SHAKE | “head_shake”, “shake_head” |
robot.wave_both() |
WAVE_BOTH | “wave_both” |
robot.clap() |
CLAP | “clap” |
robot.bow() |
BOW | “bow” |
robot.stretch() |
STRETCH | “stretch” |
robot.stop() |
IDLE | “idle”, “stop” |
See the motion list for what each motion does. The notes below cover only behavior relevant at the API level.
wave / wave_right and wave_left pick the front-right or front-left leg
— so does the choreography play([("wave_left", s)]) form. Plays its tuned
choreography once and holds neutral after.
nod / head_shake are neck-only one-shots: the first stroke starts on the
very first frame (no wind-up, so they can be fired together with a face
expression), and a head that was looking away when the gesture fires is
blended back to center over the first 0.2 s instead of snapping. Both play
once and hold center after; see “Nod / Head-Shake Gesture Tuning” for the
knobs.
wave_both / clap both first settle into the “beg” stance, which keeps the
mid+rear support quad’s front edge ahead of the center of gravity while the
front legs are airborne. Each plays once and holds neutral after.
bow / stretch are slow posture one-shots: re-selecting the motion after
it finishes replays it from the top; re-selecting it mid-gesture is ignored
(no snap under load). Switching away mid-gesture (e.g. stop()) recenters
the neck along with the legs, and once a gesture has finished, look() owns
the neck again. While a stretch is active the facade applies a quicker leg
profile (Profile_Velocity 120) and restores the driver’s configured default
on exit, mirroring the wave’s automatic per-axis tuning. Tuning knobs
(bow_depth, bow_hold_s, stretch_fold_deg, …) live on the engine —
robot.engine.bow_depth = 26.0 — with deliberately shallow, tip-safe
defaults; see the _BOW_* / _STRETCH_* ClassVars in engine.py.
stretch_fold_deg in particular should be raised gradually on the real
robot: the fold drives the knee past the IK model’s nominal range, whose
physical limit is not modeled.
dance() is the continuous body sway (run it with step() / run()) — the
default in the demo app (runs until stopped, like every motion). For a
finished one-shot performance — sway a few times, hold, then glide home —
use perform_dance() (see “Dance Tuning”); the demo plays this with
--dance-oneshot.
Core Methods
Section titled “Core Methods”robot.step() -> Dict[str, int]
Section titled “robot.step() -> Dict[str, int]”Advance one frame. Returns servo positions as {name: tick_value}.
robot.step_n(n: int) -> list[Dict[str, int]]
Section titled “robot.step_n(n: int) -> list[Dict[str, int]]”Advance N frames. Returns list of position dicts.
robot.set_motion(name: str) -> bool
Section titled “robot.set_motion(name: str) -> bool”Set motion by string name (uses _MOTION_MAP lookup). Returns False (without
changing motion) if name is "nod"/"head_shake" and robot.neck_lock_active
is True (see Neck Thermal Guard); True otherwise.
robot.run(*, seconds: float | None = None, steps: int | None = None) -> Dict[str, int]
Section titled “robot.run(*, seconds: float | None = None, steps: int | None = None) -> Dict[str, int]”Run the current motion for a fixed duration, blocking and real-time paced at
the facade’s fps. Pass exactly one of seconds (converted to
round(seconds * fps) cycles) or steps; both must be non-negative. Each
cycle streams to the driver and fires on_step, exactly like step().
Returns the final servo positions (the current pose if zero cycles ran). For
unpaced loops (simulation, tests, custom timing) use step() / step_n()
instead — they compute as fast as possible.
robot.forward()robot.run(seconds=1.5) # blocks ~1.5s, real-time pacedrobot.stop()Cancellable from another thread — see robot.cancel() below. run()
snapshots the cancel counter the instant it starts, before anything else
(so a cancel() delivered while idle never bleeds into the next call), then
checks the live counter against that snapshot once per frame inside the
shared pacing loop, raising MotionCancelled there if cancel() ran
meanwhile. perform_dance() and play_realtime() share that same pacing
loop and take their own snapshot the same way at their own entry, so a
cancel() interrupts them too — including perform_dance()’s non-paced
end-hold and glide-home sections, which poll the same counter on their own
short interval.
robot.cancel() -> None
Section titled “robot.cancel() -> None”Signal an in-flight run() (or perform_dance() / play_realtime()) to
abort at the next frame boundary, raising MotionCancelled there.
perform_dance() and the agent tool layer’s motion tools glide the legs
onto a stance pose (a short IDLE stream) before the MotionCancelled
propagates, so a cancelled gait doesn’t leave one tripod frozen mid-stride,
bent and loaded, while the caller decides its next move.
The one Palmimo method safe to call from a thread other than the one
driving the robot (the thread inside run()/step()). Every other method
assumes a single caller thread; cancel() is the exception because it only
increments an internal counter under a small lock and touches no other
facade state, so it cannot race with the motion loop’s own reads/writes.
Implemented as a monotonically increasing counter, not a threading.Event
that gets clear()ed: each paced public method (run() / perform_dance()
/ play_realtime()) snapshots the counter the moment it is entered, and
every frame boundary from then on compares the LIVE counter against that
snapshot — an increase means cancel() ran after entry. This closes a race
an Event-based design has: clearing the flag at the pacing loop’s own
entry (to stop a stale cancel bleeding into the next call) can itself race a
concurrent set() from another thread and silently lose it. The counter has
no clear() step to race against, so a cancel() delivered anywhere between
a paced method’s entry and its return is guaranteed to raise
MotionCancelled inside that call.
The one window this still leaves is on the CALLER’s side, strictly before
the paced method has even been entered (e.g. dispatching a run() on a
worker thread and calling cancel() from another thread before the worker
actually starts running it) — a caller racing cancellation that way should
treat its own cancel() request as the authoritative outcome for reporting
purposes, rather than relying on MotionCancelled having been raised.
import threading
robot = Palmimo()robot.forward()t = threading.Thread(target=lambda: robot.run(seconds=10))t.start()...robot.cancel() # from the main thread (or any other)t.join() # the worker thread's run() raised MotionCancelledA caller that wants the robot to land back in a stance after being cancelled
catches MotionCancelled around run() and calls robot.stop() in a
finally — exactly the pattern palmimo_sdk.agent.tools._run_motion already
uses for any other mid-motion failure, so no special handling is needed there.
perform_dance() guarantees this itself: whichever section a cancellation
lands in (the sway, the end-hold, or the glide home), stop() (IDLE) runs
before MotionCancelled propagates, so the facade never stays parked in the
DANCE motion after being cancelled. MotionCancelled is exported from
palmimo_sdk (from palmimo_sdk import MotionCancelled).
robot.cancel_checkpoint() -> int
Section titled “robot.cancel_checkpoint() -> int”Entry point for a hand-rolled control loop that paces its own step() /
run(steps=1) calls instead of a single run(seconds=...). Call once at the
top of the loop; pass the returned value to raise_if_cancelled() on every
iteration so a cancel() delivered between iterations (not just inside a
run() call) is still caught. Follows the same armed-scope rule every paced
method’s own entry does.
robot.raise_if_cancelled(checkpoint: int) -> None
Section titled “robot.raise_if_cancelled(checkpoint: int) -> None”Raise MotionCancelled if cancel() ran since checkpoint (from
cancel_checkpoint()).
robot.look(pitch: float | NeckPitchDegrees | NeckPitchNormalized = 0.0, yaw: float | NeckYawDegrees | NeckYawNormalized = 0.0) -> bool
Section titled “robot.look(pitch: float | NeckPitchDegrees | NeckPitchNormalized = 0.0, yaw: float | NeckYawDegrees | NeckYawNormalized = 0.0) -> bool”Control neck. Returns False (without changing the neck target) while
robot.neck_lock_active is True (see Neck Thermal Guard);
True otherwise. Can be called during any leg motion. Positive pitch tips the chin
DOWN (verified on hardware 2026-07); for yaw, which sign turns which way is not
yet verified on hardware. Each axis accepts three forms, and pitch/yaw each have
their own dedicated value-object types (so passing a yaw value as pitch=
raises TypeError instead of being silently accepted):
- a plain
float/int— normalized [-1, 1]; the historical contract, kept for backward compatibility. Out-of-range values are clamped by the engine, same as always. NeckPitchNormalized(value)/NeckYawNormalized(value)— same normalized [-1, 1] fraction of the neck’s mechanical travel on that axis, explicit about intent. Self-validating: construction raisesValueErroroutside [-1, 1].NeckPitchDegrees(value)/NeckYawDegrees(value)— a real neck angle for that axis. Converted internally by dividing by the axis’s actual mechanical travel. Yaw is symmetric (MotionEngine.NECK_YAW_TRAVEL_DEG, ~26.4° either way). Pitch is now asymmetric: chin-down usesMotionEngine.NECK_PITCH_TRAVEL_DEG(~26.4°), chin-up (negative values) uses the largerMotionEngine.NECK_PITCH_UP_TRAVEL_DEG(~37.7° at the shippedneck_rest_pitch_deg/NECK_PITCH2_SHAREdefaults) — see theneck_pitch2paragraph below for why. Self-validating: construction raisesValueErrorpast the travel for that side — no facade-side clamping, since a value object can only exist in-range once built.
NeckPitchDegrees / NeckYawDegrees / NeckPitchNormalized / NeckYawNormalized
are exported from palmimo_sdk (from palmimo_sdk import NeckPitchDegrees, NeckYawDegrees, NeckPitchNormalized, NeckYawNormalized). The conversion happens
entirely inside Palmimo.look(); MotionEngine.set_neck() still only ever
receives a normalized float.
The neck is body -> neck_pitch1 -> neck_pitch2 -> neck_yaw. look()’s
pitch target is split across both pitch joints (MotionEngine.NECK_PITCH2_SHARE,
0.5 by default) rather than loaded entirely onto neck_pitch1 — the head’s total
pitch angle is unchanged on the chin-down side, but each joint’s own swing
shrinks, which is what fixed an overheating neck_pitch1 on hardware. On the
chin-up side, splitting the pitch target across both joints lets the head reach
FARTHER than neck_pitch1 alone: once neck_pitch1 hits its own raw travel
ceiling, neck_pitch2 keeps going on its own full raw band, so the combined
chin-up reach is neck_pitch1’s remaining room plus neck_pitch2’s whole
travel (hardware observation 2026-09: neck_pitch1 alone stalled well short of
neck_pitch2’s own available room). neck_pitch2 shares neck_pitch1’s tick
direction (MotionEngine.NECK_PITCH2_SIGN = 1, hardware-confirmed 2026-09).
nod() drives neck_pitch1 and head_shake() drives neck_yaw; either way
neck_pitch2 is blended back to its raw-neutral rest point over the same
window that eases a pre-gesture look position home, so the gesture always
ends facing front on both pitch joints instead of stranding neck_pitch2 at
its pre-gesture offset.
robot.look_center() -> None
Section titled “robot.look_center() -> None”Return neck to its front/rest position (the neck_rest_pitch_deg trimmed
center — see “Nod / Head-Shake Gesture Tuning”).
robot.reset() -> None
Section titled “robot.reset() -> None”Reset all state to initial (idle, neutral position).
robot.wake(duration=1.5, start_gain=300) -> None
Section titled “robot.wake(duration=1.5, start_gain=300) -> None”Wake from limp: glide to neutral while ramping leg Position_P_Gain from start_gain up to the default, so the robot softly firms up and rises instead of snapping. The neck is held at full gain the whole time so the head can’t drop. Falls back to a plain glide if the driver can’t ramp gain.
Reads the starting pose per motor, re-asking each motor the
first sweep missed up to three times. If some still do not answer, wake() raises RuntimeError
naming them and writes nothing to the driver, not even a gain: gliding from a
guessed pose would snap those joints. A driver that cannot sense position at all
still gets a single neutral command.
connect() already runs this automatically once every resource is open
(auto_wake=True, the Palmimo(...) default) — call wake() directly only for
a custom duration / start_gain, or to re-wake without a full reconnect.
robot.sleep(duration=1.5, end_gain=300) -> None
Section titled “robot.sleep(duration=1.5, end_gain=300) -> None”The inverse of wake(): glide the legs to neutral while ramping their
Position_P_Gain down to end_gain, then soft-release the neck so the head
eases down. Defaults to where wake() starts, so the two meet — waking neither
jolts nor sags. end_gain must be in (0, 900]: at zero the legs have torque
but no holding force and fold under the body.
The robot ends compliant, not dead — torque stays on at end_gain, and the
mic and camera stay open, so a sleeping robot can still be spoken to. That is the
difference from disconnect(), which parks and closes everything.
Releases any active wave / neck-gesture tuning first (see below). The neck is
held where it is during the glide, and once the head has settled onto the top
plate its goal is rewritten to the pose it actually reached — leaving no stored
error for the next gain write to release as a jolt. The leg ramp starts from the
gain the legs are actually at, so a second sleep() does not raise them first.
If some motor cannot be read after retrying, the glide is skipped (the motors are
named in a warning log) while the gain ramp, neck release and park still run.
Degrades rather than skipping the point: a driver that can’t sense position still gets the gain ramp and the neck release (only the glide needs feedback), and one that can’t ramp gain still gets the glide.
The motion API stays callable while asleep — a forward() issued to a sleeping
robot walks it at end_gain with a limp head. Several things re-raise the gain
besides wake(): a wave’s tuning writes Position_P_Gain to every motor,
and set_p_gain() does the same. None of them jolt the head now that the neck
goal matches where it is, but the robot does silently stiffen. Wake it first.
robot.return_to_neutral(max_steps: int = 120, settle_ticks: int = 2, *, duration: float | None = None) -> None
Section titled “robot.return_to_neutral(max_steps: int = 120, settle_ticks: int = 2, *, duration: float | None = None) -> None”Ease all servos back to neutral. With duration (seconds), a single timed glide;
without it, advances control cycles (up to max_steps, a safety cap) until every
joint has settled within settle_ticks of its target. Releases any active wave
tuning first (see below). The timed glide needs a connected driver and raises
RuntimeError without one; the default path also works compute-only. The timed
glide reads the starting pose per motor, re-asking each motor the first sweep
missed up to three times; if some still do not answer it raises RuntimeError naming them
and writes nothing.
robot.set_p_gain(value: int | None) -> None
Section titled “robot.set_p_gain(value: int | None) -> None”Set Position_P_Gain on all motors (None restores per-motor defaults captured at
connect). No-op compute-only or on drivers without gain control.
robot.read_p_gain() -> dict[str, int] | None
Section titled “robot.read_p_gain() -> dict[str, int] | None”Read live Position_P_Gain back from the servos, one entry per motor. None if no
driver is attached.
robot.set_iir(enabled: bool, alpha: float | None = None) -> None
Section titled “robot.set_iir(enabled: bool, alpha: float | None = None) -> None”Enable/disable the command-side IIR low-pass on the goal stream. alpha in (0, 1]: smaller = heavier smoothing. No-op compute-only or on unsupported drivers.
robot.perform_dance(*, sways=3, end_hold=0.2, settle=1.0) -> Dict[str, int]
Section titled “robot.perform_dance(*, sways=3, end_hold=0.2, settle=1.0) -> Dict[str, int]”Play one finished dance performance, blocking and real-time paced (like run).
Sways the body sways half-cycles using the baked dance feel, softly holds the
final extreme for end_hold s, then glides back to neutral over settle s.
Returns the final positions. The continuous dance() motion is unchanged — this
is the self-terminating gesture (the dance analogue of the wave greeting). See
“Dance Tuning” for the finishing flourish and why it avoids overshoot / a hard stop.
robot.play(routine, fps: int | None = None) -> Generator
Section titled “robot.play(routine, fps: int | None = None) -> Generator”Play a choreography sequence.
- routine:
list[tuple[str, float]]— (motion_name, duration_seconds) - fps: defaults to
None, which falls back to the facade’s ownfps(the constructor’sfps=,60by default) — not a hardcoded60, so aPalmimo(fps=30)routine plays at 30 unless overridden here - Yields
Dict[str, int]servo positions per frame - Special:
"look_around"sweeps the neck sinusoidally (not in_MOTION_MAP). The(str, float)tuple form leavesmotionat itsMotion.IDLEdefault, so the legs stop for the duration of the step. To sweep the neck while the legs keep walking, pass aRoutineStep(motion=..., neck_sweep=True)instead of the tuple.
play() does no pacing. It computes a frame whenever its consumer asks
for one, and every frame it yields has already been streamed to the driver
(it is built on step()). With a driver attached, fps therefore decides
only how far the motion advances per frame — not how fast the servos are
driven. Consuming the generator eagerly (list(robot.play(routine)), or a
for loop that waits for nothing itself) plays the routine at whatever rate
that loop achieves, which is several times real time on a Pi. Nothing fails
and nothing is clamped, because every position stays inside the safe tick
range and only the rate is wrong. Use play_realtime() whenever a driver is
attached; play() is for compute-only work (simulation, tests, custom
timing) and for consumers that pace themselves — the same split as step()
versus run().
robot.play_realtime(routine, fps: int | None = None, callback=None) -> None
Section titled “robot.play_realtime(routine, fps: int | None = None, callback=None) -> None”Blocking version of play() with real-time timing enforcement — the one to
use with a driver attached. Each frame is scheduled against an absolute
deadline by the same pacing loop run() and perform_dance() use, so a
routine takes the wall-clock time its steps ask for.
- fps: same fallback to the facade’s own
fpsasplay() - callback:
Optional[Callable[[dict[str, int]], None]]— called with positions each frame
Keyframe Playback
Section titled “Keyframe Playback”Author a gesture as a list of timed joint poses and play it back, without forking the SDK or writing an engine method.
Keyframe(time: float, joints: Mapping[str, float] = {})
Section titled “Keyframe(time: float, joints: Mapping[str, float] = {})”One timed pose for play_keyframes() / iter_keyframes(), exported from palmimo_sdk.
- time: seconds from the start of playback at which the pose is reached
- joints: servo name (
leg_{1-6}_{yaw,pitch1,pitch2},neck_yaw,neck_pitch1,neck_pitch2) -> angle in degrees from the servo neutral (tick 2048). A joint left out keeps the angle it had at the previous keyframe (or at the start).
The angle is the servo’s own: the same number turns a left and a right leg in
mirrored directions, and neck_pitch1’s front is the rest-trimmed center, not 0.
robot.play_keyframes(keyframes, *, arm_legs=(), speed=1.0, callback=None) -> Dict[str, int]
Section titled “robot.play_keyframes(keyframes, *, arm_legs=(), speed=1.0, callback=None) -> Dict[str, int]”Play a list of Keyframes in real time (blocking), starting from the current
pose. Between keyframes every joint eases in and out with a raised cosine, so
the robot comes to rest at each keyframe. After the last keyframe it glides to
the neutral pose over 1.0 s and the motion is idle, so the next motion starts
from neutral. Returns the final positions.
The whole track is validated before anything moves, and a violation raises
ValueError:
| Rule | Limit |
|---|---|
| Times | finite, after 0, strictly increasing |
| Joint names | the 21 servo names above |
| Leg angles | inside SAFE_MIN_TICK..SAFE_MAX_TICK (about -162 to +163 deg) |
| Neck angles | inside the range the engine’s own neck motions reach: servo neutral ± MotionEngine.NECK_AMPLITUDE_TICKS (about ±26 deg), and for neck_pitch1 also that amplitude around its rest-trimmed front, engine.neck_pitch_center() (about -26 to +11 deg at the default trim) |
| Speed | raised-cosine peak (π/2)·Δangle/Δtime at most MAX_JOINT_SPEED_DEG_S (360 deg/s), the segment from the current pose included |
The ranges are servo limits, not collision limits: a pose can still drive a leg into the body or another leg. Play a new track with the robot lifted first.
- arm_legs: legs (1-6) the track lifts off the ground as arms. The
Overload Guard judges them by
arm_overload_current, as it does a wave’s arm; every other leg keeps the weight-bearing threshold. Only the threshold changes: the wave’s servo tuning is not applied. - speed: playback rate. Every keyframe time is divided by it (
2.0plays twice as fast) and the track is validated at that rate, so a speed that pushes a joint past the speed limit raises before anything moves, naming the fastest speed the track allows. The glide home keeps its own 1.0 s. - callback: as for
play_realtime()
Every frame goes through step(), so the neck thermal guard and the leg
overload and rail guards stay active. A guard trip ends the playback early: the
motion drops to idle and the legs glide toward neutral for 0.5 s, then
play_keyframes() returns normally, so check drain_overload_trips() /
drain_rail_trips() to tell a trip from a finished track. When the thermal lock engages, the neck glides to
center and stays out of the rest of that track; the legs keep playing.
Cancellable from another thread like run(): on cancel() the legs glide to
neutral and MotionCancelled propagates. Any other exception (a callback
raising, for one) also leaves the motion at idle, never on the track.
robot.iter_keyframes(keyframes, *, arm_legs=(), speed=1.0) -> Iterator[Dict[str, int]]
Section titled “robot.iter_keyframes(keyframes, *, arm_legs=(), speed=1.0) -> Iterator[Dict[str, int]]”The unpaced counterpart of play_keyframes(), as play() is to
play_realtime(): it validates on the call and returns an iterator that yields
one frame per step(). On a facade with no driver this is a dry run; with a
driver attached it moves the robot as fast as the loop pulls frames, so use
play_keyframes() there. The iterator ends early if a guard trip or another
motion command takes the engine off the track; closing it early stops the
motion. The track starts from the pose at the call, so pull the frames right
away rather than stepping the robot in between.
robot.read_pose(joints=None) -> Dict[str, float]
Section titled “robot.read_pose(joints=None) -> Dict[str, float]”Read where the joints are now, in Keyframe units: servo name -> degrees from
the servo neutral, ready to pass as a keyframe’s joints. None reads all 21.
A joint that does not answer a sweep and three retries of its own raises
RuntimeError; it is never reported as neutral. Needs a connected driver.
robot.limp(joints) (context manager)
Section titled “robot.limp(joints) (context manager)”Inside the with block, joints have Position_P_Gain 0, so a hand moves them
freely. On entry the motion stops and the joints are commanded to where they
are; on exit they are commanded to where the hand left them before the gain is
restored, so nothing springs back, and the engine is seeded with that pose so
the next play_keyframes() starts from it. If a joint cannot be read, sits
outside the safe tick range, or its profile cannot be switched off on exit, its
gain stays at 0, RuntimeError is raised, and later gain writes (wake(),
set_p_gain(), gestures) skip it until the next connect(). Entry and exit
also read every other joint, because the hold commands all of them; if one that
still has gain does not answer, nothing is written and RuntimeError names it.
On entry the joints then never go limp; on exit they stay limp.
⚠️ Use
limp()only on a lifted robot, with every foot in the air. A limp joint carries no load, and entry and exit re-command every other joint to where it is, which lets a loaded leg or the head sink. Do not callstep()inside the block.
with robot.limp(["leg_6_yaw", "leg_6_pitch1", "leg_6_pitch2"]): input("Pose the front-right leg, then press Enter") pose = robot.read_pose(["leg_6_yaw", "leg_6_pitch1", "leg_6_pitch2"])robot.play_keyframes([Keyframe(1.5, pose)])save_keyframes(path, keyframes) / load_keyframes(path) -> list[Keyframe]
Section titled “save_keyframes(path, keyframes) / load_keyframes(path) -> list[Keyframe]”Keep a track in a JSON file: a list of {"time": seconds, "joints": {name: degrees}},
angles rounded to 0.01 deg. load_keyframes() checks the file’s shape and raises
ValueError otherwise; ranges and speeds are checked when the track is played.
Peripherals & Connection
Section titled “Peripherals & Connection”Palmimo optionally owns the face display, the speaker, the head camera, and the
microphone alongside the servo driver, so the facade stays the single window
onto the robot:
from palmimo_sdk import ( DynamixelDriver, FaceDisplay, HeadCamera, MicStream, Palmimo, Microphone, MicrophoneConfig, Speaker, SpeakerConfig,)
robot = Palmimo( driver=DynamixelDriver(), # port=None → auto-detects the servo bus at connect display=FaceDisplay(), # device_name_hint addresses the card by id, so audio survives a USB # replug renumbering it; unset takes ALSA's default instead. speaker=Speaker(SpeakerConfig(lang="ja", device_name_hint="ReSpeaker")), camera=HeadCamera(), mic=Microphone(MicrophoneConfig(device_name_hint="ReSpeaker")), # or MicStream(device_name_hint=...) # auto_wake=False, # opt out of the connect-time wake glide (sim / calibration))with robot: # enter: connect() (wakes the robot); exit: disconnect() (parks, then closes) robot.set_expression("happy") robot.say("こんにちは") robot.camera.read() # (ok, frame) — reading stays with the caller robot.mic.record(5) # WAV bytes — recording stays with the callermic= accepts either a one-shot Microphone (shells out per recording) or a
shared MicStream (owns one background-read capture, fanned out to
subscribers) — both expose the same open() / close() / is_open contract,
so Palmimo doesn’t branch on which one it holds. See “MicStream” below for
when to reach for the streaming form.
Every peripheral is optional and each is keyword-only. A facade with any subset
of display / speaker / camera / mic but no driver is valid — a face, a voice,
eyes, and/or ears with no legs — and one with all five None is compute-only.
Unlike display/speaker, Palmimo does not add read/record methods of its own;
the camera and the mic are shared hardware resources (one physical device,
multiple readers — the same shape as the servo bus, which step() writes to
and robot.driver.write_positions also reaches directly), so consumers reach
them through the camera / mic properties.
HeadCamera.read() downscales each frame to fit within
HeadCameraConfig.proc_width x proc_height (default 1296x972, applied before
rotation) while keeping the captured frame’s aspect ratio, so a camera whose
native mode is 16:9 1920x1080 yields an undistorted 1296x729 frame. The output
can therefore be smaller than the box on one side (e.g. proc_width=640, proc_height=360 on the default 4:3 capture gives 480x360); a caller that needs
a fixed input size, such as a model input, resizes the frame itself.
Servo incident capture is enabled by default when a driver is attached. See
Servo Flight Recorder for the flight_recorder and
incident_dir constructor options.
Resource reservations
Section titled “Resource reservations”The SDK reserves servo_bus, camera, microphone, speaker, and display
when their backend opens, and holds the reservation until it closes. Standalone
backends use the same reservations as Palmimo. Speaker.say() also reserves
before starting its worker when called without open(). A Microphone keeps
its reservation between recordings, including when it delegates to MicStream.
An in-flight recording keeps an additional reference until its recorder exits,
even if Microphone.close() is called meanwhile.
Compute-only motion reserves nothing.
While another program is connected through Palmimo.connect() with a speaker or microphone, those devices cannot be used standalone either (for example a separate Speaker or MicStream); stop that program first.
On non-POSIX platforms without fcntl, reservations only count references within the process; they do not provide interprocess exclusion.
| API | Contract |
|---|---|
palmimo_sdk.reservation.reserve(*resources, timeout=None) |
Context manager; acquires logical resource names in lexical order and releases all acquired references on exit or acquisition failure. By default, contention fails immediately; a finite non-negative timeout sets a total wait budget in seconds. |
ResourceBusyError |
Exported from palmimo_sdk; attributes resource, holder_pid, holder_user, holder_app. Holder fields are None if metadata is unreadable, invalid, or names a nonexistent process. |
ReservationSetupError |
Exported from palmimo_sdk; unusable reservation storage or permissions. Attributes owner and group identify ownership when available; the error reports the unusable path and existing ownership and permissions. |
Both exceptions derive from palmimo_sdk.reservation.ReservationError, which
inherits directly from Exception. They propagate from camera read() / latest() and
microphone record(); ordinary device-open RuntimeError still produces
(False, None) / None respectively.
PALMIMO_LOCK_DIR overrides the lock directory and is created if absent.
Otherwise, when /run/lock exists, the SDK uses /run/lock/palmimo, creating it
if absent. An unusable override or system directory raises
ReservationSetupError without falling back to a temporary directory. Set
PALMIMO_LOCK_DIR to a location writable by every participating user if the
system lock directory cannot be created. Symlink directories and symlink or
non-regular lock files are rejected before any holder metadata is written.
Only a machine without /run/lock uses palmimo-locks-<uid> beneath its
system temporary directory. SDK-created directories have mode 1777; new
lock files have mode 0666, regardless of umask, so all local users can
participate. Existing files are opened without creation flags and retained
after release. PALMIMO_APP_ID supplies the holder’s app name; otherwise it is the
process name; MCP defaults to palmimo-mcp when the variable is unset. Holder metadata also records the effective user’s name and a UTC
acquired_at timestamp.
Speaker.say() and Palmimo.say() can raise reservation errors synchronously
before returning a speech handle. The servo diagnostic CLI reserves servo_bus
for the entire subcommand and exits with the holder details on contention.
A failed display close clears is_connected while retaining its reservation;
reconnect and successfully disconnect to release it.
A camera close that times out waiting for an in-flight read retains its
reservation until the read returns and closes the capture. A MicStream join
timeout retains it until the capture thread closes its stream. Serial opens
that outlive their connect deadline retain reservations until late cleanup
closes the orphaned handle. These paths log the delayed cleanup; another
process remains blocked while the device is still owned.
LeRobot plugin scripts open LeRobot’s DynamixelMotorsBus directly and therefore do not participate in SDK reservations.
robot.connect() -> Palmimo
Section titled “robot.connect() -> Palmimo”Open whichever resources are attached and return self, in this order: the
display (which also plays its boot-wake animation, mirroring a real startup),
the speaker (probes that its TTS engine, PiperEngine by default, is
available), the camera (opens its cv2.VideoCapture), the mic (a short
capture probe), and last the driver, which turns servo torque on. Raises
RuntimeError when nothing is attached.
connect() and disconnect() exclude each other across threads: a
disconnect() from another thread waits for an in-flight connect() — its
wake glide or its rollback included — to finish before parking, rather than
streaming park commands alongside the glide. An asyncio program that cancels
an await asyncio.to_thread(robot.connect) does not stop that thread, so its
cleanup’s disconnect() relies on this wait. The wait has no bound: the steps
that can stall (a camera open has no timeout; a first-run voice download takes
as long as the network does) all run before the driver arms, so a stalled
connect() keeps disconnect() waiting with torque off, and the steps after
arming (the bus connect, the wake glide) are bounded. That holds only when the
driver was not already connected before connect(): a caller that connects
the driver first (to probe the port, say) has torque on for the whole of
connect(), and so does calling connect() again on a robot that is already
connected. On the main thread, Ctrl+C (SIGINT) does not cut the wait short;
other stop signals are held off only by palmimo_sdk.shutdown.park(), which is
the path to use for a signal-driven stop.
connect() also raises RuntimeError if a disconnect() was running when it
was called or ran during it; everything it opened is closed first.
Because the driver is connected last, a missing servo bus is reported only after the display, speaker, camera and mic have opened; they are then closed before the error is raised.
If a driver is attached and connected, connect() then runs the wake() glide
automatically (limp -> gain-ramped rise to neutral) unless the facade was built
with Palmimo(..., auto_wake=False) — pass that for an immediate connect with
no glide (sim, calibration, tests).
If a later resource — including the wake glide itself — fails, the ones
already opened are closed before the error is re-raised, so a partial failure
releases their reservations when their handles close. This rollback also covers a KeyboardInterrupt
(Ctrl+C) during the wake glide, not just an Exception: it soft-releases the
neck before closing the driver, same as any other rollback, so an interrupted
wake never leaves the head held at full gain with torque about to cut.
DynamixelDriver.connect() and FaceDisplay.connect() fail fast rather than
hanging on a machine with no robot attached: the whole connect sequence (port
detection, port/bus open, and for the Dynamixel bus the motor handshake and
arming) is bounded by a per-driver connect_timeout (10s / 5s defaults, wide
margin over the sub-second real-hardware case). Exceeding it raises
DynamixelConnectTimeoutError / FaceDisplayConnectTimeoutError, which
propagate through robot.connect()’s rollback like any other connect failure
(details below and in palmimo_sdk.io._timeout).
robot.disconnect(*, park=True) -> None
Section titled “robot.disconnect(*, park=True) -> None”Parks the robot, then closes the mic, closes the camera, closes the speaker, returns the display to its neutral IDLE face and closes it, then closes the driver. Idempotent and safe when idle. Peripheral teardown is best-effort.
Parking (skipped when idle): with park=True (the default), eases the legs to
neutral via return_to_neutral(); then, regardless of park, soft-releases
the neck via a dedicated 14-step neck gain-lowering schedule — this
prevents the head dropping when torque cuts. Pass park=False to skip only the
leg return (used by __exit__ after an exception, where streaming more motion
commands could mask the error / be unsafe) — the neck soft-release still runs.
On a connected robot this makes disconnect() take ~2.5s+ (the 14-step neck
ramp alone; park=True adds the leg return on top — return_to_neutral()’s
uncapped-duration path settles onto neutral within max_steps control
cycles, 120 at the default, but that loop has no per-cycle sleep, so those
120 cycles run as fast as the bus accepts the writes rather than over a fixed
2s) — callers on a tight shutdown budget (signal handlers, service stop
timeouts) must allow for it, since a hard kill mid-ramp cuts torque at a low
gain. Ctrl+C does not shorten it: SIGINT is suppressed for the entire body of
disconnect() — the leg return, the neck ramp, every peripheral close, and
the driver disconnect that cuts torque — so no press, however many, aborts
any stage or rushes the head down; with the guard in place, a hung bus write
in that 120-cycle loop is no longer escapable with Ctrl+C either — only
SIGKILL stops it. With park=True (the default) the robot is therefore
uninterruptible for the leg return’s duration plus the ~2.5s neck ramp, not
the ~2.5s alone; with park=False it is uninterruptible for the ~2.5s neck
ramp on its own. Presses after disconnect() returns behave normally.
A BaseException raised by a teardown step — a KeyboardInterrupt from a
mashed Ctrl+C landing in a peripheral close — does not end the teardown early.
The remaining steps still run, the driver disconnect still cuts torque, and only
then is it re-raised (the first one, if several arrive). This mirrors
connect()’s own BaseException rollback: an interrupt is not an Exception,
so guarding each step against Exception alone let it escape with the
peripherals half-closed and the servos still energised.
robot.set_expression(name: str, hold_ms: int = 0) -> str | None
Section titled “robot.set_expression(name: str, hold_ms: int = 0) -> str | None”Show a face expression. name is case-insensitive; firmware aliases resolve
on-device. hold_ms > 0 auto-returns the face to IDLE after that many ms; 0
(default) holds until the next expression. Returns the firmware reply line, or
None when no display is attached (a no-op).
robot.say(text: str, lang: str | None = None) -> SpeechHandle | None
Section titled “robot.say(text: str, lang: str | None = None) -> SpeechHandle | None”Speak text through the attached speaker (non-blocking). lang ("en" /
"ja") overrides the speaker’s default voice. Returns the SpeechHandle
for the queued utterance, or None when no speaker is attached (a no-op).
Speech happens on a resident background worker, so a failure cannot reach you
by raising. Join the handle and read its error to tell speech from silence:
handle = robot.say("こんにちは")if handle is not None: handle.join() if handle.error is not None: ... # the TTS engine never spoke (missing model, wrong audio output, ...)Speaker itself is engine-agnostic — synthesis is delegated to a TtsEngine
(Speaker(config, engine=...), defaulting to PiperEngine, which wraps
piper-plus):
from palmimo_sdk import PiperEngine, Speaker, SpeakerConfig
speaker = Speaker(SpeakerConfig(), engine=PiperEngine(length_scale=1.1, volume=0.8))TtsEngine, TtsVoice, PiperEngine, and OpenAiEngine are exported from
palmimo_sdk itself (implemented in palmimo_sdk.io.tts); see PiperEngine’s
piper-plus-specific settings (voice models, data_dir, length_scale,
volume) there. PiperEngine downloads the voice Speaker opens with, at
open() and nowhere else, into ~/.cache/palmimo/models/piper/<voice>/ unless
data_dir names another root (one directory per voice inside it) — see
First-Time Setup for Voice Output.
OpenAiEngine synthesizes off the robot instead. It is an alternative voice,
not a faster one: on a Pi 5, piper takes 0.19 s per utterance against this
engine’s 1.57 s (median of three Japanese sentences, three passes each), since
the round trip dominates. What it buys is how it sounds, plus instructions.
The costs are that latency, an API key, per-utterance billing, and a hard
network dependency — a robot that cannot reach the internet cannot speak with
this engine, where the local one only sounds worse. Speaker therefore keeps
PiperEngine as its default and this is opt-in.
from palmimo_sdk import OpenAiEngine, Speaker, SpeakerConfig
speaker = Speaker(SpeakerConfig(), engine=OpenAiEngine(voice="coral", speed=1.1))Needs OPENAI_API_KEY in the environment (preflight() checks it and nothing
else — it does not spend a request validating the key). There is no model to
download, so load_voice() is cheap and startup costs nothing. instructions steers
delivery in plain language and has no counterpart in the local engines;
volume scales the returned samples (the API has no gain parameter) and is the
one setting that needs numpy.
robot.disconnect() (and closing the speaker directly) truncates any speech
in progress or still queued instead of waiting for it – disconnect must
stay prompt. join() a handle first if it needs to finish speaking before
disconnecting.
robot.stop_speech() -> None
Section titled “robot.stop_speech() -> None”Stop any in-flight speech immediately (barge-in). Idempotent, and a no-op
when no speaker is attached (same shape as say) or nothing is speaking.
Unlike cancel() (motion only), this reaches speech directly — a
background utterance started by say() has no running tool for a plain
cancel() to interrupt.
palmimo_sdk.shutdown
Section titled “palmimo_sdk.shutdown”Where every stop path ends, and the three ways a signal reaches it. Import as
from palmimo_sdk import shutdown.
| Name | Purpose |
|---|---|
shutdown.park(robot, *, legs=True, out=None) |
Release servo torque, uninterruptibly, reporting both ends of the park on stderr. Safe to call more than once and on a robot that was never connected, so it belongs in a finally without a guard. |
shutdown.park_async(robot, ...) |
The same, for a caller that must not block its event loop: the park runs in a worker thread while the loop’s own thread holds the signal protection. |
shutdown.StopRequest |
“A stop was asked for”, and how long ago. The one piece of shutdown state; delivery is the caller’s. |
shutdown.signals_ignored(*signums) |
Discard signals for a block — what makes the park the one stretch a repeat signal cannot cut short. |
shutdown.interrupt_on_signals(stop, ...) |
Delivery for a program with no loop of its own: convert the signal to KeyboardInterrupt so its own finally runs. |
shutdown.stop_flag_on_signals(stop, ...) |
Delivery for a loop that blocks synchronously: set a flag it polls, and hand the disposition back on a second signal. |
shutdown.loop_stop_on_signals(stop, on_stop, ...) |
Delivery for a loop that awaits while idle: run on_stop as an ordinary callback inside the event loop. |
shutdown.STOP_SIGNALS / shutdown.TERMINATING_SIGNALS |
SIGINT + SIGTERM + SIGHUP, and the two whose default disposition kills without unwinding. SIGHUP is included because a dropped ssh session delivers it. |
Which delivery to use is decided by one question — does this loop await while idle? — set out in explanation/shutdown.md.
Nothing here promises the servos end up released: SIGKILL and power loss run no
code, and park() takes seconds, so anything that stops waiting mid-park kills
the process with torque on. What it removes is the case where the code that
would have released torque never ran at all.
MicStream
Section titled “MicStream”palmimo_sdk.io.MicStream is a shared, background-owned microphone capture —
useful when several consumers (an always-on VAD, an on-demand ASR capture, a
denoiser, …) need the same raw audio but a physical mic can only be opened
by one stream at a time. It opens the device once, reads it continuously on a
background thread, and fans each chunk out to every subscriber. Format
defaults to 16k / mono / int16; sample_rate can be changed, but the default
processors (an EchoCanceller) is fixed at 16k — a mismatch is rejected in
open() itself (see the processors row below), not silently on every
chunk. Chunks default to 512 samples (32 ms at 16k, matching Silero VAD v5’s
window). Requires the palmimo-sdk[voice] extra (sounddevice,
lazy-imported at open()).
The default processors also means the first open() in a fresh environment
needs network access, to auto-download the DTLN-AEC model (see
palmimo_sdk.audio.dtln.ensure_dtln_models below). For offline operation,
either call ensure_dtln_models() once ahead of time (while online) so the
model is cached locally, or place the model files manually and pass their
directory via processors=[EchoCanceller(model_dir=...)].
By default every chunk has the loudspeaker cancelled out before it reaches
subscribers — open() builds a fresh EchoCanceller (see “palmimo_sdk.audio”
below) unless you pass processors yourself:
from palmimo_sdk import MicStream
stream = MicStream(device_key="respeaker") # processors=[EchoCanceller()] by defaultstream.open()with stream.stream() as chunks: for clean_chunk in chunks: ... # already echo-cancelledEchoCanceller requires a capture device that exposes the loudspeaker
loopback as an extra channel (ReSpeaker-family XMOS arrays do this natively —
6 channels by default). On a device without enough channels, open() fails;
pass processors=[Denoiser()] (GTCRN noise-only, mono) or processors=[]
(raw) instead.
| Constructor arg | Default | Description |
|---|---|---|
sample_rate |
16000 | Capture rate in Hz |
blocksize |
512 | Samples per chunk read from the device |
device_name_hint |
None |
Case-insensitive substring match against input device names (e.g. "ReSpeaker"); None uses the default input device |
device_key |
"default" |
Identity used to coordinate with a Microphone sharing the same physical mic (see below) |
input_stream_factory |
None |
Test seam; None lazily builds a real sounddevice.InputStream |
processors |
None |
Chain of AudioProcessor each chunk is cascaded through before dispatch (WAV in/out, in order). None means “echo-cancel by default” — a fresh EchoCanceller is built at open(). Pass processors=[] for raw, unprocessed audio. A processor that raises has its chunk discarded (logged), not dispatched unprocessed. open() rejects (before touching the device) any processor whose sample_rate attribute disagrees with this stream’s own sample_rate, or whose capture_channels (see AudioProcessor below) is declared by anything but the first processor in the chain |
With processors in play, a dispatched chunk’s length no longer matches
blocksize — a processor (e.g. EchoCanceller / Denoiser, which buffer
internally) may emit less than it was given, or nothing for a given read.
Silero VAD’s “one chunk = one window” assumption only holds for
processors=[].
| Method / property | Description |
|---|---|
stream.open() |
Open the mic and start the background capture thread; idempotent |
stream.close() |
Signal the capture thread to stop and wait (bounded); idempotent. The capture thread itself owns releasing the device (even across a slow/wedged stop) — see below |
stream.is_open |
bool |
stream.subscribe(maxsize=64) -> Subscription |
Register a new subscription that receives every future chunk (bounded, drop-oldest queue; Subscription.dropped counts losses). A reopened stream starts with zero subscribers — subscribe again after a close()/open() cycle |
stream.stream(maxsize=64) -> Subscription |
Sugar for stream.subscribe(maxsize) — yields chunks until close(). Returns the Subscription itself (still directly iterable and usable as a context manager), so with stream.stream() as chunks: ... unsubscribes automatically on exit |
stream.record(seconds) -> bytes | None |
Capture seconds of audio via a temporary subscription; returns WAV bytes, or None on failure |
Every dispatched chunk is read-only (chunk.flags.writeable is False) — the
same array object is fanned out to every subscriber, so call chunk.copy()
first if you need to modify it in place.
Also usable as a context manager (with MicStream(...) as stream:).
Stream ownership on close: close() never touches the underlying device
handle itself — the background capture thread that opened it owns
stopping/closing it, in its own finally, however it exits. If close()’s
join times out (the thread is wedged in a blocking read, e.g. a yanked USB
mic), a warning is logged and subscribers are notified immediately, but the
device is still released — whenever that blocked read eventually returns or
raises, even long after close() gave up waiting.
device_key: MicStream and Microphone cooperate on one physical mic
by sharing the same device_key string — when a MicStream is open, a
same-keyed Microphone.record() delegates to it instead of shelling out
(the two can’t both hold the device). This is a same-process convention, not
device-identity resolution: ALSA device strings and sounddevice device
names/indices can’t be machine-matched against each other, so two different
keys are always treated as different mics even if they’re the same hardware.
palmimo_sdk.audio
Section titled “palmimo_sdk.audio”palmimo_sdk.audio is the pure sample-processing layer next to io: io
owns hardware resources and hands back raw bytes / arrays, audio only
transforms those samples — no sounddevice, no shelling out, no device
handles. It wraps sherpa-onnx’s GTCRN noise-removal model (>=1.13.0) and the
DTLN-AEC echo-cancellation model pair (via ai-edge-litert), both behind the
palmimo-sdk[voice] extra, lazy-imported at construction.
AudioProcessor
Section titled “AudioProcessor”AudioProcessor is a Protocol — the DI seam MicStream / Microphone push
captured audio through: process(wav: bytes) -> bytes, WAV bytes in, mono
16k/S16_LE WAV bytes out (the input may be multi-channel — see
capture_channels below — but the output is always mono). Implementations
may be stateful (buffer internally), so output may be shorter than input, or
empty (a valid 0-frame WAV) — treat it as “what went in eventually comes
out”, not “N frames in, N frames out”. Any number can be cascaded in order
(each one’s output feeds the next).
| Name | Description |
|---|---|
AudioProcessor |
Protocol: process(wav: bytes) -> bytes. By convention (not part of the Protocol itself) an implementation may expose a read-only sample_rate: int attribute naming the rate it requires; Denoiser, ClipDenoiser, and EchoCanceller all do, and MicStream.open() checks it. Likewise, an implementation may expose a read-only capture_channels: int naming the number of interleaved channels it needs in its input WAV (absent means 1/mono); EchoCanceller is the only one that does. Only the first processor in a processors chain may declare capture_channels > 1 — MicStream.open() checks this too |
EchoCanceller(model_dir=None, *, size=256, channels=6, near_channel=1, reference_channel=5, far_end_silence_db=-60.0, hangover_hops=8) |
AudioProcessor implementation wrapping DTLN-AEC (palmimo_sdk.audio.dtln) — cancels the loudspeaker out of a mic channel, given a reference channel from the same read. Resolves/downloads both DTLN-AEC models and builds the interpreters in __init__ (fail-fast). Cancels the loudspeaker only — servo/motor noise has no reference channel and passes through untouched. Requires a capture device that exposes the loudspeaker loopback as an extra channel (ReSpeaker-family XMOS arrays do this natively); on a device without enough channels, MicStream.open() fails — use Denoiser() or processors=[] instead |
Denoiser(model_path=None, *, num_threads=1) |
AudioProcessor implementation wrapping StreamingDenoiser — for a continuous stream (MicStream). Resolves/downloads the GTCRN model and builds the streaming denoiser in __init__ (fail-fast). Stateful and never flushed by its AudioProcessor callers: using it on a self-contained clip cuts off the tail and bleeds residue into the next clip — use ClipDenoiser for that instead |
ClipDenoiser(model_path=None, *, num_threads=1) |
AudioProcessor implementation wrapping the offline SpeechDenoiser — for a complete, self-contained clip (Microphone’s recordings). Each process() call denoises the whole WAV it’s given in one stateless shot; no tail cut off, no residue between clips |
int16_to_wav(data, sample_rate, channels=1) -> bytes |
Encode an int16 ndarray as WAV bytes — shape (frames,) for mono, (frames, channels) for multi-channel |
wav_to_int16(wav) -> tuple[ndarray, sample_rate] |
Decode mono/16-bit WAV bytes; raises ValueError for stereo / non-16-bit input |
wav_to_int16_multi(wav) -> tuple[ndarray, sample_rate] |
Decode 16-bit WAV bytes of any channel count to a (frames, channels) ndarray |
MicStream’s default processors=[EchoCanceller()] is what gives you
echo-cancelled audio out of the box (see “MicStream” above); Microphone
defaults to processors=() (raw) and denoises only if you pass
processors=[ClipDenoiser()] explicitly (its public contract is to stay
dependency-free without the voice extra) — ClipDenoiser, not Denoiser:
Microphone calls process() once per complete recording, and Denoiser’s
streaming engine is never flushed there, so it would cut off the tail of
every recording. EchoCanceller is not offered to Microphone either — it
needs a stateful stream and a guaranteed reference channel that Microphone’s
clip-capture path doesn’t provide.
Low-level: SpeechDenoiser / StreamingDenoiser
Section titled “Low-level: SpeechDenoiser / StreamingDenoiser”| Name | Description |
|---|---|
SpeechDenoiser(model_path, *, sample_rate=16000, num_threads=1) |
Offline (batch) denoiser. .denoise(samples_int16) / .denoise_wav(wav_bytes) -> bytes — one-shot use over an already-captured clip |
StreamingDenoiser(model_path, *, sample_rate=16000, num_threads=1) |
Streaming denoiser for a continuous mic feed. .process(chunk_int16), .flush(), .stream(chunks) -> Iterator |
ensure_denoise_model(path=None) -> str |
Resolve the GTCRN model path, auto-downloading it (sha256-verified) to ~/.cache/palmimo/models (or $XDG_CACHE_HOME/palmimo/models) if missing |
Denoiser is the recommended way to plug GTCRN denoising into a MicStream /
Microphone; reach for StreamingDenoiser directly only when you need the
int16-chunk-level API instead of the AudioProcessor WAV contract — e.g.
sitting it directly on a raw (processors=[]) MicStream’s chunk feed:
from palmimo_sdk import MicStreamfrom palmimo_sdk.audio import StreamingDenoiser, ensure_denoise_model
stream = MicStream(device_key="default", processors=[]) # raw; denoise manually belowstream.open()denoiser = StreamingDenoiser(ensure_denoise_model())
# `stream.stream()` (MicStream.stream, the chunk source) feeds# `denoiser.stream(...)` (StreamingDenoiser.stream, the chunk transform) —# same method name, unrelated methods on unrelated classes.with stream.stream() as chunks: for clean_chunk in denoiser.stream(chunks): ... # feed clean_chunk to VAD / ASRProperties
Section titled “Properties”| Property | Type | Description |
|---|---|---|
robot.positions |
Dict[str, int] |
Current servo positions |
robot.motion |
str |
Current motion name as lowercase string |
robot.engine |
MotionEngine |
Direct access to engine (advanced) |
robot.p_gain |
int | None |
Effective Position_P_Gain (set value, else default) |
robot.iir |
tuple[bool, float] | None |
IIR (enabled, alpha); None if unsupported |
robot.fps |
int |
Control rate in cycles per second |
robot.dt |
float |
Seconds per control cycle (1 / fps) |
robot.driver |
ServoDriver | None |
Attached servo driver; None in compute-only mode |
robot.display |
FaceDisplay | None |
Attached face display; None when no face is wired |
robot.speaker |
Speaker | None |
Attached speaker; None when no voice is wired |
robot.camera |
HeadCamera | None |
Attached head camera; None when no camera is wired |
robot.mic |
Microphone | MicStream | None |
Attached microphone (one-shot or streaming); None when no mic is wired |
robot.incident_dir |
Path | None |
Recording destination; see Servo Flight Recorder |
robot.last_incident_path |
Path | None |
Most recently written incident; see Servo Flight Recorder |
robot.is_connected |
bool |
Whether a driver is attached and currently connected |
robot.has_connectable_resource |
bool |
Whether any of driver / display / speaker / camera / mic is attached (what connect() requires) |
robot.neck_thermal_state |
NeckThermalState |
Neck thermal guard state: NORMAL / WARM / HOT / UNMONITORED (see Neck Thermal Guard) |
robot.neck_lock_active |
bool |
Whether the HOT lockout is currently in force — see below; check this, not neck_thermal_state == "hot" |
robot.neck_temperature_c |
float | None |
Highest of the three neck motors’ last-read temperature, in °C; None while UNMONITORED |
robot.neck_telemetry_age_s |
float | None |
Seconds since the last COMPLETE neck telemetry sweep; None if none has ever completed |
Neck Thermal Guard
Section titled “Neck Thermal Guard”The neck holds the head up continuously, unlike a leg that only bears weight during its stance phase, so it has been observed latching an Overheating HW error (bit 2) and cutting torque at 71°C after a sustained holding load — one degree past the XC330’s own Temperature Limit default of 70°C, with no software warning beforehand. The legs have no temperature guard; their current is watched by the Overload Guard below.
Scope: the guard only reaches callers that go through Palmimo.step() —
run() / play(), the MCP server, and the agent tool layer
(palmimo_sdk/agent/). The LeRobot teleop integration
(integrations/lerobot/) drives the engine directly and is NOT covered —
see that package’s palmimo.py module docstring.
Palmimo.step() polls ServoDriver.read_telemetry() for the three neck
motors at most once a second (a full sweep every frame would cost bus time
the position write already needs) and classifies the highest reading:
| State | Trigger | Effect |
|---|---|---|
NORMAL |
< 55°C | None |
WARM |
>= 55°C | Logged once on entry; motion is unaffected |
HOT |
>= 62°C | The neck is glided to center every frame, overriding look()’s target AND any motion’s own neck keyframes (NOD/HEAD_SHAKE, BOW/STRETCH); look(), nod(), head_shake(), and set_motion("nod"/"head_shake") return False instead of taking effect, until the neck cools to <= 55°C |
UNMONITORED |
No driver, the driver doesn’t implement read_telemetry() or doesn’t carry the neck motors, or the last COMPLETE sweep is older than NECK_STALE_S (10s) |
neck_temperature_c is None. neck_lock_active stays True if the guard was HOT when it lost telemetry (see below) |
HOT is latched (hysteresis: released only at <= 55°C, not just below its
own 62°C entry threshold) — without it, a neck hovering near the boundary
would flap the lockout every poll.
A neck motor missing from a sweep (bus didn’t answer) is unknown, not
healthy — the guard keeps its last judgement rather than guessing, whether
one motor or all three are missing, until that judgement is older than
NECK_STALE_S, at which point neck_thermal_state falls back to
UNMONITORED rather than trust a reading that may no longer reflect reality
(neck_telemetry_age_s exposes how stale the current judgement is before
that happens). The one case that reports UNMONITORED from a missing
reading immediately, with no staleness wait, is the bootstrap one: no
judgement has ever succeeded yet.
neck_lock_active can stay True after neck_thermal_state falls to
UNMONITORED. A neck that was HOT and then loses telemetry (one motor
stops answering, then all of them go stale) is still locked in place —
reporting the state as merely “unmonitored” while quietly dropping the
lockout would let a caller drive an unobserved, possibly still-hot neck.
The lock only clears when the driver disconnects (nothing left to hold
centered) or a COMPLETE sweep reads at or below 55°C. Every call site that
decides whether to force the neck to center or reject a gesture checks
neck_lock_active, not neck_thermal_state == "hot".
While the lock is active, step() re-logs (at INFO, every 30s) how long
look()/NOD/HEAD_SHAKE have been ignored, so a long stuck-hot period is
traceable in the log. LookTool/NodTool/HeadShakeTool (the agent tool
layer) report the rejection in their result text (e.g. “neck is cooling down
(63°C), look ignored”) instead of claiming the gesture happened.
Overload Guard
Section titled “Overload Guard”A leg axis that is held or blocked keeps drawing current until the servo latches
its own Overload HW error and drops torque. Palmimo.step() samples the 18 leg
axes (the neck is not watched) and stops the motion when one holds too much
current for too long.
robot = Palmimo( driver=driver, arm_overload_current=700, # raw |current|, raised arm of wave / wave_both / clap / play_keyframes(arm_legs=...) leg_overload_current=1200, # raw |current|, every other leg axis)| Argument | Default | Meaning |
|---|---|---|
arm_overload_current |
ARM_OVERLOAD_CURRENT (700) |
Threshold for the arm channel; None disables it; <= 0 raises ValueError |
leg_overload_current |
LEG_OVERLOAD_CURRENT (1200) |
Threshold for the leg channel; None disables it; <= 0 raises ValueError |
Both constants and OVERLOAD_CONSECUTIVE are exported from palmimo_sdk.
Thresholds are raw Present_Current register units compared by magnitude. With
both channels and the Rail Undervoltage Guard
disabled, step() never reads the leg telemetry.
Judgement. An axis trips when abs(current) >= threshold on
OVERLOAD_CONSECUTIVE (3) samples in a row; one sample below the threshold
zeroes that axis’s count. An axis missing from a sweep (silent or unreached)
neither advances nor resets its count. Samples 0.2 s or more apart are not
consecutive: after a pause in stepping, or a run of failed reads, every count
restarts. A trip clears every count, and when
several axes reach the count together the first in bus order is reported. The
arm channel covers the three axes of the raised leg (wave_leg for wave, legs
3 and 6 for wave_both and clap); the leg channel covers the remaining
axes. The counts restart whenever the motion or its arm leg changes, and on
connect().
Polling. The clock is the one injected as thermal_clock. At most every
1/30 s, step() makes one read_telemetry() call for all 18 leg axes.
Reaction.
| Channel | On trip |
|---|---|
| leg | The motion drops to IDLE; gains are untouched |
| arm | The motion drops to IDLE and the three arm axes are held at Position_P_Gain 100 with torque on (soft return) |
The frame that trips is computed and passed to on_step, but not written to the
driver. IDLE, NOD and HEAD_SHAKE move no leg, so a leg trip under them
stops nothing: the motion carries on and the frame is written. It is recorded
once, and not again until every watched axis has read below the threshold or a
leg motion has started. A soft-returning arm is watched by the leg channel like
any other leg, so strain on it at the soft gain is recorded in that same way
and changes neither its gain nor its torque. The soft return
ends, and the arm’s gain goes back to the default, when all three arm axes have
read within 60 ticks of neutral on 5 consecutive polls, or when any motion
other than IDLE, NOD or HEAD_SHAKE starts, or on connect() or
wake(). sleep() ends it too, but ramps those three axes from 100 to
end_gain alongside the other legs instead of restoring the default first. A
poll that cannot read every arm position does not end it. If the
gain cannot be written (the driver raises, including NotImplementedError),
torque is cut on those three axes only with ServoDriver.set_torque_enabled()
and stays off until the driver reconnects; a driver without that either only
records the trip.
Reading trips.
| Member | Returns |
|---|---|
robot.drain_overload_trips() |
list[OverloadTrip], oldest first; the call empties the queue, which keeps the newest 64 |
robot.last_overload_trip |
The latest OverloadTrip or None; survives connect() |
OverloadTrip is a frozen dataclass: motor, current (signed raw value),
threshold, samples, motion (lowercase name) and channel ("arm" or
"leg"). Each trip is also logged as one warning.
A driver whose read_telemetry() raises NotImplementedError, or KeyError
because it does not carry all 18 leg motors, disables this guard and the rail
guard with one warning until the next connect(). Any other read failure is logged at most every 5 s,
keeps the counts and does not stop the motion.
Scope. The same as the neck guard: only Palmimo.step() callers are
covered. wake(), sleep() and return_to_neutral() write the driver
directly, and the LeRobot teleop integration drives the engine directly, so the
guard does not watch them.
Rail Undervoltage Guard
Section titled “Rail Undervoltage Guard”Resistance in the supply path pulls every axis’s input voltage down together
when the summed current is high. A sag that persists makes the servos latch an
Input Voltage error and drop torque, while each axis’s own current stays inside
its limit, which the Overload Guard cannot see.
Palmimo.step() reads telemetry.voltage from the same leg sweep the overload
guard takes (no extra bus read) and stops the motion when the rail stays low.
robot = Palmimo( driver=driver, rail_undervoltage_v=3.8, # volts, lowest leg-axis input voltage)| Argument | Default | Meaning |
|---|---|---|
rail_undervoltage_v |
RAIL_UNDERVOLTAGE_V (3.8) |
Threshold in volts; None disables the guard; <= 0 raises ValueError |
RAIL_UNDERVOLTAGE_V, RAIL_CONSECUTIVE and RailTrip are exported from
palmimo_sdk. The default sits above the servo’s Min Voltage Limit of 3.5 V.
Across a full-motion measurement of an assembled robot, the longest stretch
below 3.8 V during normal walking was 67 ms; RAIL_CONSECUTIVE (6 sweeps) takes
at least 200 ms at the 30 Hz poll, three times that. The poll runs on a frame of
step(), so at the default 60 fps a sweep can slip from every second frame to
every third, and six sweeps can take up to about 300 ms. How quickly a servo latches under a sag
has not been measured, so the guard is not guaranteed to fire before the latch.
Judgement. Every axis measures the same rail, so the lowest voltage among
the 18 leg axes in a sweep is one signal, with one count rather than one per
axis. The rail trips when that value is below the threshold on RAIL_CONSECUTIVE
sweeps in a row; one sweep at or above the threshold zeroes the count. A sweep
with no leg voltage in it neither advances nor resets the count. A trip clears
the count, and when several axes read the same lowest value the first in bus
order is reported. The count belongs to the rail, not to a motion: it survives a
motion change and restarts on connect(), after a trip, and when two sweeps are
0.2 s or more apart (a pause in stepping, or a run of failed reads). Polling is the
overload guard’s: the injected thermal_clock, at most every 1/30 s.
Reaction. The motion drops to IDLE; torque and gains are untouched. The
frame that trips is computed and passed to on_step, but not written to the
driver. IDLE, NOD and HEAD_SHAKE move no leg, so a trip under them stops
nothing: it is recorded once, and not again until a sweep has read at or above
the threshold or a leg motion has started. When an overload trip and a rail trip
land on the same sweep, both are recorded and the motion is stopped once.
Reading trips.
| Member | Returns |
|---|---|
robot.drain_rail_trips() |
list[RailTrip], oldest first; the call empties the queue, which keeps the newest 64 |
robot.last_rail_trip |
The latest RailTrip or None; survives connect() |
RailTrip is a frozen dataclass: voltage (the sweep’s lowest, in volts),
motor (the axis that read it), threshold, samples and motion (lowercase
name). Each trip is also logged as one warning.
Scope. The same as the overload guard: only Palmimo.step() callers are
covered.
Servo Flight Recorder
Section titled “Servo Flight Recorder”Servo incident recording is part of the Palmimo facade. Its design and timing
budget are explained in Servo Flight Recorder.
The recorder implementation itself is not a public API.
Configuration and destination
Section titled “Configuration and destination”Recording-related constructor parameters (other parameters omitted):
Palmimo(*, flight_recorder: bool = True, incident_dir: str | os.PathLike[str] | None = None, ...)| Parameter | Contract |
|---|---|
flight_recorder |
True enables capture with an attached driver. False disables recorder telemetry capture and hardware-error polling, ignores incident_dir, and leaves the safety guards enabled according to their own settings. No driver means no recorder activity. |
incident_dir |
Explicit destination converted to Path at construction; relative paths retain their normal working-directory meaning. None resolves the default at the first driver connection and retains it across reconnects. Construction and connection without an open incident do not create the directory. |
default_incident_dir() -> pathlib.Path is exported from palmimo_sdk. Each call
resolves the current environment without creating a directory: an absolute,
nonempty XDG_STATE_HOME gives $XDG_STATE_HOME/palmimo/incidents; otherwise the
result is Path.home() / ".local/state/palmimo/incidents". An empty or relative
XDG_STATE_HOME uses the fallback.
Read-only properties:
| Property | Contract |
|---|---|
robot.incident_dir -> Path | None |
Explicit path immediately, or resolved default after the first driver connection. None while the default is unresolved or recording is disabled. |
robot.last_incident_path -> Path | None |
Most recent path returned by a successful write in this facade’s lifetime, retained across reconnects; initially None. It can end in .jsonl.tmp if publication failed after the complete file was written. A later failed or no-op flush does not clear it. |
Capture and flushing
Section titled “Capture and flushing”The latch and baseline model defines which hardware-error responses open a window and how axes become eligible for another incident. Reconnect resets samples and connection baselines, while retaining known latched axes and the last saved path.
connect() attempts a hardware-error baseline read before the wake glide,
retrying up to three times on transient exceptions or an all-unreached result.
Any responding axis ends the retries. Unsupported reading disables recording
for that connection immediately. If retries fail, each axis’s later first
response still becomes its baseline.
Subsequent capture and deadline checks require step() calls; there is no
background sampler or writer. Direct driver operations and facade glides that
do not call step() produce no samples. A due window is flushed after the
frame’s position write, once a fast sample reaches the planned tail. If no
such sample arrives, it becomes due 0.25 seconds after the planned deadline;
actual flushing still requires a step() call. Explicit flushes do not wait
for this grace period. At most ten dumps are attempted per connection, including
failed writes; further new windows are suppressed with one warning until
reconnect. disconnect() suppresses automatic flushing during parking
and flushes after attempting driver disconnect, including when that attempt
raises. An already open window is also flushed before recorder reset on reconnect.
robot.flush_incident() -> pathlib.Path | None synchronously writes an open
window immediately, even before its deadline. It returns this flush’s path,
or None if recording is disabled, no window is open, or writing fails. It does
not create an incident from idle samples. It may return a complete .jsonl.tmp
file if renaming failed. On the main thread, SIGINT is ignored during the write;
it is discarded rather than delivered afterward. Calls from another thread do
not change signal handlers. The automatic flush inside step() uses the same
SIGINT protection and also discards Ctrl+C during that write. Other termination
signals are not covered.
Scope. Only calls through Palmimo.step() are sampled, including facade
methods that drive that loop. Direct wake/sleep glides have no samples, and
LeRobot teleoperation that drives the engine and driver directly is outside
this recorder’s scope.
Files and header
Section titled “Files and header”Files are UTF-8 JSON Lines: one header, then sample rows ordered by monotonic
sample time across both tiers. The filename is
incident-YYYYMMDD-HHMMSS.jsonl, using the first trigger’s local wall time.
Collisions add -1, -2, and so on before .jsonl. A temporary file appends
.tmp to that full name. An incident with no samples still has its header.
| Header key | Value and meaning |
|---|---|
kind |
"header" |
schema |
"palmimo-incident/1" |
ts |
First trigger’s local ISO 8601 wall-clock timestamp, including UTC offset |
monotonic_s |
First trigger’s monotonic timestamp in seconds; local clock origin, not a Unix timestamp |
trigger |
List of triggering motor names, including axes folded into this window |
trigger_bits |
Motor-name to integer raw hardware-error bitmask at each axis’s first trigger in this window; later bit changes on a latched axis do not rewrite it |
motion |
Lowercase motion name at the first trigger, after guard verdicts |
units |
{"current": "raw", "voltage": "V", "temperature": "C", "position": "tick"} |
last_clean_t |
Earliest last-zero-response time among triggering axes, relative to monotonic_s in seconds; null if any triggering axis has no known clean response. It is a snapshot at each axis’s first trigger in the window, unaffected by later recovery or re-latching. |
hardware_error_interval_s |
1.0, the nominal hardware-error polling interval; actual gaps can be longer |
post_s |
Planned deadline relative to the first trigger, in seconds: normally 2.0; newly latched axes can extend it up to 4.0 |
post_s_captured |
Maximum sample t, clamped to at least zero, across both tiers; seconds, not a guarantee of continuous coverage |
truncated |
Whether post_s_captured < post_s; early flushes or insufficient coverage may set it. Automatic flushing waits for a fast sample at or beyond the planned tail, with a 0.25-second grace limit when fast sweeps stop |
Hardware-error bits use the same definitions as
HardwareErrorRead and describe_hardware_error().
Sample rows
Section titled “Sample rows”| Row key | Value and meaning |
|---|---|
kind |
"sample" |
tier |
"fast" for a completed leg sweep, or "context" for the context timer |
t |
Seconds relative to the first detected latch (monotonic_s); negative before detection, positive afterward. For fast rows it is the sweep time; for context rows it is the context capture time. |
motion |
Lowercase motion name at capture, after guard verdicts |
axes |
Object mapping motor names to the fields below; may be empty |
hardware_errors |
Empty object for fast rows. For context rows, the recorder’s currently known latched motor-name to integer-bitmask map, captured before any hardware-error poll in that same frame; it is not a complete raw status sweep. |
t_neck |
Context rows only: included neck sweep’s time relative to the first trigger, or null if no neck sweep was included |
Each entry in axes has current (signed raw register units, not amperes),
position (ticks), goal (ticks), temperature (°C), voltage (V), and
gap (goal - position, ticks). Missing readings are null, including axes
that were silent or unreached in the sweep. gap is null if either operand
is unavailable. An absent axis or null value does not imply zero or recovery.
goal is the last successfully submitted facade position, after clamping and
partial-goal expansion, before driver-internal IIR filtering. Motors omitted
from a write are recorded as MotionEngine.NEUTRAL, matching DynamixelDriver’s
backfill rule; another driver’s handling of partial writes may differ. It is
null until such a write exists; writes made directly through the driver are
not tracked.
Context reuses the latest leg sweep only while it is at most one second old,
and includes a pending neck sweep only while it is at most one second old.
The pending neck sweep is consumed at that context tick, even when stale;
its original goal is retained, as is the cached leg sweep’s original goal.
A context row’s t therefore need not be the measurement time of every axis.
Persistence failures and logs
Section titled “Persistence failures and logs”Writing creates the directory on demand, flushes and fsyncs the temporary file, renames it to the published name, and on POSIX fsyncs the incident directory. Ordinary write failures close the window and attempt to remove the incomplete temporary file; they do not stop motion or schedule a retry. Rename failure retains the complete temporary file. Directory-sync failure after renaming retains the published path but logs reduced durability. No parent-directory sync or protection against every power-loss scenario is promised.
| Level | Events |
|---|---|
ERROR |
Preexisting latch at an axis’s first response (no window); newly latched axis; changed bits on an already latched axis; completed published dump with path, sample count, and synchronous write duration in milliseconds |
WARNING |
Unsupported hardware-error reading disables recording for that connection; other read exceptions are rate-limited to once per five seconds; destination inspection, write, rename, cleanup, or directory-sync failure; per-connection dump limit reached (once) |
WARNING at connection |
More than 100 completed dumps or more than 100,000,000 bytes; zero-byte dumps or leftover temporary files; unwritable destination or nearest existing ancestor of a missing destination |
INFO |
Destination and existing dump count (or eventual destination); unchanged known latch at reconnect (once per axis); re-armed when all known incident axes have individually reported zero; no leg telemetry when all leg guards are disabled |
For a driver without read_telemetry support, the guards’ warning reports the
absence of leg telemetry; the guards-disabled info message covers configuration
rather than unsupported hardware.
Write duration appears only in logs. Connection-time inspection does not delete or modify existing dumps or temporary files.
Kinematics Helpers
Section titled “Kinematics Helpers”palmimo_sdk.kinematics is the shared source of truth for leg geometry. Public
helpers include leg_ik() and leg_servo_ticks() for inverse kinematics, plus
leg_forward_kinematics(leg_id, ticks) for converting one leg’s three raw servo
positions into an absolute (x, y, z) foot position in the body frame.
wrap_deg(deg) normalizes an angle into the (-180, 180] range used by the
yaw computation. All distances are millimetres; tick dictionaries use the
standard leg_<id>_* keys.
Those keys have one home here too: LEG_MOTORS is all 18 leg axis names in bus
order, and leg_motors(legs) returns the three axes of each leg given.
The user-facing SDK does not depend on the plotting or tuning scripts. Those scripts consume these helpers so their geometry cannot drift from the motion engine.
Wave Gesture Tuning
Section titled “Wave Gesture Tuning”The wave ships with baked-in defaults that reproduce the tuned hardware feel — no
knobs needed. These properties override the choreography at runtime if desired;
all are settable on robot:
| Property | Default | Description |
|---|---|---|
wave_size |
160 mm | Swing size above the rest height |
wave_speed |
1.9 | Up/down stroke rate (>1 = snappier) |
wave_intro_speed |
2.3 | Prep-raise + return rate |
wave_raise_speed |
0.6 | Rate of just the first big up-stroke (slow, deliberate) |
wave_decay |
0.8 | Per-wave amplitude multiplier — each wave 0.8x the last (1 = uniform) |
wave_count |
2 | Number of up/down waves |
wave_period |
1.0 s | Seconds per up-down cycle |
wave_dwell |
0.6 | Hold fraction per half-cycle (0 = continuous wiggle) |
Automatic per-axis servo tuning: while a wave (or wave_both / clap) is
active, the facade applies a crisp arm treatment to the moving arm(s)
(Profile_Velocity 0 + IIR low-pass) and a higher Position_P_Gain, then restores
defaults when the gesture ends — so every caller gets the tuned feel without
managing servo registers. No-op on drivers without these RAM tweaks.
The baked rates were dialed in on hardware, then scaled to compensate for the tuning bench’s per-frame readback slowdown so production playback matches what was tuned. See
engine.py_WAVE_*for the provenance.
Nod / Head-Shake Gesture Tuning
Section titled “Nod / Head-Shake Gesture Tuning”Runtime knobs for the yes/no neck gestures, all settable on robot (defaults in
engine.py _NOD_* / _SHAKE_*). The two gestures keep near-equal stroke
periods so they stay tempo-matched as a reply pair. Defaults were tuned on
hardware with on-camera review (2026-07): gentle first descent, bold amplitude,
near-uniform stroke size.
| Property | Default | Description |
|---|---|---|
nod_amp_deg |
25° | Depth of the first chin-down stroke |
nod_period |
0.65 s | Seconds per down-up cycle |
nod_count |
2 | Number of down-up cycles |
nod_decay |
0.95 | Per-cycle amplitude multiplier (<1 = later dips shallower) |
nod_first_down_scale |
1.6 | Duration multiplier of just the first descent (>1 = gentler opening; start latency is unchanged) |
shake_amp_deg |
25° | Width of the first sideways swing |
shake_period |
0.66 s | Seconds per full left-right-left cycle |
shake_swings |
4 | Side peaks (4 = 2 round trips); odd values end half-way for a stronger “no-no” |
shake_decay |
0.95 | Per-swing amplitude multiplier |
neck_rest_pitch_deg |
15° | Resting pitch trim (positive raises the gaze; clamped to ±20°). Robot-wide: idle look targets and the gestures’ home position all sit on this trimmed center |
Amplitudes are clamped to the neck’s safe range (±300 ticks ≈ ±MotionEngine.NECK_PITCH_TRAVEL_DEG
/ NECK_YAW_TRAVEL_DEG, ~26.4°) regardless of the knob value, and the absolute pitch additionally never leaves the
servo-neutral band (the trim moves the working center, not the limits). The
exact play length for the current knobs is engine.gesture_seconds(Motion.NOD)
/ engine.gesture_seconds(Motion.HEAD_SHAKE) (≈1.5 s / ≈1.3 s at the defaults).
Firing a gesture mid-gait is safe: the legs glide back to the neutral stance
(the same easing as IDLE) while the head gestures.
neck_rest_pitch_degredefines what “front” means for the whole neck:look()/look_center()targets ride on the trimmed center, andreturn_to_neutralsettles the neck pitch there rather than at the raw servo neutral (2048).
Automatic servo tuning: while a gesture is active the facade sets Profile_Velocity 0 on the two driven neck axes so the ~200 ms strokes aren’t smeared by the default time-based profile (300 ≈ 300 ms per goal), then restores the default when it ends. No-op on drivers without the RAM tweak.
Default-knob trajectories: 
Two-Handed Gestures (wave_both / clap)
Section titled “Two-Handed Gestures (wave_both / clap)”Both front legs lift at once, so only the mid+rear quad supports the robot. The defaults sit in the analyzed stable zone; the knobs open them up (verify tip margin and middle-leg servo heat on hardware before relying on a wider setting).
Shared beg-stance knobs (used by both gestures):
| Property | Default | Description |
|---|---|---|
wave_both_mid_forward |
35 mm | Middle feet extra forward shift — carries the support quad’s front edge ahead of the CoG. The main stability knob: raise it before raising anything else |
wave_both_lean |
10 mm | Whole-body backward shift. Also folds the rear knees under load — keep small, prefer mid_forward for margin |
wave_both_noseup |
8 mm | Rear corners drop — the nose-up “beg” look |
wave_both_intro_speed |
2.3 | Prep/return rate for the two-hand gestures (the single wave keeps wave_intro_speed; the middle-feet drag held up fine at this clip on hardware). Rate knobs are floored at 0.2 — 0/negative would freeze/reverse the gesture |
wave_both only:
| Property | Default | Description |
|---|---|---|
wave_both_size |
110 mm | Two-hand swing above the rest height (single wave: 160). Both arms pump the body pitch in phase — raise together with mid_forward |
wave_both_yaw |
15 deg | Inward yaw on each raised arm (hands closer together). Clamped per frame so the hand separation never drops below 25 mm (unlike the clap, the wave must not make contact — unclamped, ~41°+ would collide the paddles). Also moves the arms’ mass forward, eating tip margin — raise mid_forward with it |
wave_both_phase |
0.5 | Arm offset in wave periods: 0.5 = alternation (the default look), 0 = both arms together (banzai). Alternation also cancels the arms’ pitch reaction instead of doubling it |
wave_both_speed |
2.6 | Stroke rate — own knob, so the two-hand tempo doesn’t touch the single wave’s 1.9 |
wave_both_decay |
0.9 | Per-wave amplitude multiplier (single wave: 0.8) |
The stroke count/shape reuses the shared wave knobs (wave_raise_speed,
wave_count, wave_period, wave_dwell); defaults are the 2026-07-04
hardware tuning (alternate, 2 waves per arm, quick strokes, light decay).
clap only — open/close is specified as hand separation in mm (centre-to-centre
of the front feet), converted to arm yaw internally:
| Property | Default | Description |
|---|---|---|
clap_count |
3 | Close/open strokes per gesture |
clap_period |
0.2 s | Seconds per close+open (~5 Hz patter — the 2026-07-04 hardware tuning; the narrow clap_open is what lets this tempo track at full amplitude) |
clap_gap |
20 mm | Commanded closest foot-centre separation, = the engine floor. 20 is where the paddle faces just touch — the light touch is the intended clap feel (owner-tuned); the floor only prevents jamming them harder than that. At the shipped 5 Hz tempo the arm IIR rounds the actual approach to ~31 mm (measured); slow clap_period to ≤0.4 s to reach the commanded gap (and the light touch). Raise the gap if contact is undesired |
clap_open |
90 mm | Separation between claps (small travel = fast tempo stays crisp) |
clap_height |
90 mm | Hand lift above the neutral foot height (floored at 10 mm — the open/close sweep must not scrub the grounded feet) |
clap_dwell |
0.0 | Fraction of each half-stroke held at the ends (0 = continuous patter) |
clap_decay |
1.0 | Per-beat reopen multiplier. <1 makes each reopen shallower (the hands settle toward each other) — the “repetition wants decay” lever. Every beat still closes to clap_gap, so the contact floor is unaffected |
clap_intro_speed |
1.2 | Prep/raise + lower/release rate — the clap’s own knob (deliberately slower than wave_both_intro_speed’s 2.3: the slow raise before the fast patter is the approved contrast) |
Dance Tuning
Section titled “Dance Tuning”The dance ships with baked-in defaults that reproduce the tuned hardware feel — a
small snappy body roll on an arcing head, with a brief hold at each extreme. No
knobs needed; these properties override the sway at runtime if desired, all
settable on robot:
| Property | Default | Description |
|---|---|---|
dance_roll_deg |
5° | Body roll amplitude |
dance_pivot_h |
125 mm | Roll-pivot height; near head height keeps the head ~fixed in space |
dance_dwell |
0.2 | Hold fraction at each sway extreme (0 = continuous sine) |
dance_speed |
0.0149 | Dance tempo (its own knob — does NOT change the walk gait_speed) |
dance_level_head |
False | Arc head (dips at extremes); set True to lift the body for a flat sway |
Finishing flourish (perform_dance): the finished performance sways
sways (3) times, ending on an extreme (the held side, not a center
crossing), then closes with two stages tuned on hardware:
end_hold(afterglow, 0.2 s) — the legs softly hold the actual final pose at a lowered Position_P_Gain. Holding the real pose (not the last full-amplitude goal the servos were lagging toward) avoids an overshoot past the sway just shown; the lowered gain keeps it soft instead of a hard clunk. The neck is left out of the hold so it stays commanded to neutral (head up) at full gain.settle(slow return, 1.0 s) — a single eased glide back to neutral (larger = slower/softer). Never an ease-out during the dance, which would read as extra sway.
Defaults are hardware-tuned — see
engine.py_DANCE_*(the sway) androbot.py_DANCE_*(the finish).dance_speedis fps-compensated: it is scaled by the readback-slowed framerate it was tuned at, so playback at the production 60 fps (perform_dance) holds the intended wall-clock speed instead of running faster.
ServoDriver
Section titled “ServoDriver”ServoDriver (palmimo_sdk.io.base.ServoDriver) is the abstract I/O boundary
that DynamixelDriver implements and that a third-party backend (simulator,
other serial bus) implements in its place — robot.driver above holds
whichever one is attached.
SAFE_MIN_TICK (200) and SAFE_MAX_TICK (3900) are the safe servo tick
range — clear of the mechanical limits at 0/4095 — and are part of the SDK’s
public surface:
from palmimo_sdk import SAFE_MAX_TICK, SAFE_MIN_TICKServoDriver.write_positions is a concrete method: it clamps every goal into
[SAFE_MIN_TICK, SAFE_MAX_TICK] and logs a warning on each out-of-range goal,
then hands the clamped positions to _write_positions, the abstract method a
backend (DynamixelDriver included) implements to reach the hardware. A
backend never has to apply the clamp itself.
ServoDriver.read_positions() reads present servo positions but collapses a
failed read to a safe placeholder ({} for a whole-batch failure, NEUTRAL
for a single dropped motor), so a caller cannot tell a dropout from a real
NEUTRAL reading. ServoDriver.read_positions_span(motors=None) reads the
same signal without that placeholder: it returns a ServoPositions
(palmimo_sdk.io.base.ServoPositions) whose positions mapping holds only
the motors that answered, alongside silent (asked, no answer) and
unreached (never asked because the sweep already stopped) — the same shape
as ServoTelemetry below. Callers that must not mistake a dropped read for a
real joint position (e.g. a policy loop running at a fixed control rate) use
this instead of read_positions(). Optional capability; the default raises
NotImplementedError.
ServoDriver.read_telemetry(motors=None) returns a ServoTelemetry of
current (raw signed), voltage (V), temperature (°C) and position (raw
tick) in one sweep, with the same silent / unreached bookkeeping. The
Rail Undervoltage Guard reads voltage.
ServoDriver.set_torque_enabled(enabled, motors=None) switches torque on the
named motors (every motor when None). Both are optional capabilities; the
defaults raise NotImplementedError.
ServoDriver.read_hardware_errors(motors=None) returns a HardwareErrorRead
(palmimo_sdk.io.base.HardwareErrorRead) in one sweep; None selects all motors.
Its read-only errors mapping contains raw unsigned Hardware Error Status
(register 70, one byte) for every responding motor, including zero. A missing
motor is unknown, not healthy. silent lists motors asked without an answer;
unreached lists motors the sweep did not reach (all requested motors if sending
failed). The latched property returns a dict of only nonzero entries.
An empty latched dict does not establish that all motors are healthy if
silent or unreached is nonempty. Judge each motor using errors.
The servo itself detects and latches the bits until reboot or power cycling.
Reading does not clear them or change torque. DynamixelDriver makes one bus
request without retrying. A disconnected driver raises RuntimeError; an
unknown motor name raises KeyError. Transport failures may raise exceptions;
callers can log and discard that sweep. This is an optional capability; the
base implementation raises NotImplementedError.
HardwareErrorRead, HARDWARE_ERROR_BITS, and describe_hardware_error(bits)
are exported from palmimo_sdk. The helper returns "-" for zero, otherwise
comma-separated names in ascending bit order (e.g. 0x21 becomes
"InputVoltage,Overload"), with unknown bits named "bit<N>".
HARDWARE_ERROR_BITS maps bit positions to InputVoltage (0), Overheating (2),
ElectricalShock (4), and Overload (5). Bit 3 is reserved in the XC330-M288
e-Manual and is described as "bit3".
DynamixelDriver
Section titled “DynamixelDriver”The bundled hardware backend (Dynamixel serial bus; needs the palmimo-sdk[hardware] extra).
DynamixelDriver(port=None, *, baudrate=1_000_000, motor_model="xc330-m288", profile_velocity=300, keep_torque_on_disconnect=False, calibration=None, bus_factory=..., connect_timeout=10.0)
port=Noneauto-detects the bus atconnect()(see Port Auto-Detection).motor_modelmust be one ofSUPPORTED_MOTOR_MODELS; any other value raisesValueError."xc330-m288"is the real hardware;"xl330-m288"fails the connect handshake on actual units.profile_velocityis the time-based Profile_Velocity applied to every motor at connect, in milliseconds per move (lower is faster).keep_torque_on_disconnect=Falsecuts torque ondisconnect().calibrationis accepted for API parity but not applied: the bus exchanges raw ticks.bus_factoryopens the serial bus; replace it to run the driver against a fake bus in tests.connect_timeoutbounds the whole connect sequence (seeDynamixelConnectTimeoutErrorbelow).
connect() seeds each motor’s Goal_Position from its present position before enabling torque. If any motor’s position cannot be read after four attempts, connect() raises ConnectionError naming the motors and torque stays off.
from palmimo_sdk import SUPPORTED_MOTOR_MODELS, palmimo_motor_ids
SUPPORTED_MOTOR_MODELS # ("xl330-m288", "xc330-m288")palmimo_motor_ids() # {"leg_1_yaw": 1, ..., "neck_pitch1": 19, "neck_pitch2": 20, "neck_yaw": 21}palmimo_motor_ids() -> dict[str, int] returns the 21-motor name to Dynamixel ID map: IDs 1-18 are the six legs (leg_{1-6}_{yaw,pitch1,pitch2}, body to tip), IDs 19-21 the neck. The names match the keys step() emits.
Port Auto-Detection
Section titled “Port Auto-Detection”find_servo_port() -> str
Section titled “find_servo_port() -> str”Auto-detect the servo bus’s serial port (its USB-to-servo bridge) without specifying a hard-coded path.
from palmimo_sdk import find_servo_port, PortDetectionError
try: port = find_servo_port()except PortDetectionError as e: print(e) # "Servo bus not found. Check connection or specify with --port."Detection priority:
- USB vendor-ID match (
0x2F5D) — on Windows the bridge enumerates as a nameless generic serial device, so the VID is the only identifier. - Pattern fallback (only if step 1 finds nothing):
/dev/ttyACM<n>(Linux) or/dev/cu.usbmodem*(macOS).
Returns the device path (e.g. "/dev/ttyACM0") when exactly one candidate is found.
Raises PortDetectionError when zero or multiple candidates are found.
Calling it yourself is only needed to print or log the resolved path —
DynamixelDriver(port=None) (the default) runs the same detection at
connect() time, mirroring FaceDisplay(port=None).
find_face_port() -> str | None
Section titled “find_face_port() -> str | None”Return the serial device path of the attached face display, or None when none is found. It matches the face display’s USB vendor ID (0x2E8A), so on a Pi, where the face and the servo bus both appear as /dev/ttyACM*, it picks the face and not the bus. With several face displays attached it logs a warning and returns the first.
Unlike find_servo_port(), it does not raise on a miss. FaceDisplay(port=None) (the default) calls it at connect() and raises FaceDisplayError when it returns None.
PortDetectionError
Section titled “PortDetectionError”Raised by find_servo_port() when detection fails. The message always
includes a remedy (specify --port explicitly).
from palmimo_sdk import PortDetectionErrorDynamixelConnectTimeoutError / FaceDisplayConnectTimeoutError
Section titled “DynamixelConnectTimeoutError / FaceDisplayConnectTimeoutError”Raised when the whole connect sequence exceeds connect_timeout seconds
(10s / 5s defaults; see robot.connect() above). Distinct from
PortDetectionError (auto-detection found zero or multiple candidates):
these fire when a port was resolved but opening/handshaking/arming never
completed — no robot/face attached, or the matched port is a different
device. The message names the device and the resolved port.
FaceDisplayConnectTimeoutError subclasses FaceDisplayError, so broad
catches keep working. If the abandoned connect attempt finishes late anyway,
the orphaned bus/port is closed automatically (palmimo_sdk.io._timeout).
from palmimo_sdk import DynamixelDriver, DynamixelConnectTimeoutError
try: DynamixelDriver().connect() # no robot attachedexcept DynamixelConnectTimeoutError as e: print(e) # "Timed out connecting to the Dynamixel servo bus on ... "palmimo_sdk.agent typing
Section titled “palmimo_sdk.agent typing”palmimo_sdk ships a py.typed marker (PEP 561), so its annotations reach
your type checker. The agent layer’s receiver — Tool.execute(robot) and
AgentToolSet(robot, ...) — is typed as palmimo_sdk.agent.PalmimoLike, a
Protocol listing exactly the facade surface the tools use. A real Palmimo
satisfies it as-is; a hardware-free test double satisfies it structurally by
implementing just the members it exercises, no subclassing or casting needed.
palmimo_sdk.mcp
Section titled “palmimo_sdk.mcp”Exposes a Palmimo’s agent tools (palmimo_sdk.agent.AgentToolSet /
Tool — see palmimo_sdk/agent/tools.py and toolset.py for the tool
catalog) over the Model Context Protocol,
so an MCP client
(Claude Code, Claude Desktop, or any other MCP-speaking agent) can list and
call them the same way an OpenAI tool-calling loop does. Requires the
palmimo-sdk[mcp] extra — see the MCP server guide for the
install command and CLI examples.
build_mcp_server(toolset: AgentToolSet, log_tool_calls: str = "summary") -> mcp.server.lowlevel.Server
Section titled “build_mcp_server(toolset: AgentToolSet, log_tool_calls: str = "summary") -> mcp.server.lowlevel.Server”Builds a low-level MCP server that lists toolset’s registered tools
(mcp_types.Tool.input_schema mirrors each Tool.parameters_schema()) and
dispatches tools/call requests to await toolset.call(name, arguments), via
the on_list_tools/on_call_tool handlers Server takes on the MCP Python
SDK 2.x. Building the server does not connect the robot or start serving —
see the CLI below.
from palmimo_sdk import Palmimofrom palmimo_sdk.agent import AgentToolSetfrom palmimo_sdk.mcp import build_mcp_server
robot = Palmimo()server = build_mcp_server(AgentToolSet(robot))AgentToolSet.call(name, arguments) is async: it serializes overlapping
callers with its own internal lock (toolset.is_busy() reports whether it is
currently held) and runs each tool’s execute() — synchronous and
potentially blocking for seconds, e.g. a gesture’s run(seconds=...) —
on a worker thread via asyncio.to_thread, so two concurrent MCP calls never
drive the single-threaded Palmimo facade at the same time while the event
loop stays free for other MCP protocol traffic. This server’s call_tool
handler runs toolset.call(...) as its own asyncio.Task and awaits it
behind asyncio.shield, so a cancellation of the HANDLER’s own task (a
client disconnect, the session shutting down, …) cannot reach into
toolset.call() — it keeps running to completion regardless. It is
deliberately non-preemptive — it never calls
toolset.cancel_running(), so a motion already in flight always runs to
completion rather than being cut short by a second, concurrent request (see
mcp/server.py’s module docstring for the full rationale). A caller that
wants a genuinely preemptible tool-calling loop should drive AgentToolSet
directly (await toolset.call(...) / await toolset.cancel_running())
instead of going through this MCP server.
A ToolResult’s text becomes a TextContent; each JPEG in images becomes
an ImageContent (base64, image/jpeg) after it. toolset.call() never
raises for an LLM-caused failure, so an unknown tool name or bad arguments
come back as ordinary TextContent, never an MCP protocol error.
palmimo_sdk.robot.MotionCancelled (raised when Palmimo.cancel()
interrupted an in-flight motion) is likewise reported as ordinary,
non-error TextContent ("interrupted: ...") rather than an error — an
observation, not a failure.
Each dispatched call leaves one INFO record on the palmimo_sdk.mcp.server
logger: tool call name=<tool> args={...} outcome=<ok|error|interrupted> duration_ms=<float> result=<...> images=<n>. log_tool_calls selects how much
of the call it carries — "summary" (the default) prints argument names and
numeric values but reduces every string to its length, "full" prints
argument values and the result text verbatim, and "off" emits nothing.
build_mcp_server() only emits: it installs no handler and changes no logger’s
configuration, so the records go nowhere until the embedding application
configures one (the CLI below sends them to stderr). A library that attached a
stdout handler here would corrupt every stdio-transport session, since stdout
is the MCP protocol stream on that transport.
CLI: python -m palmimo_sdk.mcp
Section titled “CLI: python -m palmimo_sdk.mcp”Wires a real Palmimo (servo bus, face display, speaker, head camera — each
degrading to None on failure the same way _build_servo_driver does in the
wake-word example) to build_mcp_server() and serves it, over either stdio
(the default, for a local MCP client sharing this process’s stdin/stdout) or
streamable HTTP (for a client on a different machine than the robot). See
the MCP server guide for the run commands, the full CLI flag
reference, and Claude Code registration commands.

