madsci.common 0.5.4

The Modular Autonomous Discovery for Science (MADSci) Common Definitions and Utilities.

MADSci Common

Shared types, utilities, validators, base classes and other common code used across the MADSci toolkit.

Installation

See the main README for installation options. This package is available as:

• PyPI: pip install madsci.common
• Docker: Included in ghcr.io/ad-sdl/madsci
• Dependency: Required by all other MADSci packages

Core Components

Types System

Pydantic-based data models for the entire MADSci ecosystem:

# Import types organized by subsystem
from madsci.common.types.workflow_types import WorkflowDefinition
from madsci.common.types.node_types import NodeDefinition
from madsci.common.types.experiment_types import ExperimentDesign
from madsci.common.types.datapoint_types import ValueDataPoint

Available type modules:

• action_types: Action definitions, parameters, and flexible return types
• experiment_types: Experiment campaigns, designs, runs
• workflow_types: Workflow and step definitions with enhanced datapoint handling
• node_types: Node configurations and status
• datapoint_types: Data storage and retrieval
• event_types: Event logging and querying
• resource_types: Resource management and tracking
• location_types: Location management and resource attachments
• parameter_types: Enhanced parameter validation and serialization
• auth_types: Ownership and authentication
• base_types: Foundation classes and utilities

Utilities

Common helper functions and validators:

from madsci.common.utils import (
    utcnow, new_ulid_str, is_valid_ulid, extract_datapoint_ids,
    threaded_task, threaded_daemon, prompt_from_pydantic_model
)
from madsci.common.validators import ulid_validator
from madsci.common.serializers import serialize_to_yaml

# Generate unique IDs (ULID format)
experiment_id = new_ulid_str()

# UTC timestamps
timestamp = utcnow()

# YAML serialization
yaml_content = serialize_to_yaml(my_pydantic_model)

# ULID validation
is_valid = ulid_validator(experiment_id)
# Alternative validation
is_valid_alt = is_valid_ulid(experiment_id)

# Extract datapoint IDs from complex data structures
data_with_ids = {"result": ["01ARZ3NDEKTSV4RRFFQ69G5FAV", "01BX5ZZKBKACTAV9WEVGEMMVRZ"]}
datapoint_ids = extract_datapoint_ids(data_with_ids)

# Threading decorators for background tasks
@threaded_task
def background_job(data):
    # Long-running task
    pass

@threaded_daemon
def daemon_process():
    # Background daemon that stops when main thread exits
    pass

# Interactive model creation
from madsci.common.types.base_types import MadsciBaseModel

class MyModel(MadsciBaseModel):
    name: str
    value: int

# Prompt user to fill model fields interactively
user_data = prompt_from_pydantic_model(MyModel, "Enter model data")
my_instance = MyModel(**user_data)

Settings Framework

Hierarchical configuration system using Pydantic Settings:

from madsci.common.types.base_types import MadsciBaseSettings

class MyManagerSettings(MadsciBaseSettings):
    server_url: str = "http://localhost:8000"
    database_url: str = "mongodb://localhost:27017"
    # Supports env vars, CLI args, config files

settings = MyManagerSettings()

Configuration sources (in precedence order):

1. Command line arguments
2. Environment variables
3. Subsystem-specific files (workcell.env, event.yaml)
4. Generic files (.env, settings.yaml)
5. Default values

Configuration options: See Configuration.md and example_lab/managers/ for examples.

ULID Best Practices

MADSci uses ULID (Universally Unique Lexicographically Sortable Identifier) for all ID generation throughout the system:

from madsci.common.utils import new_ulid_str, is_valid_ulid

# Generate new IDs
resource_id = new_ulid_str()
experiment_id = new_ulid_str()

# Validate ULID format
if is_valid_ulid(some_id):
    # Process valid ULID
    pass

When to use ULIDs:

• All resource identifiers (experiments, workflows, datapoints, etc.)
• Database primary keys
• Event tracking and correlation
• Any case requiring unique, sortable identifiers

Benefits over UUIDs:

• Performance: More efficient generation and comparison
• Lexicographical sorting: Natural time-based ordering
• Timestamp preservation: First 48 bits encode creation time
• URL-safe: Uses Crockford's Base32 encoding
• Collision resistance: 80 bits of randomness per millisecond

Validation patterns:

from madsci.common.validators import ulid_validator
from pydantic import Field

class MyModel(MadsciBaseModel):
    id: str = Field(default_factory=new_ulid_str, title="Resource ID")
    parent_id: Optional[str] = Field(None, title="Parent Resource ID")

    @field_validator("parent_id")
    @classmethod
    def validate_parent_id(cls, v):
        if v is not None:
            return ulid_validator(v)
        return v

Error Handling

MADSci provides standardized error handling patterns using the Error class and specific exceptions:

