defined-cli API Reference¶
Mission control — DefinedSession, state management, events, and world model.
Session¶
session
¶
DefinedSession — unified Python API for all robot interaction.
Single entry point that replaces both Orchestrator (one-shot) and MissionController (persistent). Manages connection lifecycle, mission execution, state persistence, and event notification.
This module has zero imports from click, rich, or textual.
Flow¶
connect()
DISCONNECTED → CONNECTING → CONNECTED (starts reconnect watchdog)
run_mission(task, rdf)
Sets COMPILING immediately (under lock), then spawns a daemon thread:
COMPILING → resolve POI references → compile YAML → BT XML
DEPLOYING → wait for executor IDLE heartbeat → deploy XML
RUNNING → monitor /task_status until SUCCESS/FAILURE/timeout
SUCCEEDED / FAILED (saved to StateStore)
stop_mission()
Signals the daemon thread to abort; transition → FAILED.
disconnect()
Stops the watchdog, closes transport, sets DISCONNECTED.
Usage (script): session = DefinedSession(target=SimTarget(), transport=RosbridgeTransport(), store=StateStore(), compile_fn=compile_task) session.connect() session.add_poi("dock", 0.0, 0.0) session.run_mission(Path("patrol.task.yaml"), Path("robot.rdf.yaml")) session.disconnect()
Usage (TUI): app = DefinedApp(session) app.run()
DefinedSession
¶
Unified Python API for robot interaction.
Manages the full lifecycle: connection, mission execution, state persistence, and event notification. Thread-safe.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
TargetBase
|
Backend lifecycle manager (SimTarget, HwTarget). |
required |
transport
|
TransportBase
|
Communication layer (RosbridgeTransport). |
required |
store
|
StateStore
|
State persistence (StateStore). |
required |
compile_fn
|
Callable[..., CompileResult]
|
Task compiler function (compile_task). |
required |
latest_readiness
property
¶
Most recent readiness check result.
connection_status
property
¶
Current transport connection state.
mission_status
property
¶
Current mission execution state.
state
property
¶
Current state snapshot (read-only copy).
blackboard
property
¶
Blackboard with current world state.
current_progress
property
¶
Latest TaskProgress update, or None if idle.
last_mission
property
¶
Most recently completed or active mission record.
steps
property
¶
Steps from the most recently compiled mission.
reports
property
¶
Recent report messages.
last_error
property
¶
Error from the most recent failed mission.
add_listener(callback)
¶
Register an event listener. Thread-safe.
remove_listener(callback)
¶
Unregister an event listener. Thread-safe.
connect()
¶
Start target (if needed), connect transport, start watchdog.
disconnect()
¶
Stop watchdog, disconnect transport, update state.
compile(task, rdf, *, verbs_dir=None)
¶
Compile a task YAML to BT XML. Works without connection.
run_mission(task, rdf, *, verbs_dir=None, timeout=_DEFAULT_TIMEOUT, param_overrides=None)
¶
Compile, deploy, and monitor a mission. Non-blocking.
Spawns a daemon thread. Use mission_status property to
track progress. Events are emitted to listeners.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
param_overrides
|
dict | None
|
Key/value pairs that override matching params in
every task step at compile time. E.g. |
None
|
stop_mission()
¶
Stop the current mission gracefully. No-op if nothing running.
restart_mission()
¶
Re-run the last mission with the same arguments.
add_poi(name, x, y, *, poi_type='static', radius=0.5)
¶
Add or update a Point of Interest.
list_pois()
¶
Return all POIs as a dict.
mark_poi(name, *, poi_type='static', radius=0.5)
¶
Mark the robot's current position as a named POI.
publish_velocity(linear_x, angular_z)
¶
Publish a velocity command to /cmd_vel. No-op if not connected.
emergency_stop()
¶
Immediately halt all motion and abort any running mission.
Publishes zero velocity to /cmd_vel, signals the mission thread to stop, and marks the mission as FAILED. Safe to call from any state.
Events¶
events
¶
Mission event types — connection states, mission states, and session events.
Pure data types with no I/O, no CLI dependencies, and no side effects.
Used by DefinedSession for lifecycle tracking and by TUI/scripts
for observing state changes.
ConnectionStatus
¶
Bases: Enum
Transport connection lifecycle states.
MissionStatus
¶
Bases: Enum
Mission execution lifecycle states.
SessionEvent
dataclass
¶
Structured event emitted by DefinedSession.
Attributes:
| Name | Type | Description |
|---|---|---|
timestamp |
datetime
|
When the event occurred (UTC). |
category |
Literal['connection', 'executor', 'mission', 'error', 'health']
|
Event category for filtering/routing. |
message |
str
|
Human-readable description (Layer 1). |
suggestion |
str | None
|
Actionable hint (Layer 2), or None. |
detail |
dict[str, Any] | None
|
Structured data (Layer 3), or None. |
POI Resolver¶
resolver
¶
POI reference resolver -- expands $world.pois.X in task dicts.
Runs before the compiler. Walks task step params and replaces
string references ($world.pois.kitchen) with concrete
coordinate values from the Blackboard.
Usage
from defined_cli.mission.resolver import resolve_references
resolved = resolve_references(task_dict, blackboard)
resolved["steps"][0]["params"]["x"] == 1.5¶
ResolverError
¶
Bases: Exception
A POI reference in the task could not be resolved.
Attributes:
| Name | Type | Description |
|---|---|---|
reference |
The unresolved reference string (e.g. |
resolve_references(task_dict, blackboard)
¶
Resolve all $world.pois.X references in a task dict.
Returns a new dict (deep copy) with references replaced by
concrete values. For target params pointing to a POI,
expands the POI center into x and y coordinates.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task_dict
|
dict
|
Parsed task YAML (from |
required |
blackboard
|
Blackboard
|
Blackboard containing POI data. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
A new task dict with all references resolved. |
Raises:
| Type | Description |
|---|---|
ResolverError
|
If a referenced POI does not exist. |
State Store¶
store
¶
State persistence -- atomic YAML read/write for robot state.
Manages the ~/.defined/state.yaml file. Writes are atomic
(temp file + os.replace) to prevent corruption from crashes
or concurrent access.
Usage
from defined_cli.state.store import StateStore
store = StateStore() snapshot = store.load() snapshot.robot.current = RobotStatus.IDLE store.save(snapshot)
StateStore
¶
Single writer for ~/.defined/state.yaml.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path | None
|
Override the default state file path (for testing). |
None
|
path
property
¶
Return the state file path.
load()
¶
Load state from disk.
Returns:
| Type | Description |
|---|---|
StateSnapshot
|
The persisted snapshot, or a default |
StateSnapshot
|
if the file does not exist. |
Raises:
| Type | Description |
|---|---|
YAMLError
|
If the file contains invalid YAML. |
save(snapshot)
¶
Write state to disk atomically.
Creates parent directories if needed. Writes to a temporary
file in the same directory, then atomically replaces the
target via os.replace.
Blackboard¶
blackboard
¶
Spatial blackboard -- dot-path key-value store with POI semantics.
The blackboard is the robot's spatial memory. POIs (Points of Interest)
are first-class entries under world.pois.*, stored with three
type tiers: constant, static, and dynamic.
Usage
from defined_cli.state.blackboard import Blackboard
bb = Blackboard() bb.set_poi("dock", (0.0, 0.0), poi_type="constant") bb.set_poi("kitchen", (1.5, 2.0)) print(bb.get_poi("kitchen"))
Blackboard
¶
Nested key-value store with dot-path access and POI convenience methods.
Wraps a plain dict internally. All access goes through dot-path methods to enforce consistent structure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict | None
|
Initial data dict (e.g. loaded from StateStore). Defaults to empty. |
None
|
get(key, default=None)
¶
Get a value by dot-path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Dot-separated path (e.g. |
required |
default
|
Any
|
Returned if the path does not exist. |
None
|
Returns:
| Type | Description |
|---|---|
Any
|
The value at the path, or default. |
set(key, value)
¶
Set a value at a dot-path, creating intermediate dicts as needed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Dot-separated path. |
required |
value
|
Any
|
Value to store. |
required |
delete(key)
¶
Delete the value at a dot-path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
str
|
Dot-separated path to delete (e.g. |
required |
Raises:
| Type | Description |
|---|---|
KeyError
|
If the path does not exist. |
list(prefix)
¶
Return all children under a dot-path prefix.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prefix
|
str
|
Dot-separated path to a dict node. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The dict at prefix, or an empty dict if the path does not exist. |
set_poi(name, center, radius=0.5, poi_type='static', frame='map')
¶
Add or update a Point of Interest.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
POI identifier (e.g. |
required |
center
|
dict | tuple
|
|
required |
radius
|
float
|
Area radius in meters. |
0.5
|
poi_type
|
str
|
One of |
'static'
|
frame
|
str
|
Coordinate frame (default |
'map'
|
get_poi(name)
¶
Return a POI dict by name, or None if not found.
list_pois()
¶
Return all POIs as {name: poi_dict}.
clear()
¶
Remove all data from the blackboard, starting fresh.
clear_pois()
¶
Remove all POIs, preserving other blackboard data.
to_dict()
¶
Return a deep copy of the internal data for StateStore serialization.
from_dict(data)
classmethod
¶
Reconstruct a Blackboard from a plain dict.
State Models¶
model
¶
State data models -- robot status, world state, and mission records.
Pure data containers with no I/O, no CLI dependencies, and no side effects.
Used by StateStore for persistence and Blackboard for spatial memory.
Usage
from defined_cli.state.model import StateSnapshot, RobotStatus
snapshot = StateSnapshot() snapshot.robot.current = RobotStatus.IDLE
RobotStatus
¶
Bases: Enum
Robot lifecycle states.
Transitions
OFFLINE -> IDLE -> ON_MISSION -> IDLE ON_MISSION -> PAUSED -> ON_MISSION | IDLE
RobotState
dataclass
¶
Current robot status and last mission metadata.
Attributes:
| Name | Type | Description |
|---|---|---|
current |
RobotStatus
|
Active lifecycle state. |
last_mission_id |
str | None
|
ID of the most recently completed mission. |
last_mission_result |
str | None
|
Outcome of the last mission (SUCCESS, FAILURE, ABORTED). |
WorldState
dataclass
¶
Spatial world model -- POIs keyed by name.
Each POI value is a plain dict:
{"center": {"x": ..., "y": ...}, "radius": ..., "type": ..., "frame": ...}
Attributes:
| Name | Type | Description |
|---|---|---|
pois |
dict[str, dict]
|
Map of POI name to spatial entity dict. |
MissionRecord
dataclass
¶
Completed or in-progress mission metadata.
Attributes:
| Name | Type | Description |
|---|---|---|
id |
str
|
Unique mission identifier. |
task_name |
str
|
Name from the task YAML. |
status |
str
|
One of PENDING, RUNNING, SUCCESS, FAILURE, ABORTED. |
started_at |
str | None
|
ISO 8601 timestamp, or None if not yet started. |
completed_at |
str | None
|
ISO 8601 timestamp, or None if not yet finished. |
result |
str | None
|
Human-readable result summary. |
StateSnapshot
dataclass
¶
Full robot state -- persisted to ~/.defined/state.yaml.
Attributes:
| Name | Type | Description |
|---|---|---|
robot |
RobotState
|
Current robot lifecycle state. |
world |
WorldState
|
Spatial world model (POIs). |
mission_queue |
list[dict]
|
FIFO queue of pending mission dicts. |
mission_history |
list[MissionRecord]
|
Completed mission records. |
schema_version |
str
|
State file format version for future migrations. |
to_dict()
¶
Serialize to a plain dict suitable for yaml.dump.
Converts enums to their string values and dataclasses to plain dicts for YAML serialization.
Returns:
| Type | Description |
|---|---|
dict
|
A plain dict with no dataclass or enum instances. |
from_dict(data)
classmethod
¶
Reconstruct from a YAML-loaded dict. Missing keys use defaults.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
dict
|
Plain dict, typically from |
required |
Returns:
| Type | Description |
|---|---|
StateSnapshot
|
A populated |
World Loader¶
world_loader
¶
World definition loader and blackboard seeding.
Loads world.yaml and populates the spatial blackboard with POIs.
This module has zero CLI-specific imports (no click, rich, textual).
Usage
from defined_cli.state.world_loader import load_world, seed_blackboard from defined_cli.state.blackboard import Blackboard
world = load_world(Path("world.yaml")) bb = Blackboard() seed_blackboard(world, bb)
POIDefinition
¶
Bases: BaseModel
A Point of Interest in the world definition.
SpawnPosition
¶
Bases: BaseModel
Robot spawn position for simulation.
SimEnvironment
¶
Bases: BaseModel
Simulation environment configuration.
ZoneDefinition
¶
Bases: BaseModel
A zone boundary (parsed but not enforced in v0.1.0).
WorldDefinition
¶
Bases: BaseModel
Complete world definition from world.yaml.
load_world(path)
¶
Parse and validate a world.yaml file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Path
|
Path to the world.yaml file. |
required |
Returns:
| Type | Description |
|---|---|
WorldDefinition
|
Validated WorldDefinition. |
seed_blackboard(world, blackboard)
¶
Populate blackboard with POIs from a world definition.
POIs are added via Blackboard.set_poi(), which handles the
conversion to the internal format (center, radius, type, frame).
Existing POIs in the blackboard are preserved (merge, not replace).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
world
|
WorldDefinition
|
Parsed world definition. |
required |
blackboard
|
Blackboard
|
Blackboard to populate. |
required |