base#

Base procedure classes for implementing experimental workflows in Mesofield.

This module defines a generic Procedure orchestrator that contains zero device-specific logic. Custom experiment subclasses live outside the package (typically under experiments/<name>/procedure.py) and are discovered via load_procedure_from_config(), which reads the optional procedure_file and procedure_class fields from experiment.json.

Lifecycle (subclass hooks shown in bold):

  1. initialize_hardware – bring devices up

  2. prerun – subclass hook (default: no-op)

  3. hardware.arm_all – per-run prep on every device

  4. connect hardware.primary.signals.finished -> _cleanup_procedure

  5. on_started – subclass hook (default: no-op)

  6. hardware.start_all

  7. on_finished – subclass hook (default: no-op)

  8. save_data + cleanup

mesofield.base.processor(*, camera, plot=False, **plot_kwargs)[source]#

Mark a Procedure method as a per-frame compute function.

The decorated function is called as func(self, img, idx, ts) and should return a float | None. At procedure init the framework builds a FrameProcessor that wraps it, attaches it to the hardware device whose device_id matches camera, registers it on DataManager, and (when plot=True) tells the GUI to add a SerialWidget.

plot_kwargs are forwarded straight to the widget — recognized keys: label, value_label, value_units, y_range, value_scale, max_points.

Example:

class MyProcedure(Procedure):
    @processor(camera="meso", plot=True, label="Frame Mean")
    def frame_mean(self, img, idx, ts):
        return float(img.mean())
Parameters:
class mesofield.base.ProcedureSignals[source]#

Bases: QObject

All procedure-level signals that a Qt GUI can connect to.

class mesofield.base.Procedure[source]#

Bases: object

Generic orchestrator for a Mesofield experiment.

Subclass this in experiments/<name>/procedure.py and override the extension hooks (prerun(), on_started(), on_finished()) and/or the lifecycle methods (run(), save_data(), cleanup()) as needed. The base class never references a specific device type – multi-camera sync is driven by the YAML primary: true flag and HardwareManager.start_all/stop_all.

__init__(hardware=None, experiment=None)[source]#
Parameters:

  • hardware (str | None)

  • experiment (str | None)

initialize_hardware()[source]#

Boot up hardware and a DataManager.

Return type:

None

load_config(hardware=None, experiment=None)[source]#

Hot-load a hardware YAML and/or experiment JSON into the live config.

The two inputs are independent: loading experiment params never touches hardware. Callers pass explicit paths (the GUI wizard resolves them from its pickers).

Parameters:

  • hardware (str | None)

  • experiment (str | None)

Return type:

None

define_config()[source]#

Subclass hook to declare experiment parameters in Python.

Override to return a @dataclass instance or a plain mapping; it is applied to config via ExperimentConfig.load_dict(), superseding any experiment.json. Default None -> load JSON.

Return type:

Any

define_hardware()[source]#

Subclass hook to construct hardware devices in Python.

Override to return a list of pre-built device objects (imported and instantiated in the procedure file). They are handed to a fresh HardwareManager, superseding any hardware.yaml. Default None -> load YAML. Device classes should be decorated with @DeviceRegistry.register(...) so the setup can later be exported to a hardware.yaml rig file via HardwareManager.to_yaml.

Return type:

Any

prerun()[source]#

Subclass hook called before arming devices. Override as needed.

Return type:

None

on_started()[source]#

Subclass hook called immediately after start_all.

Return type:

None

on_finished()[source]#

Subclass hook called immediately after the primary device finishes.

Return type:

None

run()[source]#

Drive a standard experiment run.

Subclasses may override, but the default body is generic and handles any combination of devices declared in hardware.yaml.

Return type:

None

cleanup()[source]#

Public cleanup entry-point (manual stop).

Return type:

None

run_until_finished(timeout=None)[source]#

Run the procedure and block until cleanup completes.

Starts the procedure via run(), then waits for the procedure_finished (or procedure_error) signal. Handles KeyboardInterrupt and timeout by invoking cleanup() automatically, so callers (e.g. __main__ blocks in experiment scripts) do not need to wire up their own threading events.

Parameters#

timeout:

Optional hard ceiling in seconds. When None (default), waits indefinitely for the primary device’s finished signal. When provided, forces cleanup if the deadline passes.

Returns#

bool

True if the procedure finished on its own, False if cleanup was forced by timeout or interrupt.

Parameters:

timeout (float | None)

Return type:

bool

manifest_extra()[source]#

Override to inject extra session-level metadata into the AcquisitionManifest’s extra block. Default: empty.

Return type:

Dict[str, Any]

mesofield.base.load_procedure_from_config(target)[source]#

Build the right Procedure for a launch target.

target may be a hardware.yaml, an experiment.json, a scripted procedure.py, an experiment directory, or None. Discovery of the hardware/experiment pair is delegated to _resolve_target().

When the experiment JSON declares procedure_file + procedure_class, that subclass is imported and used; otherwise a base Procedure.

Parameters:

target (str | None)

Return type:

Procedure