from madsci.common.types.base_types import Error
from madsci.common.exceptions import (
    ActionNotImplementedError, WorkflowFailedError,
    ExperimentCancelledError, LocationNotFoundError
)

# Create errors from exceptions
try:
    # Some operation that might fail
    pass
except ValueError as e:
    error = Error.from_exception(e)
    # Error has: message, error_type, logged_at

# Create errors manually
error = Error(
    message="Custom error occurred",
    error_type="ValidationError"
)

# MADSci-specific exceptions
raise ActionNotImplementedError("Action 'analyze_sample' not implemented")
raise WorkflowFailedError("Sample preparation workflow failed at step 3")
raise ExperimentCancelledError("User cancelled experiment 'batch_synthesis'")
raise LocationNotFoundError("Location 'sample_rack_1' not found in lab")

Error handling in actions:

def my_action(self, sample_id: str) -> ActionResult:
    try:
        # Action implementation
        result = process_sample(sample_id)
        return self.request.succeeded(json_result=result)
    except Exception as e:
        # Convert exception to MADSci Error
        error = Error.from_exception(e)
        return self.request.failed(errors=[error])

Usage Patterns

Creating Custom Types

from madsci.common.types.base_types import MadsciBaseModel
from pydantic import Field
from typing import Optional

class MyCustomType(MadsciBaseModel):
    name: str = Field(description="Object name")
    value: float = Field(gt=0, description="Positive value")
    metadata: dict = Field(default_factory=dict)
    optional_field: Optional[str] = Field(None, description="Optional parameter")

# Automatic validation, serialization to JSON/YAML
obj = MyCustomType(name="test", value=42.0)
json_str = obj.model_dump_json()
yaml_str = obj.model_dump_yaml()  # YAML serialization supported

Action Parameter Types

from madsci.common.types.action_types import ActionFiles
from pathlib import Path
from typing import Union

class ProcessingFiles(ActionFiles):
    """Custom file collection for action returns."""
    log_file: Path
    results_file: Path
    optional_config: Optional[Path] = None

# Complex parameter handling
def my_action(
    sample_id: str,
    parameters: dict[str, Union[str, int, float]],
    file_input: Path,
    optional_metadata: Optional[dict] = None
) -> ProcessingFiles:
    """Action with complex parameter types and file return."""
    # MADSci automatically handles serialization/deserialization
    pass

Extending Base Settings

from madsci.common.types.base_types import MadsciBaseSettings
from pydantic import Field
from typing import Optional

class CustomSettings(MadsciBaseSettings, env_prefix="CUSTOM_"):
    api_key: str = Field(description="API authentication key")
    timeout: int = Field(default=30, description="Request timeout")
    advanced_config: Optional[dict[str, str]] = Field(
        default=None,
        description="Advanced configuration options"
    )

# Reads from CUSTOM_API_KEY, CUSTOM_TIMEOUT environment variables
settings = CustomSettings()

Working with Complex Types

from madsci.common.types.parameter_types import ParameterDefinition
from typing import Union, Optional, get_origin

# Handle complex nested types
complex_type = dict[str, list[Union[int, float]]]
origin = get_origin(complex_type)  # Returns dict

# Parameter validation for action arguments
param_def = ParameterDefinition(
    name="complex_param",
    type_hint=complex_type,
    required=True,
    description="Complex nested parameter"
)

Manager Base Class

Create standardized manager services with AbstractManagerBase:

from madsci.common.manager_base import AbstractManagerBase
from madsci.common.types.base_types import MadsciBaseSettings, MadsciBaseModel
from madsci.common.types.manager_types import ManagerHealth

class MyManagerSettings(MadsciBaseSettings):
    model_config = {"env_prefix": "MY_MANAGER_"}
    database_url: str = "mongodb://localhost:27017"

class MyManagerDefinition(MadsciBaseModel):
    name: str = "My Manager"
    description: str = "Custom manager service"

class MyManager(AbstractManagerBase[MyManagerSettings, MyManagerDefinition]):
    SETTINGS_CLASS = MyManagerSettings
    DEFINITION_CLASS = MyManagerDefinition
    # ENABLE_ROOT_DEFINITION_ENDPOINT = True  # Default: enabled

    def get_health(self) -> ManagerHealth:
        """Override to implement custom health checks."""
        return ManagerHealth(healthy=True, description="Manager is healthy")

# Create and run the manager
manager = MyManager()
manager.run_server()  # Starts FastAPI server with auto-generated endpoints

Built-in endpoints:

• GET / - Manager definition (configurable with ENABLE_ROOT_DEFINITION_ENDPOINT)
• GET /definition - Manager definition (always available)
• GET /health - Health status

Configurable root endpoint:

class CustomManager(AbstractManagerBase[Settings, Definition]):
    ENABLE_ROOT_DEFINITION_ENDPOINT = False  # Disable root endpoint
    # Allows custom root endpoint implementation or static file serving for UIs