ad-sdl/madsci/madsci-nodes

madsci new node
# or: madsci new module  (full module with interface, types, tests, Dockerfile)

from madsci.node_module.rest_node_module import RestNode
from madsci.node_module.helpers import action
class MyInstrumentNode(RestNode):
    """Node for controlling MyInstrument."""
    def startup_handler(self):
        """Initialize instrument connections."""
        self.instrument = MyInstrumentDriver()
    def shutdown_handler(self):
        """Clean up instrument connections."""
        self.instrument.disconnect()
    @action(name="measure", description="Take a measurement")
    def measure(self, sample_id: str, temperature: float = 25.0) -> dict:
        result = self.instrument.measure(sample_id, temp=temperature)
        return {"sample_id": sample_id, "value": result}
if __name__ == "__main__":
    node = MyInstrumentNode()
    node.start_node()

# RestNodeConfig key fields:
node_name: Optional[str]       # Human name (defaults to class name)
node_id: Optional[str]         # ULID (auto-generated)
node_type: Optional[NodeType]  # DEVICE, COMPUTE, etc.
node_url: AnyUrl               # Default: http://127.0.0.1:2000
status_update_interval: float  # Default: 2.0s
state_update_interval: float   # Default: 2.0s
enable_registry_resolution: bool  # Default: True
enable_rate_limiting: bool     # Default: True

node = MyNode(node_config=RestNodeConfig(node_url="http://0.0.0.0:3000"))

from madsci.node_module.helpers import action
# With explicit parameters
@action(name="dispense", description="Dispense liquid", blocking=True)
def dispense(self, volume_ul: float, well: str) -> dict:
    ...
# Minimal (name from function, description from docstring, blocking=True)
@action
def home(self):
    """Home all axes to reference position."""
    self.instrument.home()

@action
def process(self, sample_id: str, count: int = 1, options: Optional[dict] = None) -> dict:
    ...

from pathlib import Path
@action
def analyze_image(self, image: Path) -> dict:
    """Single file upload."""
    data = image.read_bytes()
    return {"size": len(data)}
@action
def batch_process(self, files: list[Path]) -> dict:
    """Multiple file upload."""
    return {"count": len(files)}

from madsci.common.types.location_types import LocationArgument
@action
def transfer(self, source: LocationArgument, destination: LocationArgument) -> dict:
    ...

from typing import Annotated
@action
def calibrate(self, offset: Annotated[float, "Offset in mm, range -10 to 10"]) -> dict:
    ...

from madsci.common.types.action_types import ActionFiles, ActionDatapoints
from madsci.common.types.data_types import DataPoint
# Plain dict/Pydantic model -> json_result
@action
def measure(self) -> dict:
    return {"value": 42}
# Path -> single file result
@action
def capture(self) -> Path:
    return Path("/tmp/image.png")
# ActionFiles -> multiple named files
@action
def generate_report(self) -> ActionFiles:
    return ActionFiles(report=Path("report.pdf"), data=Path("data.csv"))
# Tuple -> (json_result, files, datapoints)
@action
def full_result(self) -> tuple[dict, ActionFiles, ActionDatapoints]:
    return (
        {"status": "ok"},
        ActionFiles(output=Path("out.csv")),
        ActionDatapoints(dp=DataPoint(label="measurement", value=42)),
    )

__init__()
  ├── Load NodeConfig (walk-up discovery)
  ├── Synthesize NodeInfo
  ├── Resolve registry identity (if enabled)
  ├── Configure clients (EventClient, ResourceClient, DataClient)
  └── Discover @action methods
 
start_node()
  ├── Establish event_client_context
  ├── Log startup
  └── _startup() thread:
      ├── startup_handler()       ← Override this
      ├── template_handler()      ← Override for resource/location templates
      └── Status/state update loops (every N seconds)
          ├── status_handler()    ← Override to populate node_status
          └── state_handler()     ← Override to populate node_state
 
Per-action execution:
  ├── Parse args (TypeAnalyzer)
  ├── Validate required args
  ├── Check node ready (not busy/locked/errored)
  ├── Execute in daemon thread (with _action_lock if blocking)
  ├── Process result -> ActionResult
  └── Update action_history & log status
 
shutdown:
  ├── shutdown_handler()          ← Override for cleanup
  ├── Release registry identity
  └── Teardown clients

# Operational flags:
busy: bool          # Currently executing an action
paused: bool        # Manually paused
locked: bool        # Locked via admin command
stopped: bool       # Node stopped
errored: bool       # Node in error state
disconnected: bool  # Lost connection
initializing: bool  # Still starting up
# Computed:
ready: bool         # True only if no flags set and no waiting_for_config
description: str    # Human-readable reason if not ready

class MyNode(RestNode):
    resource_templates: ClassVar[list[NodeResourceTemplateDefinition]] = [
        NodeResourceTemplateDefinition(
            template_name="sample_plate",
            description="96-well sample plate",
            version="1.0",
        ),
    ]
    # Representation templates define how this node "sees" a location
    location_representation_templates: ClassVar[
        list[NodeRepresentationTemplateDefinition]
    ] = [
        NodeRepresentationTemplateDefinition(
            template_name="mynode_slot_repr",
            default_values={"slot_index": 0},
            required_overrides=["slot_index"],
            version="1.0.0",
        ),
    ]
    # Intrinsic locations are auto-created on startup with '{node_name}.' prefix
    intrinsic_locations: ClassVar[list[NodeIntrinsicLocationDefinition]] = [
        NodeIntrinsicLocationDefinition(
            location_name="slot_1",  # becomes "{node_name}.slot_1"
            representation_template_name="mynode_slot_repr",
            representation_overrides={"slot_index": 1},
            resource_template_name="sample_plate",
        ),
    ]
    def template_handler(self):
        """Called at startup to register templates. Override for custom logic."""
        super().template_handler()

@action
def measure(self) -> dict:
    value = self.instrument.read()
    # Single value datapoint
    dp_id = self.create_and_upload_value_datapoint(value=value, label="measurement")
    # File datapoint
    file_id = self.create_and_upload_file_datapoint(
        file_path=Path("output.csv"), label="raw_data"
    )
    # Custom datapoint
    from madsci.common.types.data_types import DataPoint
    dp = DataPoint(label="custom", value={"nested": "data"})
    self.upload_datapoint(dp)
    return {"measurement": value, "datapoint_id": dp_id}

from fastapi.testclient import TestClient
def test_node_action():
    node = MyInstrumentNode(
        node_config=RestNodeConfig(
            enable_registry_resolution=False,  # Critical for tests
            node_url="http://127.0.0.1:2000",
        )
    )
    node.start_node(testing=True)  # Configures routes without starting uvicorn
    client = TestClient(node.rest_api)
    # Check info
    response = client.get("/info")
    assert response.status_code == 200
    assert "measure" in response.json()["actions"]
    # Run action (no file params)
    response = client.post("/action/measure", json={"sample_id": "ABC123"})
    assert response.status_code == 200
    action_id = response.json()["action_id"]
    # Start and get result
    client.post(f"/action/measure/{action_id}/start")
    import time; time.sleep(0.5)
    result = client.get(f"/action/measure/{action_id}/result")
    assert result.json()["json_result"]["sample_id"] == "ABC123"