Observations

An Observation is the typed snapshot of every sensor reading the robot publishes each frame. Observations flow from the robot through the relay server to every controller, and they are the same Python object you record to disk, replay, and feed to a learned policy.

Defined in solarian/core/observation.py.

What is an Observation?

An Observation is a top-level container of ModuleObs instances, one per Module on your robot. Each ModuleObs in turn is a container of typed Data values — the actual sensor readings.

Observation
 ├── arm: ArmObs
 │    ├── shoulder_pan_pos: PosData
 │    └── shoulder_lift_pos: PosData
 └── gripper_camera: CameraObs
      ├── rgb: ImageData
      └── bbox: BoundingBoxData    (computed on the server)

The class hierarchy is mirrored on disk and on the wire: every ModuleObs attribute matches the corresponding module's attribute name on your Robot, and every SensorField references a specific Data descriptor on a sensor.

ModuleObs

Subclass ModuleObs and bind each field to a Data reference on your Robot:

from solarian.core.observation import SensorField, ModuleObs
from myrobot.robot import MyRobot

class ArmObs(ModuleObs):
    shoulder_pan_pos = SensorField(MyRobot.arm.shoulder_pan.pos)
    shoulder_lift_pos = SensorField(MyRobot.arm.shoulder_lift.pos)
    elbow_flex_pos = SensorField(MyRobot.arm.elbow_flex.pos)

MyRobot.arm.shoulder_pan.pos is a PosData descriptor on the FeetechServo actuator. Solarian uses it to extract:

• Type information (dtype, shape) — used to allocate shared-memory buffers and validate runtime values.
• Path ((module_name, component_name, data_name)) — used to route the sensor read into the right field on the right module.

Observation

Subclass Observation and attach ModuleObs instances. The attribute name on Observation must equal the attribute name on Robot — Solarian validates this at class definition time:

from solarian.core.observation import Observation
from myrobot.robot import MyRobot

class MyObservation(Observation):
    arm = ArmObs()                # MyRobot.arm
    gripper_camera = CameraObs()  # MyRobot.gripper_camera

If the names disagree, you'll get a TypeError at import time:

MyObservation.arm: SensorField 'shoulder_pan_pos' references module 'arm',
but the Observation attribute is named 'gripper_camera'. Observation
attribute names must match the module names declared on the Robot subclass.

This catches schema drift before any of it makes it to the wire.

Declaring sensor fields

A SensorField binds an observation field to a single Data point on a sensor. It carries dtype/shape information from that Data and is used at runtime to wrap raw values.

from solarian.core.observation import SensorField

class CameraObs(ModuleObs):
    rgb = SensorField(MyRobot.gripper_camera.sensor.rgb_image)
    depth = SensorField(MyRobot.gripper_camera.sensor.depth_image)

You can introspect a ModuleObs subclass to discover its fields:

ArmObs.get_sensor_fields_spec()
# {'shoulder_pan_pos': SensorField(...), 'shoulder_lift_pos': SensorField(...), ...}

Calculated fields

Some observation fields aren't direct sensor reads — they're computed from other fields. A CalculatedField declares the inputs, the processor that produces the output, and where the processor runs: ExecutionPlace.SERVER or ExecutionPlace.ROBOT, the same enum that places skills.

from solarian.core.observation import CalculatedField
from solarian.core.data import BoundingBoxData
from solarian.core.execution import ExecutionPlace
from myproject.processors import BBoxProcessor

class CameraObs(ModuleObs):
    rgb = SensorField(MyRobot.gripper_camera.sensor.rgb_image)
    bbox = CalculatedField(
        inputs=[MyRobot.gripper_camera.sensor.rgb_image],
        processor=BBoxProcessor,
        execution_place=ExecutionPlace.SERVER,
        output=BoundingBoxData(),
        conf="conf/services/bbox_tracker.yaml",
    )

Leave out execution_place= to read it from an execution_place: key in the conf instead, so a deployment can move the processor between hosts. At startup, each host calls Observation.create_processors(place) to instantiate every processor whose execution_place is its own. Processors run inline as observations stream through, so by the time the controller sees the message, obs.gripper_camera.bbox is already populated.

See Server for the relay-side processor lifecycle.

No host builds CONTROLLER processors.

Data classes

A Data is the typed wrapper around every value that flows through Solarian. Sensor reads return Data instances; actuator writes accept them; observations and actions are containers of them. A component declares its channels as Data class attributes, and a SensorField or ActuatorField binds to one of those.

Defined in solarian/core/data.py. All Data subclasses are Pydantic models, so they validate on construction and serialize cleanly.

Every Data carries:

The built-in subclasses:

Custom Data subclasses

Define your own when you need a typed channel that isn't covered by the built-ins:

from solarian.core.data import Data
from pydantic import Field

class IMUData(Data):
    dtype: type = float
    shape: tuple = (6,)   # (ax, ay, az, gx, gy, gz)
    payload: list[float] = Field(default_factory=lambda: [0.0]*6, alias="value")

Then attach it to a sensor:

class IMUSensor(Sensor):
    bus_class = IMUBus
    imu = IMUData()

IMUSensor.imu becomes the Data reference you bind to a SensorField in your ModuleObs.

Reading an observation

In a controller, the latest observation is always available as self.last_observation, protected by self.observation_lock:

class MyController(Controller):
    def control_step(self) -> MyAction:
        with self.observation_lock:
            obs = self.last_observation
        pan = obs.arm.shoulder_pan_pos.value     # auto-uncompresses if needed
        bbox = obs.gripper_camera.bbox.value
        ...

Data.value is a property that auto-uncompresses on read — useful when working with ImageData, which is JPEG-encoded on the wire and decoded only when you actually access .value.

Building observations from scratch

The easiest way to allocate a zero-initialized observation is Observation.zeros():

obs = MyObservation.zeros()
obs.arm.shoulder_pan_pos.value = 0.42
print(obs.to_dict())

Observation.zeros() walks every declared ModuleObs and SensorField/CalculatedField, allocates an empty Data of the right type, and returns a fully populated container. Useful in tests, replay, and policy code.

Compression and serialization

Each Data subclass picks its codec with _default_codec(): ImageData uses JPEG via OpenCV, and everything else falls back to zlib + pickle. Compression is opt-in: the framework calls obs.compress() before sending, and the consumer's lazy .value access decompresses on first read. The relay-bound code path is roughly:

robot:        obs.compress()  →  msgpack(obs.to_dict())  →  ZMQ
controller:   ZMQ  →  msgpack.unpackb()  →  Observation(data)
              .value access auto-uncompresses lazily

This means the controller pays the JPEG decode only for fields it actually reads.

See also

• Actions — the counterpart that flows the other way.
• Modules — where the underlying sensor data lives.
• Server — running CalculatedFields on the relay.