Build

Adapters

If your instruments already answer to SiLA 2, PyLabRobot, MADSci, OPC UA, Modbus or ROS 2, you do not write a driver. You write a few lines of bindings, and the adapter builds an MHP driver with every safety gate, describe level, job and e-stop included.

All five adapters rest on one mechanism. A binding pairs an MHP name with a callable into the foreign layer, and BoundDriver turns three lists of bindings into a complete driver. The ecosystem adapters just build those lists by introspecting their layer. Each adapter's mapping logic is tested against injected fakes; the exact client-library calls should be confirmed against a live system in your lab, and the paths are configurable for that reason.

Layer	Function	What becomes what
SiLA 2	`sila_device`	property → signal; one-parameter command → setting; command → action, observable commands stream progress and accept cancel
PyLabRobot	`plr_device`	every public coroutine on a Machine → action, params from its signature; `setup()` on start, `stop()` on e-stop
MADSci	`madsci_node`	self-describing: node info → actions with params and notes; state → signals; status → ping; no binding map needed
OPC UA	`opcua_device`	variable node → signal or setting; method node → action; abort method → e-stop. Modbus uses `BoundDriver` with two register callables
ROS 2	`ros2_device`	topic subscription → signal; topic publication → setting; action server → action with feedback progress; Trigger service → e-stop

Each device still gets a device package: the DEVICE.md frontmatter and instructions are what agents read, and the adapter call lives in driver.py as a module-level DEVICE. Identity from the frontmatter is merged in, so the adapter call carries only bindings.

SiLA 2

# devices/arm-01/driver.py
from openmhp.adapters.sila2 import sila_device
DEVICE = sila_device("10.0.0.12", 50052, device={},
    signals={"position": "RobotController.Position",
             "safe_zone_clear": ("SafetyController.SafeZoneClear", "boolean")},
    settings={"speed": ("RobotController.SetSpeed.Speed", {"min": 1, "max": 100})},
    actions={"move_to": ("RobotController.MoveTo", {"observable": True, "interlocks": ["safe_zone_clear"]})},
    estop="RobotController.EmergencyStop")

Discover feature identifiers with client.SiLAService.ImplementedFeatures.get(). MHP leases replace SiLA's LockController; do not also lock from another client.

PyLabRobot

from pylabrobot.liquid_handling import LiquidHandler
from pylabrobot.liquid_handling.backends import STAR
from pylabrobot.resources import STARLetDeck
from openmhp.adapters.pylabrobot import plr_device

lh = LiquidHandler(backend=STAR(), deck=STARLetDeck())
DEVICE = plr_device(lh, device={},
    actions=["pick_up_tips", "drop_tips", "aspirate", "dispense", "move_plate"],
    limits={"aspirate": {"vols": [0, 1000]}, "dispense": {"vols": [0, 1000]}})

Omit actions to expose every public coroutine. Resources are passed by name; wrap the machine in a thin class that looks names up on the deck if agents should not need PyLabRobot object graphs.

MADSci

from openmhp.adapters.madsci import madsci_node
DEVICE = madsci_node("http://192.168.1.40:2000", device={},
                     approval={"transfer": "auto", "free_drive": "forbid"})

Nodes describe themselves through /info, /state and /status; the adapter needs no binding map. A MADSci workcell manager can also act as a host and point at the MHP directory instead of per-node REST.

OPC UA and Modbus

from openmhp.adapters.opcua import opcua_device
DEVICE = opcua_device("opc.tcp://furnace-plc:4840", device={},
    signals={"zone1_temperature": ("ns=2;s=Furnace.Zone1.PV", "degC"),
             "door_closed": ("ns=2;s=Furnace.DoorClosed", "boolean")},
    settings={"zone1_setpoint": ("ns=2;s=Furnace.Zone1.SP", {"min": 20, "max": 1400}, "degC")},
    actions={"start_program": ("ns=2;s=Furnace", "ns=2;s=Furnace.Start",
                               {"interlocks": ["door_closed"], "approval": "confirm", "params": {"program": "int"}})},
    estop=("ns=2;s=Furnace", "ns=2;s=Furnace.Abort"))

Method arguments are passed positionally in the order of params. For Modbus, write two callables (read register, write register) and use BoundDriver directly. PLC safety logic stays primary; MHP limits are the layer that stops an agent from asking.

ROS 2

from openmhp.adapters.ros2 import ros2_device
DEVICE = ros2_device("mhp_ur5e", device={},
    signals={"joint_positions": ("/joint_states", "sensor_msgs/msg/JointState", "position"),
             "estop_clear": ("/safety/estop_clear", "std_msgs/msg/Bool", "data")},
    settings={"speed_scale": ("/speed_scaling", "std_msgs/msg/Float64", "data", {"min": 0.05, "max": 1.0})},
    actions={"move_to_joints": ("/follow_joint_trajectory", "control_msgs/action/FollowJointTrajectory",
                                {"interlocks": ["estop_clear"], "duration": "long"})},
    estop=("/ur_hardware_interface/dashboard/stop", "std_srvs/srv/Trigger"))

Source your ROS 2 workspace first so rclpy and message packages import. Goal messages are built from params, so nested dicts must match the message layout; put an example in the action's examples.

Anything else: bindings

from openmhp.adapters import BoundDriver, Signal, Setting, Action
DEVICE = BoundDriver(device={},
    signals=[Signal("plate_temperature", read=lambda: plc.read(0x10), unit="degC")],
    settings=[Setting("target_temperature", write=lambda v: plc.write(0x20, v), limits={"min": 20, "max": 300})],
    actions=[Action("shutdown", run=lambda job, p: plc.write(0x21, 0), approval="confirm")],
    estop=lambda: plc.write(0x21, 0))

A whole fleet

The openmhp-adapt-fleet skill walks a harness through an inventory, one package per device, a directory manifest and the bridge config, with a proof checklist at the end. By hand, the steps are: one folder per device as above, serve_fleet.py to run them, build_manifest.py to index them, mhp serve-directory, then mhp-mcp --directory. Adapters do not exempt a device from the safety gates: a layer without limits still gets limits in MHP.