wei.core.state_manager

StateManager for WEI

Classes

StateManager()

Manages state for WEI, providing transactional access to reading and writing state with optimistic check-and-set and locking.

class wei.core.state_manager.StateManager

Manages state for WEI, providing transactional access to reading and writing state with optimistic check-and-set and locking.

__init__() → None: Initialize a StateManager for a given workcell.

campaign_lock(campaign_id: str) → Redlock: Get a lock on a particular campaign. This should be called before editing a Campaign.

clear_state(reset_locations: bool = True, clear_workflow_runs: bool = False) → None: Clears the state of the workcell, optionally leaving the locations state intact.

clear_workcell() → None: Empty the workcell definition

delete_campaign(campaign_id: str) → None: Deletes a campaign by ID

delete_experiment(experiment_id: str) → None: Deletes an experiment by ID

delete_location(location_name: str) → None: Deletes a location by name

delete_module(module_name: str) → None: Deletes a module by name

delete_workflow_run(run_id: str) → None: Deletes a workflow by ID

property error: str: Latest error on the server

experiment_lock(experiment_id: str) → Redlock: Get a lock on a particular experiment. This should be called before editing an experiment.

get_all_campaigns() → Dict[str, Campaign]: Returns all campaigns

get_all_datapoints() → Dict[str, DataPoint]: Returns all datapoints

get_all_events() → Dict[str, Event]: Returns all events

get_all_experiments() → Dict[str, Experiment]: Returns all experiments

get_all_locations() → Dict[str, Location]: Returns all locations

get_all_modules() → Dict[str, Module]: Returns all modules

get_all_workflow_runs() → Dict[str, WorkflowRun]: Returns all workflow runs

get_campaign(campaign_id: str) → Campaign: Returns a campaign by ID

get_datapoint(data_id: str) → DataPoint: Returns an event by ID

get_event(event_id: str) → Event: Returns an event by ID

get_experiment(experiment_id: str) → Experiment: Returns an experiment by ID

get_location(location_name: str) → Location: Returns a location by name

get_module(module_name: str) → Module: Returns a module by name

get_state() → Dict[str, Dict[Any, Any]]: Return a dict containing the current state of the workcell.

get_workcell() → Workcell: Returns the current workcell as a Workcell object

get_workcell_id() → str: Returns the workcell ID

get_workflow_run(run_id: str) → WorkflowRun: Returns a workflow by ID

has_state_changed() → bool: Returns True if the state has changed since the last time this method was called

property locked: bool: Get the lock state of the workcell

mark_state_changed() → int: Marks the state as changed and returns the current state change counter

property paused: bool: Get the pause state of the workcell

set_campaign(campaign: Campaign) → None: Sets a campaign by ID

set_datapoint(datapoint: DataPoint) → None: Sets an event by ID

set_event(event: Event) → None: Sets an event by ID

set_experiment(experiment: Experiment) → None: Sets an experiment by ID

set_location(location_name: str, location: Location | Dict[str, Any]) → None: Sets a location by name

set_module(module_name: str, module: Module | ModuleDefinition | Dict[str, Any]) → None: Sets a module by name

set_workcell(workcell: Workcell) → None: Sets the active workcell

set_workflow_run(wf: WorkflowRun) → None: Sets a workflow by ID

property shutdown: bool: Get the shutdown state of the workcell

update_campaign(campaign_id: str, func: Callable[[...], Any], *args: Any, **kwargs: Any) → None: Updates the state of a campaign.

update_experiment(experiment_id: str, func: Callable[[...], Any], *args: Any, **kwargs: Any) → None: Updates the state of an experiment.

update_location(location_name: str, func: Callable[[...], Any], *args: Any, **kwargs: Any) → None: Updates the state of a location.

update_module(module_name: str, func: Callable[[...], Any], *args: Any, **kwargs: Any) → None: Updates the state of a module.

update_workflow_run(run_id: str, func: Callable[[...], Any], *args: Any, **kwargs: Any) → None: Updates the state of a workflow.

wc_state_lock() → Redlock: Gets a lock on the workcell’s state. This should be called before any state updates are made, or where we don’t want the state to be changing underneath us (i.e., in the engine).

property wc_status: ModuleStatus: The current status of the workcell