Server

• icon.server.api
• icon.server.data_access.models.enums
• icon.server.data_access.models.sqlite
• icon.server.data_access.repositories
• icon.server.hardware_processing
• icon.server.post_processing
• icon.server.pre_processing
• icon.server.scheduler
• icon.server.utils.types
• icon.server.web_server

icon.server.api

This module defines the API layer of ICON, implemented as a pydase.DataService.

The main entry point is the APIService, which is exposed by the IconServer. The IconServer itself is a pydase.Server hosting the API.

Structure

The APIService aggregates multiple “controller” services as attributes. Each controller is itself a pydase.DataService exposing related API methods.

Background tasks

Controllers can define periodic pydase tasks, which are asyncio tasks automatically started with the service.

api_service

APIService

APIService(
    pre_processing_event_queues: list[Queue[UpdateQueue]],
    experiment_library_client: ReconfigurableExperimentLibraryClient,
    devices: Devices,
)

Bases: DataService

Aggregates ICON’s API controllers and manages background tasks.

The APIService groups multiple controllers, each of which is a pydase.DataService exposing related API methods. It also defines background tasks for keeping experiment and parameter metadata in sync with the experiment library and InfluxDB.

Note

Controllers are pydase.DataService instances exposed as attributes to group related API methods. Background tasks are implemented with pydase tasks.

Create a new APIService.

pre_processing_event_queues: Queues used by ScansController to notify pre-processing workers. experiment_library_client: Client for an experiment library devices: Controllers for the hardware devices

Source code in src/icon/server/api/api_service.py

def __init__(
    self,
    pre_processing_event_queues: list[multiprocessing.Queue[UpdateQueue]],
    experiment_library_client: ReconfigurableExperimentLibraryClient,
    devices: Devices,
) -> None:
    """Create a new APIService.

    Args:
    pre_processing_event_queues: Queues used by `ScansController` to notify
        pre-processing workers.
    experiment_library_client: Client for an experiment library
    devices: Controllers for the hardware devices
    """
    super().__init__()

    self.devices = DevicesController()
    """Controller for managing external pydase-based devices."""
    self.parameters = ParametersController()
    """Controller for parameter metadata and shared parameter values."""
    self.scheduler = SchedulerController(
        devices_controller=self.devices,
        parameters_controller=self.parameters,
    )
    """Controller to submit, inspect, and cancel scheduled jobs."""
    self.experiments = ExperimentsController()
    """Controller for experiment metadata."""
    self.config = ConfigurationController()
    """Controller for managing and updating the application's configuration."""
    self.data = ExperimentDataController()
    """Controller for accessing stored experiment data."""
    self.scans = ScansController(
        pre_processing_update_queues=pre_processing_event_queues
    )
    """Controller for triggering update events for jobs across multiple worker
    processes."""
    self.status = StatusController(devices)
    """Controller for system status monitoring."""
    self._experiment_library_client = experiment_library_client

config instance-attribute

config = ConfigurationController()

Controller for managing and updating the application’s configuration.

data instance-attribute

data = ExperimentDataController()

Controller for accessing stored experiment data.

devices instance-attribute

devices = DevicesController()

Controller for managing external pydase-based devices.

experiments instance-attribute

experiments = ExperimentsController()

Controller for experiment metadata.

parameters instance-attribute

parameters = ParametersController()

Controller for parameter metadata and shared parameter values.

scans instance-attribute

scans = ScansController(
    pre_processing_update_queues=pre_processing_event_queues
)

Controller for triggering update events for jobs across multiple worker processes.

scheduler instance-attribute

scheduler = SchedulerController(
    devices_controller=devices,
    parameters_controller=parameters,
)

Controller to submit, inspect, and cancel scheduled jobs.

status instance-attribute

status = StatusController(devices)

Controller for system status monitoring.

configuration_controller

ConfigurationController

Bases: DataService

Controller for managing and updating the application’s configuration.

This class provides an API to get and update the configuration, validate it, and save the updated configuration back to the source file.

get_config

get_config() -> dict[str, Any]

Get current configuration dictionary.

Source code in src/icon/server/api/configuration_controller.py

def get_config(self) -> dict[str, Any]:
    """Get current configuration dictionary."""
    return get_config().model_dump()

update_config_option

update_config_option(key: str, value: Any) -> bool

Update a specific configuration option.

Traverses the configuration using the dot-separated key, updates the specified value, validates the entire configuration, and saves the changes.

Parameters:

Name	Type	Description	Default
key	str	The dot-separated key of the configuration option (e.g., “experiment_library.git_repository”).	required
value	Any	The new value for the configuration option.	required

Returns:

Type	Description
bool	True if the update is successful, False otherwise.

Source code in src/icon/server/api/configuration_controller.py

def update_config_option(self, key: str, value: Any) -> bool:
    """Update a specific configuration option.

    Traverses the configuration using the dot-separated key, updates the specified
    value, validates the entire configuration, and saves the changes.

    Args:
        key:
            The dot-separated key of the configuration option (e.g.,
            "experiment_library.git_repository").
        value:
            The new value for the configuration option.

    Returns:
        True if the update is successful, False otherwise.
    """
    try:
        current_config = get_config().model_dump()
        set_nested(current_config, key, value)

        # Validate the updated configuration
        updated_config = ServiceConfig(config_sources=DataSource(current_config))

        # Save the updated configuration back to the file
        save_config(updated_config)
        emit_queue.put(
            {"event": "config.update", "data": updated_config.model_dump()}
        )
    except KeyError:
        logger.exception("Failed to update configuration")
        return False
    return True

set_nested

set_nested(
    config: dict[str, Any], nested_key: str, value: Any
) -> None

Set a value in a nested dict.

Source code in src/icon/server/api/configuration_controller.py

def set_nested(config: dict[str, Any], nested_key: str, value: Any) -> None:
    """Set a value in a nested dict."""
    current: dict[str, Any] | list[Any] = config
    *fields, last_field = parse_config_key(nested_key)
    # Traverse to the nested key
    for field in fields:
        if isinstance(current, dict) and (
            not isinstance(field, str) or field not in current
        ):
            raise KeyError(f"Key {nested_key!r} not found in configuration.")
        if isinstance(current, list) and (
            not isinstance(field, int) or field >= len(current)
        ):
            raise IndexError(
                f"Configuration error: Index out of range: {field} in {nested_key!r}"
            )
        current = current[field]  # type: ignore[index]

    # Update the value
    current[last_field] = value  # type: ignore[index]

devices_controller

DeviceParameterValueyType module-attribute

DeviceParameterValueyType = int | bool | float

Allowed primitive types for device parameter values.

A parameter value sent to or retrieved from a device may be one of these basic types. Quantities with units are handled separately via pydase.units.Quantity.

DevicesController

DevicesController()

Bases: DataService

Controller for managing external pydase-based devices.

Maintains client connections to configured devices, exposes helpers to add/update device entries in SQLite, and provides async accessors for device parameter values through pydase proxies. Also discovers scannable device parameters for integration with ICON scans.

Source code in src/icon/server/api/devices_controller.py

def __init__(self) -> None:
    super().__init__()
    self._devices: dict[str, pydase.Client] = {}
    self.device_proxies: dict[str, ProxyClass] = {}
    """Live pydase proxies keyed by device name."""

device_proxies instance-attribute

device_proxies: dict[str, ProxyClass] = {}

Live pydase proxies keyed by device name.

add_device

add_device(
    *,
    name: str,
    url: str,
    status: Literal["disabled", "enabled"] = "enabled",
    description: str | None = None,
    retry_delay_seconds: float = 0.0,
    retry_attempts: int = 3,
) -> Device

Create a device record in SQLite and (optionally) connect to it.

If status=="enabled", a non-blocking pydase client is created and its proxy is registered.

Parameters:

Name	Type	Description	Default
name	str	Unique device name.	required
url	str	pydase server URL of the device.	required
status	Literal['disabled', 'enabled']	Whether the device should be connected immediately.	'enabled'
description	str | None	Optional human-readable description.	None
retry_delay_seconds	float	Backoff delay used by device-side logic.	0.0
retry_attempts	int	Number of retries used by device-side logic.	3

Returns:

Type	Description
Device	The Device SQLAlchemy model.

Source code in src/icon/server/api/devices_controller.py

def add_device(
    self,
    *,
    name: str,
    url: str,
    status: Literal["disabled", "enabled"] = "enabled",
    description: str | None = None,
    retry_delay_seconds: float = 0.0,
    retry_attempts: int = 3,
) -> Device:
    """Create a device record in SQLite and (optionally) connect to it.

    If `status=="enabled"`, a non-blocking pydase client is created and its
    proxy is registered.

    Args:
        name: Unique device name.
        url: pydase server URL of the device.
        status: Whether the device should be connected immediately.
        description: Optional human-readable description.
        retry_delay_seconds: Backoff delay used by device-side logic.
        retry_attempts: Number of retries used by device-side logic.

    Returns:
        The `Device` SQLAlchemy model.
    """
    device = DeviceRepository.add_device(
        device=Device(
            name=name,
            url=url,
            status=DeviceStatus(status),
            description=description,
            retry_delay_seconds=retry_delay_seconds,
            retry_attempts=retry_attempts,
        )
    )

    if status == "enabled":
        client = pydase.Client(
            url=device.url,
            client_id="icon-devices-controller",
            block_until_connected=False,
        )
        self._devices[name] = client
        self.device_proxies[name] = client.proxy

    return device

get_devices_by_status

get_devices_by_status(
    *, status: DeviceStatus | None = None
) -> dict[str, DeviceDict]

List devices (optionally filtered by status) with reachability & scan info.

Augments each device entry with

• reachable: Whether a live proxy is connected.
• scannable_params: Flat list of scannable parameter access paths.

Parameters:

Name	Type	Description	Default
status	DeviceStatus | None	Optional filter (ENABLED, DISABLED, or None for all).	None

Returns:

Type	Description
dict[str, DeviceDict]	Mapping from device name to a DeviceDict payload suitable for the API.

Source code in src/icon/server/api/devices_controller.py

def get_devices_by_status(
    self, *, status: DeviceStatus | None = None
) -> dict[str, DeviceDict]:
    """List devices (optionally filtered by status) with reachability & scan info.

    Augments each device entry with

    - `reachable`: Whether a live proxy is connected.
    - `scannable_params`: Flat list of scannable parameter access paths.

    Args:
        status: Optional filter (`ENABLED`, `DISABLED`, or `None` for all).

    Returns:
        Mapping from device name to a `DeviceDict` payload suitable for the API.
    """
    device_dict: dict[str, DeviceDict] = {
        device.name: SQLAlchemyDictEncoder.encode(device)
        for device in DeviceRepository.get_devices_by_status(status=status)
    }

    for name, value in device_dict.items():
        client = self._devices.get(name, None)
        value["reachable"] = False
        value["scannable_params"] = []

        if client is not None:
            value["reachable"] = client.proxy.connected
            value["scannable_params"] = get_scannable_params_list(
                client.proxy.serialize(),
                prefix=f'devices.device_proxies["{name}"].',
            )

    return device_dict

get_parameter_value async

get_parameter_value(*, name: str, parameter_id: str) -> Any

Get a parameter value from a connected device.

Logs a warning if the device is not connected or not found.

Parameters:

Name	Type	Description	Default
name	str	Device name.	required
parameter_id	str	Access path on the device service.	required

Returns:

Type	Description
Any	The parameter value as returned by the device, or None if the device is unreachable or unknown.

Source code in src/icon/server/api/devices_controller.py

async def get_parameter_value(self, *, name: str, parameter_id: str) -> Any:
    """Get a parameter value from a connected device.

    Logs a warning if the device is not connected or not found.

    Args:
        name: Device name.
        parameter_id: Access path on the device service.

    Returns:
        The parameter value as returned by the device, or `None` if the device is
            unreachable or unknown.
    """
    try:
        return await asyncio.to_thread(
            self._devices[name].get_value, access_path=parameter_id
        )
    except BadNamespaceError:
        logger.warning(
            'Could not get %r. Device %r at ("%s") is not connected.',
            parameter_id,
            name,
            self._devices[name]._url,
        )
    except KeyError:
        logger.warning("Device with name %r not found. Is it enabled?", name)

update_device

update_device(
    *,
    name: str,
    status: Literal["disabled", "enabled"] | None = None,
    url: str | None = None,
    retry_attempts: int | None = None,
    retry_delay_seconds: float | None = None,
) -> Device

Update a device record and its live connection.

When transitioning to disabled, the client is disconnected and removed. When transitioning to enabled, a client is (re)created and registered.

Parameters:

Name	Type	Description	Default
name	str	Device name.	required
status	Literal['disabled', 'enabled'] | None	Target enable/disable status.	None
url	str | None	Updated pydase URL.	None
retry_attempts	int | None	Updated retry attempts metadata.	None
retry_delay_seconds	float | None	Updated retry delay metadata.	None

Returns:

Type	Description
Device	The updated Device model.

Source code in src/icon/server/api/devices_controller.py

def update_device(
    self,
    *,
    name: str,
    status: Literal["disabled", "enabled"] | None = None,
    url: str | None = None,
    retry_attempts: int | None = None,
    retry_delay_seconds: float | None = None,
) -> Device:
    """Update a device record and its live connection.

    When transitioning to `disabled`, the client is disconnected and removed.
    When transitioning to `enabled`, a client is (re)created and registered.

    Args:
        name: Device name.
        status: Target enable/disable status.
        url: Updated pydase URL.
        retry_attempts: Updated retry attempts metadata.
        retry_delay_seconds: Updated retry delay metadata.

    Returns:
        The updated `Device` model.
    """
    device = DeviceRepository.update_device(
        name=name,
        url=url,
        status=DeviceStatus(status) if status is not None else None,
        retry_attempts=retry_attempts,
        retry_delay_seconds=retry_delay_seconds,
    )

    if status == "disabled" and name in self._devices:
        if name in self.device_proxies:
            self.device_proxies.pop(name)
        if name in self._devices:
            client = self._devices.pop(name)
            client.disconnect()
    elif status == "enabled":
        client = pydase.Client(
            url=device.url,
            client_id="icon-devices-controller",
            block_until_connected=False,
        )
        self._devices[name] = client
        self.device_proxies[device.name] = client.proxy

    return device

update_parameter_value async

update_parameter_value(
    *,
    name: str,
    parameter_id: str,
    new_value: DeviceParameterValueyType | QuantityDict,
    type_: Literal["float", "int", "Quantity"],
) -> None

Set a parameter value on a connected device.

Performs type-normalization (float, int, or Quantity) before delegating to the device client.

Logs a warning if the device is not connected or not found.

Parameters:

Name	Type	Description	Default
name	str	Device name.	required
parameter_id	str	Access path on the device service.	required
new_value	DeviceParameterValueyType | QuantityDict	New value (native type or quantity dict).	required
type_	Literal['float', 'int', 'Quantity']	Expected type of the value for normalization.	required

Source code in src/icon/server/api/devices_controller.py

async def update_parameter_value(
    self,
    *,
    name: str,
    parameter_id: str,
    new_value: DeviceParameterValueyType | u.QuantityDict,
    type_: Literal["float", "int", "Quantity"],
) -> None:
    """Set a parameter value on a connected device.

    Performs type-normalization (`float`, `int`, or `Quantity`) before delegating
    to the device client.

    Logs a warning if the device is not connected or not found.

    Args:
        name: Device name.
        parameter_id: Access path on the device service.
        new_value: New value (native type or quantity dict).
        type_: Expected type of the value for normalization.
    """
    if type_ == "float" and not isinstance(new_value, dict):
        new_value = float(new_value)
    elif type_ == "int" and not isinstance(new_value, dict):
        new_value = int(new_value)
    elif type_ == "Quantity" and isinstance(new_value, dict):
        new_value = u.Quantity(new_value["magnitude"], new_value["unit"])  # type: ignore

    try:
        await asyncio.to_thread(
            self._devices[name].update_value,
            access_path=parameter_id,
            new_value=new_value,
        )
    except BadNamespaceError:
        logger.warning(
            'Could not set %r. Device %r at ("%s") is not connected.',
            parameter_id,
            name,
            self._devices[name]._url,
        )
    except KeyError:
        logger.warning("Device with name %r not found. Is it enabled?", name)

experiment_data_controller

ExperimentDataController

Bases: DataService

Controller for accessing stored experiment data.

Provides API methods to fetch experiment data associated with jobs.

delete_fit async

delete_fit(job_id: int, result_channel: str) -> None

Delete a fit result for a result channel.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
result_channel	str	Name of the result channel whose fit to remove.	required

Source code in src/icon/server/api/experiment_data_controller.py

async def delete_fit(
    self,
    job_id: int,
    result_channel: str,
) -> None:
    """Delete a fit result for a result channel.

    Args:
        job_id: Job identifier.
        result_channel: Name of the result channel whose fit to remove.
    """
    await asyncio.to_thread(
        delete_fit_result_by_job_id,
        job_id=job_id,
        result_channel=result_channel,
    )
    emit_queue.put(
        {
            "event": f"experiment_fit_{job_id}",
            "data": {"result_channel": result_channel, "deleted": True},
        }
    )

get_experiment_data_by_job_id async

get_experiment_data_by_job_id(
    job_id: int,
    max_transfer_bytes: int = 50000000,
    *,
    include_hardware_instructions: bool = False,
) -> dict[str, Any]

Return experiment data for a given job.

Parameters:

Name	Type	Description	Default
job_id	int	The unique identifier of the job.	required
max_transfer_bytes	int	Approximate cap on the serialised payload size in bytes. The number of data points loaded is derived from HDF5 metadata so that the response stays within this budget. Defaults to 50 MB.	50000000
include_hardware_instructions	bool	If True, include per-point pulse hardware_instructions blobs in the response. Defaults to False because those strings dominate the payload for large scans (~tens of MB) and are not needed for plotting/fitting; live updates still carry hardware_instructions per data point.	False

Returns:

Type	Description
dict[str, Any]	The experiment data linked to the job as a dict resulting
dict[str, Any]	from serializing an
dict[str, Any]	[ExperimentData][icon.server.data_access.repositories.experiment_data_repository.ExperimentData]
dict[str, Any]	instance.

Source code in src/icon/server/api/experiment_data_controller.py

async def get_experiment_data_by_job_id(
    self,
    job_id: int,
    max_transfer_bytes: int = 50_000_000,
    *,
    include_hardware_instructions: bool = False,
) -> dict[str, Any]:
    """Return experiment data for a given job.

    Args:
        job_id: The unique identifier of the job.
        max_transfer_bytes: Approximate cap on the serialised payload
            size in bytes.  The number of data points loaded is
            derived from HDF5 metadata so that the response stays
            within this budget.  Defaults to 50 MB.
        include_hardware_instructions: If True, include per-point pulse
            ``hardware_instructions`` blobs in the response.  Defaults to False
            because those strings dominate the payload for large scans
            (~tens of MB) and are not needed for plotting/fitting; live
            updates still carry ``hardware_instructions`` per data point.

    Returns:
        The experiment data linked to the job as a dict resulting
        from serializing an
        [ExperimentData][icon.server.data_access.repositories.experiment_data_repository.ExperimentData]
        instance.
    """
    result = await asyncio.to_thread(
        ExperimentDataRepository.get_experiment_data_by_job_id,
        job_id=job_id,
        max_transfer_bytes=max_transfer_bytes,
        include_hardware_instructions=include_hardware_instructions,
    )
    return asdict(result)

run_fit async

run_fit(
    job_id: int,
    result_channel: str,
    func_type: str,
    x_range: list[float] | None = None,
    init: dict[str, float] | None = None,
) -> dict[str, Any]

Run a curve fit on a result channel of a finished job.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
result_channel	str	Name of the result channel to fit.	required
func_type	str	Fit model name (e.g. “lorentzian”).	required
x_range	list[float] | None	Optional [min, max] to restrict fit domain.	None
init	dict[str, float] | None	Optional initial parameter overrides.	None

Returns:

Type	Description
dict[str, Any]	Serialised FitResult dict.

Source code in src/icon/server/api/experiment_data_controller.py

async def run_fit(
    self,
    job_id: int,
    result_channel: str,
    func_type: str,
    x_range: list[float] | None = None,
    init: dict[str, float] | None = None,
) -> dict[str, Any]:
    """Run a curve fit on a result channel of a finished job.

    Args:
        job_id: Job identifier.
        result_channel: Name of the result channel to fit.
        func_type: Fit model name (e.g. "lorentzian").
        x_range: Optional [min, max] to restrict fit domain.
        init: Optional initial parameter overrides.

    Returns:
        Serialised FitResult dict.
    """
    data = await asyncio.to_thread(
        ExperimentDataRepository.get_experiment_data_by_job_id,
        job_id=job_id,
    )

    # Find the first non-timestamp scan parameter for x-values
    scan_param_name = next(
        (p for p in data.scan_parameters if p != "timestamp"), None
    )
    if scan_param_name is None:
        return asdict(
            run_curve_fit(
                x=np.array([]),
                y=np.array([]),
                result_channel=result_channel,
                func_type=func_type,  # type: ignore[arg-type]
            )
        )

    scan_values = data.scan_parameters[scan_param_name]
    channel_values = data.readouts.result_channels.get(result_channel, {})

    # Build aligned x, y arrays sorted by index
    indices = sorted(set(scan_values.keys()) & set(channel_values.keys()))
    x = np.array([float(scan_values[i]) for i in indices])
    y = np.array([float(channel_values[i]) for i in indices])

    fit_result = await asyncio.to_thread(
        run_curve_fit,
        x=x,
        y=y,
        result_channel=result_channel,
        func_type=func_type,  # type: ignore[arg-type]
        x_range=x_range,
        init=init,
    )

    if fit_result.success:
        await asyncio.to_thread(
            write_fit_result_by_job_id,
            job_id=job_id,
            fit_result=fit_result,
        )

    result_dict = asdict(fit_result)
    emit_queue.put(
        {
            "event": f"experiment_fit_{job_id}",
            "data": result_dict,
        }
    )
    return result_dict

experiments_controller

ExperimentsController

ExperimentsController()

Bases: DataService

Controller for experiment metadata.

Stores the current set of experiments and exposes them to the API. Updates are compared against the existing metadata and, if changes are detected, an update event is pushed to the Socket.IO emit queue.

Source code in src/icon/server/api/experiments_controller.py

def __init__(self) -> None:
    super().__init__()
    self._experiments: ExperimentDict = {}
    self.hardware_description: str = ""

get_experiments

get_experiments() -> ExperimentDict

Return the current experiment metadata.

Returns:

Type	Description
ExperimentDict	Mapping of experiment IDs to their metadata.

Source code in src/icon/server/api/experiments_controller.py

def get_experiments(self) -> ExperimentDict:
    """Return the current experiment metadata.

    Returns:
        Mapping of experiment IDs to their metadata.
    """
    return self._experiments

get_hardware_description

get_hardware_description() -> str

Return a json string describing the experiment setup.

Source code in src/icon/server/api/experiments_controller.py

def get_hardware_description(self) -> str:
    """Return a json string describing the experiment setup."""
    return self.hardware_description

get_metadata

get_metadata(experiment_id: str) -> ExperimentMetadata

Serve experiment metadata for experiment id experiment_id.

Source code in src/icon/server/api/experiments_controller.py

def get_metadata(self, experiment_id: str) -> ExperimentMetadata:
    """Serve experiment metadata for experiment id `experiment_id`."""
    return self._experiments[experiment_id]

models

device_dict

DeviceDict

Bases: TypedDict

Dictionary representation of a device returned by the API.

created instance-attribute

created: str

Creation timestamp in ISO format.

description instance-attribute

description: str | None

Optional human-readable description.

id instance-attribute

id: int

Database identifier of the device.

name instance-attribute

name: str

Unique device name.

reachable instance-attribute

reachable: bool

Whether the device is currently connected.

scannable_params instance-attribute

scannable_params: list[str]

List of scannable parameter access paths.

status instance-attribute

status: str

Device status, e.g. “enabled” or “disabled”.

url instance-attribute

url: str

pydase server URL of the device.

experiment_dict

ExperimentDict module-attribute

ExperimentDict = dict[str, ExperimentMetadata]

Dictionary mapping the unique experiment identifier to its metadata.

Example

experiment_dict: ExperimentDict = {
    "experiment_library.experiments.my_experiment.MyExperiment (Cool Det)": ExperimentMetadata(
        class_name="MyExperiment",
        constructor_kwargs={
            "name": "Cool Det",
        },
        parameters={
            "Local Parameters": {
                "namespace='experiment_library.experiments.my_experiment.MyExperiment.Cool Det' parameter_group='default' param_type='ParameterTypes.AMPLITUDE'": {
                    "allowed_values": None,
                    "default_value": 0.0,
                    "display_name": "amplitude",
                    "max_value": 100.0,
                    "min_value": 0.0,
                    "unit": "%",
                },
            },
            "ParameterGroup": {
                "namespace='experiment_library.globals.global_parameters' parameter_group='ParameterGroup' param_type='ParameterTypes.AMPLITUDE'": {
                    "allowed_values": None,
                    "default_value": 0.0,
                    "display_name": "amplitude",
                    "max_value": 100.0,
                    "min_value": 0.0,
                    "unit": "%",
                },
            },
        },
    ),
}

ExperimentMetadata dataclass

ExperimentMetadata(
    class_name: str,
    constructor_kwargs: dict[str, Any],
    parameters: dict[str, dict[str, ParameterMetadata]],
    device_parameter_groups: list[Any] = list(),
)

Metadata for a single experiment.

class_name instance-attribute

class_name: str

Name of the experiment class.

constructor_kwargs instance-attribute

constructor_kwargs: dict[str, Any]

Constructor keyword arguments used to instantiate the experiment.

device_parameter_groups class-attribute instance-attribute

device_parameter_groups: list[Any] = field(
    default_factory=list
)

Device parameter groups associated with this experiment.

parameters instance-attribute

parameters: dict[str, dict[str, ParameterMetadata]]

Mapping of display groups to parameter metadata.

parameter_metadata

ParameterMetadata

Bases: TypedDict

Metadata describing a single parameter.

allowed_values instance-attribute

allowed_values: list[Any] | None

Explicit list of allowed values (for ComboboxParameters), otherwise None.

default_value instance-attribute

default_value: float | int

Default value assigned to the parameter.

display_name instance-attribute

display_name: str

Human-readable name of the parameter.

max_value instance-attribute

max_value: float | None

Maximum allowed value for the parameter.

min_value instance-attribute

min_value: float | None

Minimum allowed value for the parameter.

unit instance-attribute

unit: str

Unit in which the parameter value is expressed.

scan_parameter

DatabaseParameter dataclass

DatabaseParameter(
    id: str, values: list[float | int | bool | str]
)

Specification of a database parameter to scan during a job.

id instance-attribute

id: str

Unique identifier of the parameter.

values instance-attribute

values: list[float | int | bool | str]

List of explicit values to scan for this parameter.

DeviceParameter dataclass

DeviceParameter(
    id: str, values: list[float | int], device_name: str
)

Specification of a device parameter to scan during a job.

device_name instance-attribute

device_name: str

Name of the device this parameter belongs to.

id instance-attribute

id: str

Unique identifier of the parameter.

values instance-attribute

values: list[float | int]

List of explicit values to scan for this parameter.

RealtimeParameter dataclass

RealtimeParameter(n_scan_points: int)

Specification of the realtime parameter to scan during a job.

n_scan_points instance-attribute

n_scan_points: int

Number of discrete scan points.

If 0, the scan is continuous.

parameters_controller

ParametersController

ParametersController()

Bases: DataService

Controller for parameter metadata and shared parameter values.

Maintains metadata for all parameters and their display groups, exposes read/write access to parameter value via the API, and ensures parameters are initialized in the InfluxDB backend.

Source code in src/icon/server/api/parameters_controller.py

def __init__(self) -> None:
    super().__init__()
    self._all_parameter_metadata: dict[str, ParameterMetadata] = {}
    self._display_group_metadata: dict[str, dict[str, ParameterMetadata]] = {}

get_all_parameters

get_all_parameters() -> dict[str, DatabaseValueType]

Return the current values of all shared parameters.

Returns:

Type	Description
dict[str, DatabaseValueType]	Mapping of parameter IDs to their values.

Source code in src/icon/server/api/parameters_controller.py

def get_all_parameters(self) -> dict[str, DatabaseValueType]:
    """Return the current values of all shared parameters.

    Returns:
        Mapping of parameter IDs to their values.
    """
    return dict(ParametersRepository.get_shared_parameters())

get_display_groups

get_display_groups() -> dict[
    str, dict[str, ParameterMetadata]
]

Return metadata grouped by display group.

Returns:

Type	Description
dict[str, dict[str, ParameterMetadata]]	Mapping from display group names to parameter metadata.

Source code in src/icon/server/api/parameters_controller.py

def get_display_groups(self) -> dict[str, dict[str, ParameterMetadata]]:
    """Return metadata grouped by display group.

    Returns:
        Mapping from display group names to parameter metadata.
    """
    return self._display_group_metadata

get_parameter_by_id

get_parameter_by_id(parameter_id: str) -> DatabaseValueType

Return the current value of a single parameter.

Parameters:

Name	Type	Description	Default
parameter_id	str	The unique identifier of the parameter.	required

Returns:

Type	Description
DatabaseValueType	The current value stored in the shared parameters dict.

Source code in src/icon/server/api/parameters_controller.py

def get_parameter_by_id(self, parameter_id: str) -> DatabaseValueType:
    """Return the current value of a single parameter.

    Args:
        parameter_id: The unique identifier of the parameter.

    Returns:
        The current value stored in the shared parameters dict.
    """
    return ParametersRepository.get_shared_parameters()[parameter_id]

initialise_parameters_repository

initialise_parameters_repository() -> None

Initialize the global ParametersRepository.

Loads existing parameters from InfluxDB, populates the shared parameters dict in the shared resource manager, and marks the ParametersRepository as initialized.

Source code in src/icon/server/api/parameters_controller.py

def initialise_parameters_repository(self) -> None:
    """Initialize the global `ParametersRepository`.

    Loads existing parameters from InfluxDB, populates the shared parameters dict in
    the shared resource manager, and marks the `ParametersRepository` as
    initialized.
    """
    icon.server.shared_resource_manager.SRM.parameters_dict.update(
        ParametersRepository.get_influxdb_parameters()
    )
    ParametersRepository.initialize(
        shared_parameters=icon.server.shared_resource_manager.SRM.parameters_dict
    )
    logger.info("ParametersRepository successfully initialised.")

update_parameter_by_id

update_parameter_by_id(
    parameter_id: str, value: Any
) -> None

Update a single parameter value in InfluxDB.

Parameters:

Name	Type	Description	Default
parameter_id	str	The unique identifier of the parameter.	required
value	Any	The new value to assign.	required

Source code in src/icon/server/api/parameters_controller.py

def update_parameter_by_id(self, parameter_id: str, value: Any) -> None:
    """Update a single parameter value in InfluxDB.

    Args:
        parameter_id: The unique identifier of the parameter.
        value: The new value to assign.
    """
    ParametersRepository.update_parameters(parameter_mapping={parameter_id: value})

get_added_removed_and_updated_keys

get_added_removed_and_updated_keys(
    new_dict: dict[str, Any], cached_dict: dict[str, Any]
) -> tuple[list[str], list[str], list[str]]

Compare two dictionaries and return added, removed, and updated keys.

Parameters:

Name	Type	Description	Default
new_dict	dict[str, Any]	The latest dictionary state.	required
cached_dict	dict[str, Any]	The previously cached dictionary state.	required

Returns:

Type	Description
tuple[list[str], list[str], list[str]]	A tuple of three lists: added keys removed keys updated keys (present in both but with changed values)

Source code in src/icon/server/api/parameters_controller.py

def get_added_removed_and_updated_keys(
    new_dict: dict[str, Any], cached_dict: dict[str, Any]
) -> tuple[list[str], list[str], list[str]]:
    """Compare two dictionaries and return added, removed, and updated keys.

    Args:
        new_dict: The latest dictionary state.
        cached_dict: The previously cached dictionary state.

    Returns:
        A tuple of three lists:

            - added keys
            - removed keys
            - updated keys (present in both but with changed values)
    """
    keys1 = set(cached_dict)
    keys2 = set(new_dict)

    added_keys = keys2 - keys1
    removed_keys = keys1 - keys2

    intersect_keys = keys1 & keys2
    updated_keys = {key for key in intersect_keys if new_dict[key] != cached_dict[key]}

    return list(added_keys), list(removed_keys), list(updated_keys)

scans_controller

ScansController

ScansController(
    pre_processing_update_queues: list[Queue[UpdateQueue]],
)

Bases: DataService

Controller for triggering update events for jobs across multiple worker processes.

Each worker process has its own update queue ([multiprocessing.Queue][]), which this controller writes to when an update event is triggered.

Source code in src/icon/server/api/scans_controller.py

def __init__(
    self,
    pre_processing_update_queues: list[multiprocessing.Queue[UpdateQueue]],
) -> None:
    super().__init__()
    self._pre_processing_update_queues = pre_processing_update_queues

trigger_update_job_params async

trigger_update_job_params(
    *, job_id: int | None = None
) -> None

Triggers an ‘update_parameters’ event for the given job ID.

Parameters:

Name	Type	Description	Default
job_id	int | None	The ID of the job whose parameters should be updated. If None, all jobs will update their parameters.	None

Source code in src/icon/server/api/scans_controller.py

async def trigger_update_job_params(self, *, job_id: int | None = None) -> None:
    """Triggers an 'update_parameters' event for the given job ID.

    Args:
        job_id: The ID of the job whose parameters should be updated. If None, all
            jobs will update their parameters.
    """
    for pre_processing_update_queue in self._pre_processing_update_queues:
        pre_processing_update_queue.put(
            {
                "event": "update_parameters",
                "job_id": job_id,
            }
        )

scheduler_controller

SchedulerController

SchedulerController(
    devices_controller: DevicesController,
    parameters_controller: ParametersController,
)

Bases: DataService

Controller to submit, inspect, and cancel scheduled jobs.

Provides methods to submit new jobs, cancel pending or running jobs, and query jobs or runs by ID or status. Ensures scan parameters are cast to the correct runtime type before persisting them.

Create a new SchedulerController.

Parameters:

Name	Type	Description	Default
devices_controller	DevicesController	Reference to the devices controller. Used to read current values of device parameters when casting scan values.	required
parameters_controller	ParametersController	Reference to the parameters controller. Used to resolve display names for scan parameters at submission time.	required

Source code in src/icon/server/api/scheduler_controller.py

def __init__(
    self,
    devices_controller: DevicesController,
    parameters_controller: ParametersController,
) -> None:
    """Create a new SchedulerController.

    Args:
        devices_controller: Reference to the devices controller. Used to read
            current values of device parameters when casting scan values.
        parameters_controller: Reference to the parameters controller. Used to
            resolve display names for scan parameters at submission time.
    """
    super().__init__()
    self._devices_controller = devices_controller
    self._parameters_controller = parameters_controller

cancel_job

cancel_job(*, job_id: int) -> None

Cancel a queued or running job.

The following status updates are performed:

• Job: PROCESSING/SUBMITTED → PROCESSED
• JobRun: PENDING/PROCESSING → CANCELLED

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to cancel.	required

Source code in src/icon/server/api/scheduler_controller.py

def cancel_job(self, *, job_id: int) -> None:
    """Cancel a queued or running job.

    The following status updates are performed:

    - Job: PROCESSING/SUBMITTED → PROCESSED
    - JobRun: PENDING/PROCESSING → CANCELLED

    Args:
        job_id: ID of the job to cancel.
    """
    job = JobRepository.get_job_by_id(job_id=job_id)
    if job.status in (JobStatus.PROCESSING, JobStatus.SUBMITTED):
        JobRepository.update_job_status(job=job, status=JobStatus.PROCESSED)
        job_run = JobRunRepository.get_run_by_job_id(job_id=job_id)
        if job_run.status in (
            JobRunStatus.PENDING,
            JobRunStatus.PROCESSING,
            JobRunStatus.PAUSED,
        ):
            JobRunRepository.update_run_by_id(
                run_id=job_run.id,
                status=JobRunStatus.CANCELLED,
                log="Cancelled through user interaction.",
            )

get_job_by_id

get_job_by_id(*, job_id: int) -> Job

Fetch a job with its experiment source and scan parameters.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required

Returns:

Type	Description
Job	The job record.

Source code in src/icon/server/api/scheduler_controller.py

def get_job_by_id(self, *, job_id: int) -> Job:
    """Fetch a job with its experiment source and scan parameters.

    Args:
        job_id: Job identifier.

    Returns:
        The job record.
    """
    return JobRepository.get_job_by_id(
        job_id=job_id, load_experiment_source=True, load_scan_parameters=True
    )

get_job_run_by_id

get_job_run_by_id(*, job_id: int) -> JobRun

Fetch the run record for a given job.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required

Returns:

Type	Description
JobRun	The associated run record.

Source code in src/icon/server/api/scheduler_controller.py

def get_job_run_by_id(self, *, job_id: int) -> JobRun:
    """Fetch the run record for a given job.

    Args:
        job_id: Job identifier.

    Returns:
        The associated run record.
    """
    return JobRunRepository.get_run_by_job_id(job_id=job_id)

get_scheduled_jobs

get_scheduled_jobs(
    *,
    status: JobStatus | None = None,
    start: str | None = None,
    stop: str | None = None,
) -> dict[int, Job]

List jobs filtered by status and optional ISO timeframe.

Parameters:

Name	Type	Description	Default
status	JobStatus | None	Optional job status filter.	None
start	str | None	Optional ISO8601 start timestamp (inclusive).	None
stop	str | None	Optional ISO8601 stop timestamp (exclusive).	None

Returns:

Type	Description
dict[int, Job]	Mapping from job ID to job record.

Source code in src/icon/server/api/scheduler_controller.py

def get_scheduled_jobs(
    self,
    *,
    status: JobStatus | None = None,
    start: str | None = None,
    stop: str | None = None,
) -> dict[int, Job]:
    """List jobs filtered by status and optional ISO timeframe.

    Args:
        status: Optional job status filter.
        start: Optional ISO8601 start timestamp (inclusive).
        stop: Optional ISO8601 stop timestamp (exclusive).

    Returns:
        Mapping from job ID to job record.
    """
    start_date = datetime.fromisoformat(start) if start is not None else None
    stop_date = datetime.fromisoformat(stop) if stop is not None else None

    return {
        job.id: job
        for job in JobRepository.get_jobs_by_status_and_timeframe(
            status=status, start=start_date, stop=stop_date
        )
    }

pause_job

pause_job(*, job_id: int) -> None

Pause a running job.

The pre-processing worker holding the job will finish any in-flight work and then block in a polling loop, keeping the remaining scan state in memory. Tasks already queued for the hardware worker are diverted back to the pre-processing worker via the existing outdated_tasks rewind mechanism.

No-op if the job run is not in PROCESSING state.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to pause.	required

Source code in src/icon/server/api/scheduler_controller.py

def pause_job(self, *, job_id: int) -> None:
    """Pause a running job.

    The pre-processing worker holding the job will finish any in-flight work and
    then block in a polling loop, keeping the remaining scan state in memory.
    Tasks already queued for the hardware worker are diverted back to the
    pre-processing worker via the existing ``outdated_tasks`` rewind mechanism.

    No-op if the job run is not in ``PROCESSING`` state.

    Args:
        job_id: ID of the job to pause.
    """
    job_run = JobRunRepository.get_run_by_job_id(job_id=job_id)
    if job_run.status == JobRunStatus.PROCESSING:
        JobRunRepository.update_run_by_id(
            run_id=job_run.id,
            status=JobRunStatus.PAUSED,
            log="Paused through user interaction.",
        )

resume_job

resume_job(*, job_id: int) -> None

Resume a paused job.

The pre-processing worker observes the status change, regenerates any tasks the hardware worker diverted while paused (picking up fresh parameter values in the process), and continues producing data points from where it left off.

No-op if the job run is not in PAUSED state.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to resume.	required

Source code in src/icon/server/api/scheduler_controller.py

def resume_job(self, *, job_id: int) -> None:
    """Resume a paused job.

    The pre-processing worker observes the status change, regenerates any tasks
    the hardware worker diverted while paused (picking up fresh parameter values
    in the process), and continues producing data points from where it left off.

    No-op if the job run is not in ``PAUSED`` state.

    Args:
        job_id: ID of the job to resume.
    """
    job_run = JobRunRepository.get_run_by_job_id(job_id=job_id)
    if job_run.status == JobRunStatus.PAUSED:
        JobRunRepository.update_run_by_id(
            run_id=job_run.id,
            status=JobRunStatus.PROCESSING,
        )

submit_job async

submit_job(
    *,
    experiment_id: str,
    scan_parameters: list[dict[str, Any]],
    priority: int = 20,
    local_parameters_timestamp: datetime | None = None,
    repetitions: int = 1,
    number_of_shots: int = 50,
    git_commit_hash: str | None = None,
    auto_calibration: bool = False,
) -> int

Create and submit a job with typed scan parameters.

Each scan parameter’s values are cast to the current type of the target parameter (device parameter via DevicesController or shared parameter via ParametersRepository).

Parameters:

Name	Type	Description	Default
experiment_id	str	Experiment identifier (from experiment library).	required
scan_parameters	list[dict[str, Any]]	List of scan parameter specs (id, values, optional device_name).	required
priority	int	Higher values run sooner.	20
local_parameters_timestamp	datetime | None	ISO timestamp to snapshot local parameters; defaults to datetime.now(tz=timezone).	None
repetitions	int	Number of experiment repetitions.	1
number_of_shots	int	Shots per data point.	50
git_commit_hash	str | None	Git commit to associate with the job; if None, job is marked debug_mode=True.	None
auto_calibration	bool	Whether to run auto-calibration for the job.	False

Returns:

Type	Description
int	The persisted job ID.

Source code in src/icon/server/api/scheduler_controller.py

async def submit_job(
    self,
    *,
    experiment_id: str,
    scan_parameters: list[dict[str, Any]],
    priority: int = 20,
    local_parameters_timestamp: datetime | None = None,
    repetitions: int = 1,
    number_of_shots: int = 50,
    git_commit_hash: str | None = None,
    auto_calibration: bool = False,
) -> int:
    """Create and submit a job with typed scan parameters.

    Each scan parameter's values are cast to the current type of the target
    parameter (device parameter via `DevicesController` or shared parameter via
    `ParametersRepository`).

    Args:
        experiment_id: Experiment identifier (from experiment library).
        scan_parameters: List of scan parameter specs (id, values, optional
            device_name).
        priority: Higher values run sooner.
        local_parameters_timestamp: ISO timestamp to snapshot local parameters;
            defaults to `datetime.now(tz=timezone)`.
        repetitions: Number of experiment repetitions.
        number_of_shots: Shots per data point.
        git_commit_hash: Git commit to associate with the job; if `None`, job is
            marked `debug_mode=True`.
        auto_calibration: Whether to run auto-calibration for the job.

    Returns:
        The persisted job ID.
    """
    if local_parameters_timestamp is None:
        local_parameters_timestamp = now()

    experiment_source = ExperimentSource(experiment_id=experiment_id)

    experiment_source = ExperimentSourceRepository.get_or_create_experiment(
        experiment_source=experiment_source
    )

    def to_sqlite_model(
        param: ScanParameter,
    ) -> sqlite_scan_parameter.ScanParameter:
        if isinstance(param, RealtimeParameter):
            return sqlite_scan_parameter.ScanParameter(
                name="Real Time",
                variable_id="Real Time",
                scan_values=[1] * param.n_scan_points,
                realtime=True,
            )
        if isinstance(param, DatabaseParameter):
            return sqlite_scan_parameter.ScanParameter(
                name=self._resolve_display_name(param.id),
                variable_id=param.id,
                scan_values=param.values,
            )
        return sqlite_scan_parameter.ScanParameter(
            name=self._resolve_display_name(param.id),
            variable_id=param.id,
            scan_values=param.values,
            device_id=DeviceRepository.get_device_by_name(
                name=param.device_name
            ).id,
        )

    concretized_params = [
        scan_parameter_from_dict(
            {**param, "values": await self._cast_scan_values_to_param_type(**param)}
        )
        for param in scan_parameters
    ]
    realtime_params = [
        param
        for param in concretized_params
        if isinstance(param, RealtimeParameter)
    ]
    if len(realtime_params) > 1:
        raise ValueError("Only 0 or 1 realtime parameter is allowed")
    if (
        realtime_params
        and realtime_params[0].n_scan_points == 0
        and repetitions > 1
    ):
        raise ValueError(
            "Only 1 repetition possible if continuous realtime is present"
        )
    sqlite_scan_parameters = [
        to_sqlite_model(param) for param in concretized_params
    ]

    job = Job(
        experiment_source=experiment_source,
        priority=priority,
        local_parameters_timestamp=local_parameters_timestamp,
        scan_parameters=sqlite_scan_parameters,
        repetitions=repetitions,
        git_commit_hash=git_commit_hash,
        number_of_shots=number_of_shots,
        auto_calibration=auto_calibration,
        debug_mode=git_commit_hash is None,
    )
    job = JobRepository.submit_job(job=job)

    return job.id

status_controller

StatusController

StatusController(devices: Devices)

Bases: DataService

Controller for system status monitoring.

Periodically checks availability of InfluxDB and hardware and emits status events via the Socket.IO queue.

Source code in src/icon/server/api/status_controller.py

def __init__(self, devices: Devices) -> None:
    super().__init__()
    self.__devices = devices
    self._influxdb_available = False
    self._hardware_available: HardwareStatus = {}

check_hardware_status async

check_hardware_status() -> None

Check hardware connection and reconnect if necessary.

Ensures the hardware controller matches the configured host/port and reconnects in a background thread if required.

Emits a "status.hardware" event to the Socket.IO queue.

Source code in src/icon/server/api/status_controller.py

async def check_hardware_status(self) -> None:
    """Check hardware connection and reconnect if necessary.

    Ensures the hardware controller matches the configured host/port and reconnects
    in a background thread if required.

    Emits a `"status.hardware"` event to the Socket.IO queue.
    """
    await asyncio.to_thread(self.__devices.reload)

    status: HardwareStatus = {
        dev_id: (
            dev.controller.connected
            if isinstance(dev.controller, HardwareController)
            else {"msg": format(dev.controller)}
        )
        for dev_id, dev in self.__devices.items()
    }
    self._hardware_available = status
    emit_queue.put({"event": "status.hardware", "data": status})

check_influxdb_status

check_influxdb_status() -> None

Check if InfluxDB is responsive and update status.

Emits a "status.influxdb" event to the Socket.IO queue.

Source code in src/icon/server/api/status_controller.py

def check_influxdb_status(self) -> None:
    """Check if InfluxDB is responsive and update status.

    Emits a `"status.influxdb"` event to the Socket.IO queue.
    """
    status = influxdb_v1.is_responsive()

    self._influxdb_available = status
    emit_queue.put({"event": "status.influxdb", "data": status})

get_status

get_status() -> Status

Return the current system status flags.

Returns:

Type	Description
Status	A dictionary with: "influxdb": Whether InfluxDB is responsive. "hardware": Whether the hardware connection is active.

Source code in src/icon/server/api/status_controller.py

def get_status(self) -> Status:
    """Return the current system status flags.

    Returns:
        A dictionary with:

            - `"influxdb"`: Whether InfluxDB is responsive.
            - `"hardware"`: Whether the hardware connection is active.
    """
    return {
        "influxdb": self._influxdb_available,
        "hardware": self._hardware_available,
    }

icon.server.data_access.models.enums

This module defines enums used by the SQLAlchemy models.

These enums represent database-level states for jobs, job runs, and devices. They are stored as strings in the database and used throughout ICON’s scheduling and device management logic.

DeviceStatus

Bases: Enum

Operational states of a device.

DISABLED class-attribute instance-attribute

DISABLED = 'disabled'

Device is disabled and should not be used.

ENABLED class-attribute instance-attribute

ENABLED = 'enabled'

Device is enabled and may be connected.

JobRunStatus

Bases: Enum

Lifecycle states of a job run.

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

Run was cancelled before completion.

DONE class-attribute instance-attribute

DONE = 'done'

Run completed successfully.

FAILED class-attribute instance-attribute

FAILED = 'failed'

Run ended unsuccessfully due to an error.

PAUSED class-attribute instance-attribute

PAUSED = 'paused'

Run has been paused by the user; its pre-processing worker is holding its remaining state in memory until the run is resumed or cancelled.

PENDING class-attribute instance-attribute

PENDING = 'pending'

Run is queued but has not started yet.

PROCESSING class-attribute instance-attribute

PROCESSING = 'processing'

Run is currently executing.

JobStatus

Bases: Enum

Lifecycle states of a job submission.

PROCESSED class-attribute instance-attribute

PROCESSED = 'processed'

Job has finished or was cancelled and is no longer active.

PROCESSING class-attribute instance-attribute

PROCESSING = 'processing'

Job has been put into the pre-processing task queue.

SUBMITTED class-attribute instance-attribute

SUBMITTED = 'submitted'

Job has been created and is waiting to be scheduled.

icon.server.data_access.models.sqlite

This module contains the SQLAlchemy models for ICON.

All models must be imported and added to the __all__ list here so that Alembic can correctly detect them during schema autogeneration. Alembic inspects Base.metadata, which is only populated with models that are actually imported at runtime.

Base

Bases: DeclarativeBase

Base class for all SQLAlchemy ORM models in ICON.

This class configures the declarative mapping and provides a datetime type mapping for all models that inherit from it.

type_annotation_map class-attribute

type_annotation_map: dict[type, Any] = {
    datetime: TIMESTAMP(timezone=True)
}

Custom type mapping used when interpreting Python type annotations.

Currently, datetime.datetime is mapped to sqlalchemy.TIMESTAMP(timezone=True) to ensure timezone-aware timestamps across all models.

Device

Bases: Base

SQLAlchemy model for a registered device.

Represents an external device accessible via a pydase service. Stores configuration, connection details, and retry behaviour. A device may be linked to multiple scan parameters.

created class-attribute instance-attribute

created: Mapped[datetime] = mapped_column(default=now)

Timestamp when the device entry was created.

description class-attribute instance-attribute

description: Mapped[str | None] = mapped_column(
    default=None
)

Optional human-readable description of the device.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the device.

name class-attribute instance-attribute

name: Mapped[str] = mapped_column(unique=True, index=True)

Unique name of the device.

retry_attempts class-attribute instance-attribute

retry_attempts: Mapped[int] = mapped_column(
    default=3, nullable=False
)

Number of attempts to verify the device value was set correctly.

retry_delay_seconds class-attribute instance-attribute

retry_delay_seconds: Mapped[float] = mapped_column(
    default=0.0, nullable=False
)

Delay in seconds between retry attempts

scan_parameters class-attribute instance-attribute

scan_parameters: Mapped[list[ScanParameter]] = relationship(
    "ScanParameter", back_populates="device"
)

Relationship to scan parameters linked to this device.

status class-attribute instance-attribute

status: Mapped[DeviceStatus] = mapped_column(
    default=ENABLED, index=True
)

Current status of the device (enabled or disabled).

url class-attribute instance-attribute

url: Mapped[str] = mapped_column()

pydase service URL of the device.

ExperimentSource

Bases: Base

SQLAlchemy model for experiment sources.

Represents a unique experiment identifier from the experiment library. Each experiment source may be linked to multiple jobs.

experiment_id class-attribute instance-attribute

experiment_id: Mapped[str] = mapped_column()

Unique experiment identifier string (as defined in the experiment library).

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the experiment source.

jobs class-attribute instance-attribute

jobs: Mapped[list[Job]] = relationship(
    back_populates="experiment_source"
)

Relationship to jobs associated with this experiment source.

Job

Bases: Base

SQLAlchemy model for experiment jobs.

Represents a scheduled or running experiment job, including its metadata, status, and relationships to experiment sources, runs, and scan parameters.

Constraints

• priority must be between 0 and 20.
• Indexed by (experiment_source_id, status, priority, created).

auto_calibration class-attribute instance-attribute

auto_calibration: Mapped[bool] = mapped_column(
    default=False
)

Whether auto-calibration is enabled for this job. Currently unused.

created class-attribute instance-attribute

created: Mapped[datetime] = mapped_column(default=now)

Timestamp when the job was created. This cannot be set manually.

debug_mode class-attribute instance-attribute

debug_mode: Mapped[bool] = mapped_column(default=False)

Whether the job was submitted in debug mode (no commit hash).

experiment_source class-attribute instance-attribute

experiment_source: Mapped[ExperimentSource] = relationship(
    back_populates="jobs"
)

Relationship to the experiment source.

experiment_source_id class-attribute instance-attribute

experiment_source_id: Mapped[int] = mapped_column(
    ForeignKey("experiment_sources.id")
)

Foreign key referencing the associated experiment source.

git_commit_hash class-attribute instance-attribute

git_commit_hash: Mapped[str | None] = mapped_column(
    default=None
)

Git commit hash of the experiment code associated with the job.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the job.

local_parameters_timestamp class-attribute instance-attribute

local_parameters_timestamp: Mapped[datetime] = (
    mapped_column(default=now)
)

Timestamp of the local parameter snapshot used for this job.

number_of_shots class-attribute instance-attribute

number_of_shots: Mapped[int] = mapped_column(default=50)

Number of shots per repetition.

parent_job class-attribute instance-attribute

parent_job: Mapped[Job | None] = relationship(
    "Job",
    remote_side=[id],
    back_populates="resubmitted_jobs",
)

Relationship to the parent job from which this job was resubmitted.

parent_job_id class-attribute instance-attribute

parent_job_id: Mapped[int | None] = mapped_column(
    ForeignKey("job_submissions.id"), nullable=True
)

Foreign key referencing the original job if this job was resubmitted.

priority class-attribute instance-attribute

priority: Mapped[int] = mapped_column(default=20)

Job priority, between 0 (lowest) and 20 (highest).

repetitions class-attribute instance-attribute

repetitions: Mapped[int] = mapped_column(default=1)

Number of times the experiment should be repeated.

resubmitted_jobs class-attribute instance-attribute

resubmitted_jobs: Mapped[list[Job]] = relationship(
    "Job", back_populates="parent_job"
)

List of jobs resubmitted from this job.

run class-attribute instance-attribute

run: Mapped[JobRun] = relationship(back_populates='job')

Relationship to the job run associated with this job.

scan_parameters class-attribute instance-attribute

scan_parameters: Mapped[list[ScanParameter]] = relationship(
    back_populates="job"
)

List of scan parameters associated with this job.

status class-attribute instance-attribute

status: Mapped[JobStatus] = mapped_column(default=SUBMITTED)

Current status of the job (submitted, processing, etc.).

JobRun

Bases: Base

SQLAlchemy model for job runs.

Represents the execution of a job, including its scheduled time, current status, and log messages.

Constraints

• Indexed by (job_id, status, scheduled_time).
• scheduled_time must be unique across runs.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the job run.

job class-attribute instance-attribute

job: Mapped[Job] = relationship(back_populates='run')

Relationship to the job associated with this run.

job_id class-attribute instance-attribute

job_id: Mapped[int] = mapped_column(
    ForeignKey("job_submissions.id")
)

Foreign key referencing the job being executed.

log class-attribute instance-attribute

log: Mapped[str | None] = mapped_column(default=None)

Optional log message for this run (e.g., cancellation reason).

parameter_update_timestamp class-attribute instance-attribute

parameter_update_timestamp: Mapped[datetime | None] = (
    mapped_column(default=None)
)

Timestamp of the last parameter update.

scheduled_time class-attribute instance-attribute

scheduled_time: Mapped[datetime] = mapped_column(
    default=now
)

Time when the run was scheduled to start.

status class-attribute instance-attribute

status: Mapped[JobRunStatus] = mapped_column(
    default=PENDING
)

Current status of the run (pending, processing, cancelled, etc.).

ScanParameter

Bases: Base

SQLAlchemy model for scan parameters.

Represents a parameter scanned during a job execution. Each parameter is linked to a job and optionally to a device.

device class-attribute instance-attribute

device: Mapped[Device | None] = relationship(
    back_populates="scan_parameters", lazy="joined"
)

Relationship to the device associated with this parameter.

device_id class-attribute instance-attribute

device_id: Mapped[int | None] = mapped_column(
    ForeignKey("devices.id"), nullable=True
)

Foreign key referencing the associated device, if any.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the scan parameter.

job class-attribute instance-attribute

job: Mapped[Job] = relationship(
    back_populates="scan_parameters"
)

Relationship to the job.

job_id class-attribute instance-attribute

job_id: Mapped[int] = mapped_column(
    ForeignKey("job_submissions.id")
)

Foreign key referencing the job this parameter belongs to.

name class-attribute instance-attribute

name: Mapped[str] = mapped_column(nullable=False)

Human-friendly display name persisted at submission time.

scan_values class-attribute instance-attribute

scan_values: Mapped[list[DatabaseValueType]] = (
    mapped_column(JSONEncodedList, nullable=False)
)

List of values scanned for this parameter (stored as JSON).

variable_id class-attribute instance-attribute

variable_id: Mapped[str] = mapped_column()

Identifier of the parameter being scanned.

unique_id

unique_id() -> str

Return a unique identifier for the parameter.

Returns:

Type	Description
str	"Device(<device_name>) <variable_id>" if a device is associated, otherwise just <variable_id>.

Source code in src/icon/server/data_access/models/sqlite/scan_parameter.py

def unique_id(self) -> str:
    """Return a unique identifier for the parameter.

    Returns:
        `"Device(<device_name>) <variable_id>"` if a device is associated, otherwise
            just `<variable_id>`.
    """
    return (
        f"Device({self.device.name}) {self.variable_id}"
        if self.device is not None
        else self.variable_id
    )

icon.server.data_access.repositories

This module contains the repository layer for ICON’s data access.

Repositories encapsulate database access logic and hide the underlying persistence technology (SQLAlchemy sessions, InfluxDB queries, etc.) from the rest of the application. They expose simple, intention-revealing methods for creating, retrieving, and updating domain objects, while emitting Socket.IO events when relevant.

By using repositories, controllers and services can work with high-level operations (e.g. “submit a job”, “update a device”) without needing to know how the data is stored or which database backend is used. This keeps the codebase modular, easier to maintain, and allows the persistence layer to evolve independently of business logic.

device_repository

DeviceRepository

Repository for Device entities.

Provides methods to create, update, and query devices in the SQLite database. All methods open their own SQLAlchemy session and return detached ORM objects.

add_device staticmethod

add_device(*, device: Device) -> Device

Insert a new device into the database.

Parameters:

Name	Type	Description	Default
device	Device	Device instance to persist.	required

Returns:

Type	Description
Device	The persisted device with database-generated fields (e.g., id) populated.

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def add_device(*, device: Device) -> Device:
    """Insert a new device into the database.

    Args:
        device: Device instance to persist.

    Returns:
        The persisted device with database-generated fields (e.g., `id`) populated.
    """
    with sqlalchemy.orm.session.Session(engine) as session:
        session.add(device)
        session.commit()
        session.refresh(device)
        logger.debug("Added new device %s", device)

    emit_queue.put(
        {
            "event": "device.new",
            "data": {
                "device": SQLAlchemyDictEncoder.encode(obj=device),
            },
        }
    )

    return device

get_all_device_names staticmethod

get_all_device_names() -> Sequence[str]

Return the names of all devices.

Returns:

Type	Description
Sequence[str]	List of device names.

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def get_all_device_names() -> Sequence[str]:
    """Return the names of all devices.

    Returns:
        List of device names.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = sqlalchemy.select(Device.name)
        return session.execute(stmt).scalars().all()

get_device_by_id staticmethod

get_device_by_id(*, id: int) -> Device

Return a device by database ID.

Parameters:

Name	Type	Description	Default
id	int	Primary key identifier of the device.	required

Returns:

Type	Description
Device	The matching device.

Raises:

Type	Description
NoResultFound	If no device exists with the given ID.

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def get_device_by_id(*, id: int) -> Device:
    """Return a device by database ID.

    Args:
        id: Primary key identifier of the device.

    Returns:
        The matching device.

    Raises:
        NoResultFound: If no device exists with the given ID.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = sqlalchemy.select(Device).where(Device.id == id)
        return session.execute(stmt).scalar_one()

get_device_by_name staticmethod

get_device_by_name(*, name: str) -> Device

Return a device by unique name.

Parameters:

Name	Type	Description	Default
name	str	Device name.	required

Returns:

Type	Description
Device	The matching device.

Raises:

Type	Description
NoDeviceFoundError	If no device exists with the given name.

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def get_device_by_name(*, name: str) -> Device:
    """Return a device by unique name.

    Args:
        name: Device name.

    Returns:
        The matching device.

    Raises:
        NoDeviceFoundError: If no device exists with the given name.
    """
    try:
        with sqlalchemy.orm.Session(engine) as session:
            stmt = sqlalchemy.select(Device).where(Device.name == name)
            return session.execute(stmt).scalar_one()
    except sqlalchemy.exc.NoResultFound:
        raise NoDeviceFoundError(
            f"Device with name {name!r} does not exist.",
        ) from None

get_devices_by_status staticmethod

get_devices_by_status(
    *, status: DeviceStatus | None = None
) -> Sequence[Device]

Return devices filtered by status.

Parameters:

Name	Type	Description	Default
status	DeviceStatus | None	Optional device status to filter on.	None

Returns:

Type	Description
Sequence[Device]	All devices matching the filter (or all devices if no filter is given).

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def get_devices_by_status(
    *,
    status: DeviceStatus | None = None,
) -> Sequence[Device]:
    """Return devices filtered by status.

    Args:
        status: Optional device status to filter on.

    Returns:
        All devices matching the filter (or all devices if no filter is given).
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = sqlalchemy.select(Device)

        if status is not None:
            stmt = stmt.where(Device.status == status)

        return session.execute(stmt).scalars().all()

update_device staticmethod

update_device(
    *,
    name: str,
    url: str | None = None,
    status: DeviceStatus | None = None,
    retry_attempts: int | None = None,
    retry_delay_seconds: float | None = None,
) -> Device

Update an existing device by name.

Parameters:

Name	Type	Description	Default
name	str	Unique device name.	required
url	str | None	New device URL (cannot change if the device is enabled).	None
status	DeviceStatus | None	New device status (enabled/disabled).	None
retry_attempts	int | None	Updated retry attempt count.	None
retry_delay_seconds	float | None	Updated retry delay in seconds.	None

Returns:

Type	Description
Device	The updated device.

Raises:

Type	Description
RuntimeError	If attempting to change the URL of an enabled device.
NoDeviceFoundError	If no device with the given name exists.

Source code in src/icon/server/data_access/repositories/device_repository.py

@staticmethod
def update_device(
    *,
    name: str,
    url: str | None = None,
    status: DeviceStatus | None = None,
    retry_attempts: int | None = None,
    retry_delay_seconds: float | None = None,
) -> Device:
    """Update an existing device by name.

    Args:
        name: Unique device name.
        url: New device URL (cannot change if the device is enabled).
        status: New device status (enabled/disabled).
        retry_attempts: Updated retry attempt count.
        retry_delay_seconds: Updated retry delay in seconds.

    Returns:
        The updated device.

    Raises:
        RuntimeError: If attempting to change the URL of an enabled device.
        NoDeviceFoundError: If no device with the given name exists.
    """
    updated_properties = {
        name: new_value
        for name, new_value in {
            "url": url,
            "status": status if status is not None else None,
            "retry_attempts": retry_attempts,
            "retry_delay_seconds": retry_delay_seconds,
        }.items()
        if new_value is not None
    }

    if "url" in updated_properties:
        device = DeviceRepository.get_device_by_name(name=name)
        if device.status == DeviceStatus.ENABLED:
            raise RuntimeError("Cannot change url of an enabled device")

    with sqlalchemy.orm.Session(engine) as session:
        session.execute(
            update(Device).where(Device.name == name).values(updated_properties)
        )
        session.commit()

        device = session.execute(
            select(Device).where(Device.name == name)
        ).scalar_one()
        session.expunge(device)

        logger.debug("Updated device %s", device)

    serialized_properties = {
        key: value.value if isinstance(value, enum.Enum) else value
        for key, value in updated_properties.items()
    }

    if "status" in updated_properties:
        serialized_properties["reachable"] = False

    emit_queue.put(
        {
            "event": "device.update",
            "data": {
                "device_name": device.name,
                "updated_properties": serialized_properties,
            },
        }
    )

    return device

NoDeviceFoundError

Bases: Exception

Raised when a device could not be found by the given identifier.

experiment_data_repository

ExperimentDataRepository

Repository for HDF5-based experiment data.

Manages HDF5 file creation and updates (metadata, results, parameters), with hdf5-level locking to support concurrent writers.

get_experiment_data_by_job_id staticmethod

get_experiment_data_by_job_id(
    *,
    job_id: int,
    max_transfer_bytes: int = 50000000,
    include_hardware_instructions: bool = False,
) -> ExperimentData

Load stored data for a job from its HDF5 file.

When loading all data would exceed max_transfer_bytes, only the last N data points that fit within the budget are returned. The budget is estimated from HDF5 metadata (channel count, shots per channel) without reading actual data.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
max_transfer_bytes	int	Approximate cap on the serialised payload size in bytes. Defaults to 50 MB.	50000000
include_hardware_instructions	bool	If True, load hardware_instructions entries into hardware_instructions. Defaults to False — those blobs are large (~27 KB each, one per changed point) and are omitted from the default RPC response.	False

Returns:

Type	Description
ExperimentData	Experiment data payload suitable for the API.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

@staticmethod
def get_experiment_data_by_job_id(
    *,
    job_id: int,
    max_transfer_bytes: int = 50_000_000,
    include_hardware_instructions: bool = False,
) -> ExperimentData:
    """Load stored data for a job from its HDF5 file.

    When loading all data would exceed *max_transfer_bytes*, only the
    last N data points that fit within the budget are returned.  The
    budget is estimated from HDF5 metadata (channel count, shots per
    channel) without reading actual data.

    Args:
        job_id: Job identifier.
        max_transfer_bytes: Approximate cap on the serialised payload
            size in bytes.  Defaults to 50 MB.
        include_hardware_instructions: If True, load ``hardware_instructions`` entries
            into ``hardware_instructions``.  Defaults to False — those blobs are
            large (~27 KB each, one per changed point) and are omitted
            from the default RPC response.

    Returns:
        Experiment data payload suitable for the API.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename

    if not Path(h5_path).exists():
        logger.warning("The file %s does not exist.", h5_path)
        return ExperimentData()

    with h5_open(h5_path, "r") as h5file:
        return load_experiment_data(
            h5file,
            max_transfer_bytes,
            include_hardware_instructions=include_hardware_instructions,
        )

update_metadata_by_job_id staticmethod

update_metadata_by_job_id(
    *,
    job_id: int,
    number_of_shots: int,
    repetitions: int,
    readout_metadata: ReadoutMetadata,
    local_parameter_timestamp: datetime | None = None,
    parameters: list[ScanParameter] | None = None,
) -> None

Create or update HDF5 metadata for a job.

Initializes datasets, sets file-level attributes, and stores plot window metadata for result/shot/vector channels.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
number_of_shots	int	Shots per data point.	required
repetitions	int	Number of repetitions.	required
readout_metadata	ReadoutMetadata	Plot/window/channel metadata.	required
local_parameter_timestamp	datetime | None	Optional timestamp for local parameters.	None
parameters	list[ScanParameter] | None	Scan parameters.	None

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

@staticmethod
def update_metadata_by_job_id(
    *,
    job_id: int,
    number_of_shots: int,
    repetitions: int,
    readout_metadata: ReadoutMetadata,
    local_parameter_timestamp: datetime | None = None,
    parameters: list[ScanParameter] | None = None,
) -> None:
    """Create or update HDF5 metadata for a job.

    Initializes datasets, sets file-level attributes, and stores plot window
    metadata for result/shot/vector channels.

    Args:
        job_id: Job identifier.
        number_of_shots: Shots per data point.
        repetitions: Number of repetitions.
        readout_metadata: Plot/window/channel metadata.
        local_parameter_timestamp: Optional timestamp for local parameters.
        parameters: Scan parameters.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    job = JobRepository.get_job_by_id(job_id=job_id, load_experiment_source=True)

    with h5_open(h5_path, "a") as h5file:
        prepare_readout_metadata(
            h5file,
            job_id=job_id,
            experiment_id=job.experiment_source.experiment_id,
            number_of_shots=number_of_shots,
            repetitions=repetitions,
            readout_metadata=readout_metadata,
            local_parameter_timestamp=local_parameter_timestamp,
            parameters=parameters or [],
        )

    metadata_key_remap = {
        "readout_channel_windows": "result_channels",
        "shot_channel_windows": "shot_channels",
        "vector_channel_windows": "vector_channels",
    }
    emit_queue.put(
        {
            "event": f"experiment_{job_id}_metadata",
            "data": {
                "readout_metadata": {
                    metadata_key_remap[key]: val
                    for key, val in asdict(readout_metadata).items()
                    if key in metadata_key_remap
                }
            },
        }
    )

write_experiment_data_by_job_id staticmethod

write_experiment_data_by_job_id(
    *, job_id: int, data_point: ExperimentDataPoint
) -> None

Append a complete data point to the HDF5 file and emit an event.

Writes scan parameters, result/shot/vector channels, and hardware instructions.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
data_point	ExperimentDataPoint	Data point payload to append.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

@staticmethod
def write_experiment_data_by_job_id(
    *,
    job_id: int,
    data_point: ExperimentDataPoint,
) -> None:
    """Append a complete data point to the HDF5 file and emit an event.

    Writes scan parameters, result/shot/vector channels, and hardware instructions.

    Args:
        job_id: Job identifier.
        data_point: Data point payload to append.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename

    with h5_open(h5_path, "a") as h5file:
        write_experiment_data_point(h5file, data_point)
    logger.debug("Appended data to %s", h5_path)

    emit_queue.put(
        {
            "event": f"experiment_{job_id}",
            "data": asdict(data_point),
        }
    )

write_parameter_update_by_job_id staticmethod

write_parameter_update_by_job_id(
    *,
    job_id: int,
    timestamp: str,
    parameter_values: dict[str, str | int | float | bool],
) -> None

Append parameter updates under the ‘parameters’ group.

Creates a dataset per parameter storing (timestamp, value) entries. Appends only when the value changed from the last entry.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
timestamp	str	ISO timestamp string.	required
parameter_values	dict[str, str | int | float | bool]	Mapping of parameter id to value.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

@staticmethod
def write_parameter_update_by_job_id(
    *,
    job_id: int,
    timestamp: str,
    parameter_values: dict[str, str | int | float | bool],
) -> None:
    """Append parameter updates under the 'parameters' group.

    Creates a dataset per parameter storing (timestamp, value) entries.
    Appends only when the value changed from the last entry.

    Args:
        job_id: Job identifier.
        timestamp: ISO timestamp string.
        parameter_values: Mapping of parameter id to value.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    parameter_updates = {}
    with h5_open(h5_path, "a") as h5file:
        parameters_group = h5file.require_group("parameters")

        for param_id, value in parameter_values.items():
            dtype = [("timestamp", "S26"), ("value", get_hdf5_dtype(value))]

            if param_id in parameters_group:
                ds: h5py.Dataset = parameters_group[param_id]
                if ds.shape[0] > 0:
                    last_entry = ds[-1]
                    last_value = last_entry["value"]
                    if isinstance(value, str):
                        if last_value.decode() == value:
                            continue
                    elif last_value == value:
                        continue

                index = ds.shape[0]
                resize_dataset(ds, next_index=index, axis=0)
            else:
                ds = parameters_group.create_dataset(
                    param_id,
                    shape=(1,),
                    maxshape=(None,),
                    dtype=dtype,
                )
                index = 0

            ds[index] = (timestamp.encode(), value)
            parameter_updates[param_id] = ParameterValue(timestamp, value)

        logger.debug(
            "Wrote parameter update for job %d at %s",
            job_id,
            timestamp,
        )
    emit_queue.put(
        {
            "event": f"experiment_params_{job_id}",
            "data": {
                param_id: asdict(val) for param_id, val in parameter_updates.items()
            },
        }
    )

delete_fit_result_by_job_id

delete_fit_result_by_job_id(
    *, job_id: int, result_channel: str
) -> None

Delete a fit result for a specific channel from the HDF5 file.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
result_channel	str	Name of the result channel whose fit to delete.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def delete_fit_result_by_job_id(*, job_id: int, result_channel: str) -> None:
    """Delete a fit result for a specific channel from the HDF5 file.

    Args:
        job_id: Job identifier.
        result_channel: Name of the result channel whose fit to delete.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    with h5_open(h5_path, "a") as h5file:
        if "fits" in h5file and result_channel in h5file["fits"]:
            del h5file["fits"][result_channel]

estimate_bytes_per_data_point

estimate_bytes_per_data_point(
    total: int,
    shot_channels_group: Group | None,
    result_channel_dataset: Group | None,
    vector_channels_group: Group | None,
    scan_parameters: Dataset | None,
) -> int

Estimate bytes per data point from HDF5 metadata.

Return total number of data points in h5file and estimated bytes per data point.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def estimate_bytes_per_data_point(
    total: int,
    shot_channels_group: h5py.Group | None,
    result_channel_dataset: h5py.Group | None,
    vector_channels_group: h5py.Group | None,
    scan_parameters: h5py.Dataset | None,
) -> int:
    """Estimate bytes per data point from HDF5 metadata.

    Return total number of data points in `h5file` and estimated bytes per data point.
    """
    bytes_per_point = sum(
        ds.shape[1] * ds.dtype.itemsize for ds in (shot_channels_group or {}).values()
    ) + sum(
        ds.dtype.itemsize
        for ds in (result_channel_dataset, scan_parameters)
        if ds is not None
    )

    # Add vector channel size (average across all data points)
    total_vector_bytes = sum(
        dataset.shape[0] * dataset.dtype.itemsize
        for channel_group in (vector_channels_group or {}).values()
        for dataset in cast("h5py.Group", channel_group).values()
    )
    if total > 0:
        bytes_per_point += total_vector_bytes // total
    # JSON serialisation roughly doubles the raw size
    return max(bytes_per_point * 2, 1)

get_filename_by_job_id

get_filename_by_job_id(job_id: int) -> str

Return the HDF5 filename for a job.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required

Returns:

Type	Description
str	Filename derived from the job’s scheduled time (e.g., “.h5”).

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def get_filename_by_job_id(job_id: int) -> str:
    """Return the HDF5 filename for a job.

    Args:
        job_id: Job identifier.

    Returns:
        Filename derived from the job's scheduled time (e.g., "<iso>.h5").
    """
    scheduled_time = JobRunRepository.get_scheduled_time_by_job_id(job_id=job_id)
    return f"{scheduled_time}.h5"

get_fit_results_by_job_id

get_fit_results_by_job_id(
    *, job_id: int
) -> dict[str, FitResult]

Read all fit results for a job from its HDF5 file.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required

Returns:

Type	Description
dict[str, FitResult]	Dict mapping result channel names to their fit result dicts.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def get_fit_results_by_job_id(*, job_id: int) -> dict[str, FitResult]:
    """Read all fit results for a job from its HDF5 file.

    Args:
        job_id: Job identifier.

    Returns:
        Dict mapping result channel names to their fit result dicts.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    if not h5_path.exists():
        return {}

    with h5_open(h5_path, "r") as h5file:
        return _read_fits_from_hdf5(h5file)

get_hdf5_dtype

get_hdf5_dtype(
    value: str | float | bool,
) -> type[float64 | bool | int64] | Datatype

Return the HDF5-compatible dtype.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def get_hdf5_dtype(
    value: str | float | bool,  # noqa: FBT001
) -> type[np.float64 | np.bool | np.int64] | h5py.Datatype:
    """Return the HDF5-compatible dtype."""
    if isinstance(value, str):
        return h5py.string_dtype()
    if isinstance(value, bool):
        return np.bool
    if isinstance(value, int):
        return np.int64
    if isinstance(value, float):
        return np.float64

    raise TypeError(f"Unsupported parameter type: {type(value)}")

load_experiment_data

load_experiment_data(
    h5file: File,
    max_transfer_bytes: int = 50000000,
    *,
    include_hardware_instructions: bool = False,
) -> ExperimentData

Load stored data for a job from its HDF5 file.

When loading all data would exceed max_transfer_bytes, only the last N data points that fit within the budget are returned. The budget is estimated from HDF5 metadata (channel count, shots per channel) without reading actual data.

Parameters:

Name	Type	Description	Default
h5file	File	File to load from.	required
max_transfer_bytes	int	Approximate cap on the serialised payload size in bytes. Defaults to 50 MB.	50000000
include_hardware_instructions	bool	Wether to include hardware instructions	False

Returns: Experiment data payload suitable for the API.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def load_experiment_data(
    h5file: h5py.File,
    max_transfer_bytes: int = 50_000_000,
    *,
    include_hardware_instructions: bool = False,
) -> ExperimentData:
    """Load stored data for a job from its HDF5 file.

    When loading all data would exceed *max_transfer_bytes*, only the
    last N data points that fit within the budget are returned.  The
    budget is estimated from HDF5 metadata (channel count, shots per
    channel) without reading actual data.

    Args:
        h5file: File to load from.
        max_transfer_bytes: Approximate cap on the serialised payload
            size in bytes.  Defaults to 50 MB.
        include_hardware_instructions: Wether to include hardware instructions
    Returns:
        Experiment data payload suitable for the API.
    """
    total = int(h5file.attrs.get("number_of_data_points", 0))
    data = ExperimentData(
        realtime_scan=bool(h5file.attrs.get("realtime_scan", False)),
        total_data_points=total,
    )
    shot_channels_group: h5py.Group | None = h5file.get("shot_channels")
    result_channel_dataset = h5file.get("result_channels")
    scan_parameters: h5py.Dataset | None = h5file.get("scan_parameters")
    vector_channels_group: h5py.Group | None = h5file.get("vector_channels")

    # Estimate bytes per data point from HDF5 metadata
    bytes_per_point = estimate_bytes_per_data_point(
        total,
        shot_channels_group,
        result_channel_dataset,
        vector_channels_group,
        scan_parameters,
    )

    max_data_points = max_transfer_bytes // bytes_per_point
    start_index = max(0, total - max_data_points)
    if start_index > 0:
        logger.info(
            "Loading last %d of %d data points (~%d bytes/point, %d MB budget)",
            total - start_index,
            total,
            bytes_per_point,
            max_transfer_bytes // 1_000_000,
        )

    if scan_parameters is not None:
        scan_parameters: npt.NDArray = scan_parameters[start_index:]  # type: ignore
        data.scan_parameters = {
            param: {
                start_index + i: value[0].item().decode()
                if isinstance(value[0], np.bytes_)
                else value[0].item()
                for i, value in enumerate(scan_parameters[param])
            }
            for param in cast("tuple[str, ...]", scan_parameters.dtype.names)
        }

    if result_channel_dataset is not None:
        plot_metadata: str | None = result_channel_dataset.attrs.get(
            "Plot window metadata"
        )
        data.plot_windows.result_channels = [
            PlotWindowMetadata(**d)
            for d in (json.loads(plot_metadata) if plot_metadata else [])
        ]
        result_channels = cast("npt.NDArray[Any]", result_channel_dataset[start_index:])  # type: ignore
        data.readouts.result_channels = {
            channel_name: dict(
                enumerate(
                    cast("list[float]", result_channels[channel_name].tolist()),
                    start=start_index,
                )
            )
            for channel_name in cast("tuple[str, ...]", result_channels.dtype.names)
        }

    # Convert shot channels into dicts with index as key
    if shot_channels_group is not None:
        plot_metadata = shot_channels_group.attrs.get("Plot window metadata")
        data.plot_windows.shot_channels = [
            PlotWindowMetadata(**d)
            for d in (json.loads(plot_metadata) if plot_metadata else [])
        ]
        data.readouts.shot_channels = {
            key: dict(enumerate(value[start_index:].tolist(), start=start_index))  # type: ignore
            for key, value in cast(
                "Sequence[tuple[str, h5py.Dataset]]",
                shot_channels_group.items(),
            )
        }

    if vector_channels_group is not None:
        plot_metadata = vector_channels_group.attrs.get("Plot window metadata")
        data.plot_windows.vector_channels = [
            PlotWindowMetadata(**d)
            for d in (json.loads(plot_metadata) if plot_metadata else [])
        ]
        data.readouts.vector_channels = {
            channel_name: {
                int(data_point): vector_dataset[:].tolist()
                for data_point, vector_dataset in cast(
                    "Sequence[tuple[str, h5py.Dataset]]", vector_group.items()
                )
            }
            for channel_name, vector_group in cast(
                "Sequence[tuple[str, h5py.Group]]",
                vector_channels_group.items(),
            )
        }

    if include_hardware_instructions:
        data.hardware_instructions = [
            (
                cast("np.int32", entry["index"]).item(),
                entry["Sequence"].decode(),
            )
            for entry in cast(
                "h5py.Dataset | tuple[()]", h5file.get("hardware_instructions", ())
            )
        ]
    data.parameters = extract_parameter_values(h5file)
    data.fits = _read_fits_from_hdf5(h5file)
    return data

resize_dataset

resize_dataset(
    dataset: Dataset, next_index: int, axis: int
) -> None

Resize a dataset to accommodate writing at a target index.

Parameters:

Name	Type	Description	Default
dataset	Dataset	HDF5 dataset to resize.	required
next_index	int	Index that must be writable.	required
axis	int	Axis along which to grow.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def resize_dataset(dataset: h5py.Dataset, next_index: int, axis: int) -> None:
    """Resize a dataset to accommodate writing at a target index.

    Args:
        dataset: HDF5 dataset to resize.
        next_index: Index that must be writable.
        axis: Axis along which to grow.
    """
    dataset.resize(next_index + 1, axis)

write_fit_result_by_job_id

write_fit_result_by_job_id(
    *, job_id: int, fit_result: FitResult
) -> None

Write a fit result into the HDF5 file for a job.

Creates or overwrites the fits/<result_channel> group.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
fit_result	FitResult	The fit result to persist.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_fit_result_by_job_id(
    *,
    job_id: int,
    fit_result: FitResult,
) -> None:
    """Write a fit result into the HDF5 file for a job.

    Creates or overwrites the ``fits/<result_channel>`` group.

    Args:
        job_id: Job identifier.
        fit_result: The fit result to persist.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    with h5_open(h5_path, "a") as h5file:
        fits_group = h5file.require_group("fits")
        channel = fit_result.result_channel
        if channel in fits_group:
            del fits_group[channel]
        grp = fits_group.create_group(channel)
        grp.attrs["fit_result"] = json.dumps(asdict(fit_result))

write_hardware_instructions_to_dataset

write_hardware_instructions_to_dataset(
    h5file: File,
    data_point_index: int,
    hardware_instructions: str,
) -> None

Append hardware instructions if it changed since the last entry.

Parameters:

Name	Type	Description	Default
h5file	File	Open HDF5 file handle.	required
data_point_index	int	Index of the current data point.	required
hardware_instructions	str	Serialized hardware instructions to append.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_hardware_instructions_to_dataset(
    h5file: h5py.File,
    data_point_index: int,
    hardware_instructions: str,
) -> None:
    """Append hardware instructions if it changed since the last entry.

    Args:
        h5file: Open HDF5 file handle.
        data_point_index: Index of the current data point.
        hardware_instructions: Serialized hardware instructions to append.
    """
    hw_instructions_dtype = [
        ("index", np.int32),
        ("Sequence", h5py.string_dtype()),
    ]
    hw_instructions_dataset = h5file.require_dataset(
        "hardware_instructions",
        shape=(0,),
        maxshape=(None,),
        chunks=True,
        dtype=hw_instructions_dtype,
        compression="gzip",
        compression_opts=9,
    )

    index = hw_instructions_dataset.shape[0]
    if index > 0:
        _, hw_instructions_old = cast(
            "tuple[int, bytes]", hw_instructions_dataset[index - 1]
        )
        if hw_instructions_old.decode() == hardware_instructions:
            logger.debug("Hardware instructions didn't change.")
            return

    resize_dataset(hw_instructions_dataset, next_index=index, axis=0)

    hw_instructions_dataset[index] = (
        data_point_index,
        hardware_instructions,
    )

write_results_to_dataset

write_results_to_dataset(
    h5file: File,
    data_point_index: int,
    result_channels: dict[str, float],
    number_of_data_points: int,
) -> None

Write scalar result channels into the ‘result_channels’ dataset.

Parameters:

Name	Type	Description	Default
h5file	File	Open HDF5 file handle.	required
data_point_index	int	Index of the current data point.	required
result_channels	dict[str, float]	Mapping of channel name to float value.	required
number_of_data_points	int	Current total number of stored data points.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_results_to_dataset(
    h5file: h5py.File,
    data_point_index: int,
    result_channels: dict[str, float],
    number_of_data_points: int,
) -> None:
    """Write scalar result channels into the 'result_channels' dataset.

    Args:
        h5file: Open HDF5 file handle.
        data_point_index: Index of the current data point.
        result_channels: Mapping of channel name to float value.
        number_of_data_points: Current total number of stored data points.
    """
    if not result_channels:
        return

    sorted_keys = sorted(result_channels)

    result_dataset = get_result_channels_dataset(
        h5file=h5file,
        result_channels=sorted_keys,
        number_of_data_points=number_of_data_points,
    )

    if set(result_dataset.dtype.names) != set(sorted_keys):
        raise RuntimeError(
            f"Result channels changed from {list(result_dataset.dtype.names)} to "
            f"{sorted_keys}"
        )

    if data_point_index >= number_of_data_points:
        resize_dataset(result_dataset, next_index=data_point_index, axis=0)

    result_dataset[data_point_index] = tuple(result_channels[k] for k in sorted_keys)

write_scan_parameters_and_timestamp_to_dataset

write_scan_parameters_and_timestamp_to_dataset(
    h5file: File,
    data_point_index: int,
    scan_params: dict[str, DatabaseValueType],
    timestamp: str,
    number_of_data_points: int,
) -> None

Write scan parameters and timestamp to the ‘scan_parameters’ dataset.

Parameters:

Name	Type	Description	Default
h5file	File	Open HDF5 file handle.	required
data_point_index	int	Index of the current data point.	required
scan_params	dict[str, DatabaseValueType]	Parameter values for this data point.	required
timestamp	str	Acquisition timestamp (ISO string).	required
number_of_data_points	int	Current total number of stored data points.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_scan_parameters_and_timestamp_to_dataset(
    h5file: h5py.File,
    data_point_index: int,
    scan_params: dict[str, DatabaseValueType],
    timestamp: str,
    number_of_data_points: int,
) -> None:
    """Write scan parameters and timestamp to the 'scan_parameters' dataset.

    Args:
        h5file: Open HDF5 file handle.
        data_point_index: Index of the current data point.
        scan_params: Parameter values for this data point.
        timestamp: Acquisition timestamp (ISO string).
        number_of_data_points: Current total number of stored data points.
    """
    scan_parameter_dtype = [
        ("timestamp", "S26"),  # timestamps are strings of length 26
        *[(key, np.float64) for key in scan_params],
    ]
    scan_params_dataset = h5file.require_dataset(
        "scan_parameters",
        shape=(number_of_data_points, 1),
        maxshape=(None, 1),
        chunks=True,
        dtype=scan_parameter_dtype,
        compression="gzip",
        compression_opts=9,
    )

    if data_point_index >= number_of_data_points:
        resize_dataset(scan_params_dataset, next_index=data_point_index, axis=0)

    parameter_values = tuple(scan_params[key] for key in scan_params)
    scan_params_dataset[data_point_index] = (
        timestamp,
        *parameter_values,
    )

write_shot_channels_to_datasets

write_shot_channels_to_datasets(
    h5file: File,
    data_point_index: int,
    shot_channels: dict[str, list[int]],
    number_of_data_points: int,
    number_of_shots: int,
) -> None

Write per-shot data into datasets under the ‘shot_channels’ group.

Parameters:

Name	Type	Description	Default
h5file	File	Open HDF5 file handle.	required
data_point_index	int	Index of the current data point.	required
shot_channels	dict[str, list[int]]	Mapping of channel to per-shot integers.	required
number_of_data_points	int	Current total number of stored data points.	required
number_of_shots	int	Expected number of shots per channel.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_shot_channels_to_datasets(
    h5file: h5py.File,
    data_point_index: int,
    shot_channels: dict[str, list[int]],
    number_of_data_points: int,
    number_of_shots: int,
) -> None:
    """Write per-shot data into datasets under the 'shot_channels' group.

    Args:
        h5file: Open HDF5 file handle.
        data_point_index: Index of the current data point.
        shot_channels: Mapping of channel to per-shot integers.
        number_of_data_points: Current total number of stored data points.
        number_of_shots: Expected number of shots per channel.
    """
    shot_group = h5file.require_group("shot_channels")
    for key, value in shot_channels.items():
        shot_dataset = shot_group.require_dataset(
            key,
            shape=(number_of_data_points, number_of_shots),
            maxshape=(None, number_of_shots),
            chunks=True,
            dtype=np.float64,
            compression="gzip",
            compression_opts=9,
        )

        if data_point_index >= number_of_data_points:
            resize_dataset(shot_dataset, next_index=data_point_index, axis=0)
        shot_dataset[data_point_index] = value

write_vector_channels_to_datasets

write_vector_channels_to_datasets(
    h5file: File,
    data_point_index: int,
    vector_channels: dict[str, list[float]],
) -> None

Write vector channel data under the ‘vector_channels’ group.

Creates one dataset per channel per data point.

Parameters:

Name	Type	Description	Default
h5file	File	Open HDF5 file handle.	required
data_point_index	int	Index of the current data point.	required
vector_channels	dict[str, list[float]]	Mapping of channel to vector of floats.	required

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py

def write_vector_channels_to_datasets(
    h5file: h5py.File,
    data_point_index: int,
    vector_channels: dict[str, list[float]],
) -> None:
    """Write vector channel data under the 'vector_channels' group.

    Creates one dataset per channel per data point.

    Args:
        h5file: Open HDF5 file handle.
        data_point_index: Index of the current data point.
        vector_channels: Mapping of channel to vector of floats.
    """
    vector_group = h5file.require_group("vector_channels")
    for channel_name, vector in vector_channels.items():
        channel_group = vector_group.require_group(channel_name)
        if str(data_point_index) not in channel_group:
            channel_group.create_dataset(
                str(data_point_index),
                data=vector,
                compression="gzip",
                compression_opts=9,
            )

experiment_source_repository

ExperimentSourceRepository

Repository for ExperimentSource entities.

Provides methods to query and persist experiment sources in the database. Encapsulates the SQLAlchemy session and query logic.

get_or_create_experiment staticmethod

get_or_create_experiment(
    *, experiment_source: ExperimentSource
) -> ExperimentSource

Return an existing experiment source or create it if not found.

Parameters:

Name	Type	Description	Default
experiment_source	ExperimentSource	The experiment source to look up by experiment_id. If no matching row exists, this instance is inserted into the database.	required

Returns:

Type	Description
ExperimentSource	The existing or newly created experiment source.

Source code in src/icon/server/data_access/repositories/experiment_source_repository.py

@staticmethod
def get_or_create_experiment(
    *,
    experiment_source: ExperimentSource,
) -> ExperimentSource:
    """Return an existing experiment source or create it if not found.

    Args:
        experiment_source: The experiment source to look up by `experiment_id`. If
            no matching row exists, this instance is inserted into the database.

    Returns:
        The existing or newly created experiment source.
    """
    with sqlalchemy.orm.Session(engine) as session:
        experiment = (
            session.query(ExperimentSource)
            .filter_by(experiment_id=experiment_source.experiment_id)
            .first()
        )

        if not experiment:
            experiment = experiment_source
            session.add(experiment)
            session.commit()
            session.refresh(experiment)  # Refresh to get the ID
            logger.debug("Inserted new experiment %s", experiment)

    return experiment

job_repository

JobRepository

Repository for Job entities.

Encapsulates SQLAlchemy session/query logic and emits Socket.IO events on changes. All methods open their own session and return detached ORM objects.

get_job_by_experiment_source_and_status staticmethod

get_job_by_experiment_source_and_status(
    *,
    experiment_source_id: int,
    status: JobStatus | None = None,
) -> Sequence[Row[tuple[Job]]]

List jobs for an experiment source, optionally filtered by status.

Parameters:

Name	Type	Description	Default
experiment_source_id	int	Foreign key of the experiment source.	required
status	JobStatus | None	Optional status filter.	None

Returns:

Type	Description
Sequence[Row[tuple[Job]]]	Rows containing Job objects, ordered by priority then creation time.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def get_job_by_experiment_source_and_status(
    *,
    experiment_source_id: int,
    status: JobStatus | None = None,
) -> Sequence[sqlalchemy.Row[tuple[Job]]]:
    """List jobs for an experiment source, optionally filtered by status.

    Args:
        experiment_source_id: Foreign key of the experiment source.
        status: Optional status filter.

    Returns:
        Rows containing `Job` objects, ordered by priority then creation time.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = select(Job).where(Job.experiment_source_id == experiment_source_id)

        if status:
            stmt = stmt.where(Job.status == status)

        stmt = stmt.options(
            sqlalchemy.orm.joinedload(Job.experiment_source)
        ).order_by(Job.priority.asc(), Job.created.asc())

        jobs = session.execute(stmt).all()
        logger.debug("Got jobs by experiment_source_id %s", experiment_source_id)
    return jobs

get_job_by_id staticmethod

get_job_by_id(
    *,
    job_id: int,
    load_experiment_source: bool = False,
    load_scan_parameters: bool = False,
) -> Job

Fetch a job by ID with optional eager-loading.

Parameters:

Name	Type	Description	Default
job_id	int	Job identifier.	required
load_experiment_source	bool	If True, eager-load experiment_source.	False
load_scan_parameters	bool	If True, eager-load scan_parameters.	False

Returns:

Type	Description
Job	The requested job.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def get_job_by_id(
    *,
    job_id: int,
    load_experiment_source: bool = False,
    load_scan_parameters: bool = False,
) -> Job:
    """Fetch a job by ID with optional eager-loading.

    Args:
        job_id: Job identifier.
        load_experiment_source: If True, eager-load `experiment_source`.
        load_scan_parameters: If True, eager-load `scan_parameters`.

    Returns:
        The requested job.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = select(Job).where(Job.id == job_id)

        if load_experiment_source:
            stmt = stmt.options(sqlalchemy.orm.joinedload(Job.experiment_source))
        if load_scan_parameters:
            stmt = stmt.options(sqlalchemy.orm.joinedload(Job.scan_parameters))

        return session.execute(stmt).unique().scalar_one()

get_jobs_by_status_and_timeframe staticmethod

get_jobs_by_status_and_timeframe(
    *,
    status: JobStatus | None = None,
    start: datetime | None = None,
    stop: datetime | None = None,
) -> Sequence[Job]

List jobs filtered by status and optional creation time window.

Parameters:

Name	Type	Description	Default
status	JobStatus | None	Optional status filter.	None
start	datetime | None	Inclusive start timestamp.	None
stop	datetime | None	Exclusive stop timestamp.	None

Returns:

Type	Description
Sequence[Job]	Matching jobs ordered by priority then creation time.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def get_jobs_by_status_and_timeframe(
    *,
    status: JobStatus | None = None,
    start: datetime.datetime | None = None,
    stop: datetime.datetime | None = None,
) -> Sequence[Job]:
    """List jobs filtered by status and optional creation time window.

    Args:
        status: Optional status filter.
        start: Inclusive start timestamp.
        stop: Exclusive stop timestamp.

    Returns:
        Matching jobs ordered by priority then creation time.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            select(Job)
            .options(sqlalchemy.orm.joinedload(Job.experiment_source))
            .options(sqlalchemy.orm.joinedload(Job.scan_parameters))
            .order_by(Job.priority.asc())
            .order_by(Job.created.asc())
        )

        if status is not None:
            stmt = stmt.where(Job.status == status)
        if start is not None:
            stmt = stmt.where(Job.created >= start)
        if stop is not None:
            stmt = stmt.where(Job.created < stop)

        return session.execute(stmt).unique().scalars().all()

resubmit_job_by_id staticmethod

resubmit_job_by_id(*, job_id: int) -> Job

Clone an existing job as a new submission.

If the source job is not itself a resubmission, the new job’s parent_job_id is set to the original job’s id.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job to clone.	required

Returns:

Type	Description
Job	The newly created job.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def resubmit_job_by_id(*, job_id: int) -> Job:
    """Clone an existing job as a new submission.

    If the source job is not itself a resubmission, the new job's `parent_job_id` is
    set to the original job's id.

    Args:
        job_id: ID of the job to clone.

    Returns:
        The newly created job.
    """
    with sqlalchemy.orm.Session(engine) as session:
        job = session.execute(
            sqlalchemy.select(Job).where(Job.id == job_id)
        ).scalar_one()
        sqlalchemy.orm.make_transient(job)

        if not job.parent_job_id:
            job.parent_job_id = job.id

        # PK and created are set by DB on insert
        job.id = None  # type: ignore
        job.created = None  # type: ignore

        session.add(job)
        session.commit()
        session.refresh(job)
        session.expunge(job)

    emit_queue.put(
        {
            "event": "job.new",
            "data": {
                "job": SQLAlchemyDictEncoder.encode(
                    JobRepository.get_job_by_id(
                        job_id=job.id,
                        load_experiment_source=True,
                        load_scan_parameters=True,
                    )
                ),
            },
        }
    )

    return job

submit_job staticmethod

submit_job(*, job: Job) -> Job

Insert a new job and emit a creation event.

Parameters:

Name	Type	Description	Default
job	Job	The job instance to persist.	required

Returns:

Type	Description
Job	The persisted job with generated fields populated.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def submit_job(*, job: Job) -> Job:
    """Insert a new job and emit a creation event.

    Args:
        job: The job instance to persist.

    Returns:
        The persisted job with generated fields populated.
    """
    with sqlalchemy.orm.Session(engine) as session:
        session.add(job)
        session.commit()
        session.refresh(job)
        session.expunge(job)

        logger.debug("Submitted new job %s", job)

    emit_queue.put(
        {
            "event": "job.new",
            "data": {
                "job": SQLAlchemyDictEncoder.encode(
                    JobRepository.get_job_by_id(
                        job_id=job.id,
                        load_experiment_source=True,
                        load_scan_parameters=True,
                    )
                ),
            },
        }
    )

    return job

update_job_status staticmethod

update_job_status(*, job: Job, status: JobStatus) -> Job

Update a job’s status and emit an update event.

Parameters:

Name	Type	Description	Default
job	Job	Job to update (identified by its id).	required
status	JobStatus	New job status.	required

Returns:

Type	Description
Job	The updated job with relationships loaded.

Source code in src/icon/server/data_access/repositories/job_repository.py

@staticmethod
def update_job_status(*, job: Job, status: JobStatus) -> Job:
    """Update a job's status and emit an update event.

    Args:
        job: Job to update (identified by its `id`).
        status: New job status.

    Returns:
        The updated job with relationships loaded.
    """
    with sqlalchemy.orm.Session(engine) as session:
        session.execute(update(Job).where(Job.id == job.id).values(status=status))
        session.commit()

        job = (
            session.execute(
                select(Job)
                .where(Job.id == job.id)
                .options(
                    sqlalchemy.orm.joinedload(Job.experiment_source),
                    sqlalchemy.orm.joinedload(Job.scan_parameters),
                )
            )
            .unique()
            .scalar_one()
        )
        session.expunge(job)

        logger.debug("Updated job %s", job)

    emit_queue.put(
        {
            "event": "job.update",
            "data": {
                "job_id": job.id,
                "updated_properties": {"status": status.value},
            },
        }
    )

    return job

job_run_repository

JobRunRepository

Repository for JobRun entities.

Provides methods to insert, update, and query job runs from the database. Emits Socket.IO events when job runs are created or updated.

get_parameter_update_timestamp staticmethod

get_parameter_update_timestamp(*, run_id: int) -> datetime

Get the paramter update timestamp.

Parameters:

Name	Type	Description	Default
run_id	int	ID of the job.	required

Returns:

Type	Description
datetime	The parameter update timestamp.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def get_parameter_update_timestamp(*, run_id: int) -> datetime:
    """Get the paramter update timestamp.

    Args:
        run_id: ID of the job.

    Returns:
        The parameter update timestamp.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = select(JobRun.parameter_update_timestamp).where(JobRun.id == run_id)

        timestamp = session.execute(stmt).scalar_one()
        logger.debug("Got parameter update timestamp for run %s", run_id)

    return timestamp.replace(tzinfo=UTC)

get_run_by_job_id staticmethod

get_run_by_job_id(
    *, job_id: int, load_job: bool = False
) -> JobRun

Return the run associated with a given job ID.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job.	required
load_job	bool	If True, eagerly load the related Job.	False

Returns:

Type	Description
JobRun	The run linked to the given job.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def get_run_by_job_id(*, job_id: int, load_job: bool = False) -> JobRun:
    """Return the run associated with a given job ID.

    Args:
        job_id: ID of the job.
        load_job: If True, eagerly load the related `Job`.

    Returns:
        The run linked to the given job.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            select(JobRun)
            .where(JobRun.job_id == job_id)
            .order_by(JobRun.scheduled_time.desc())
            .limit(1)
        )

        if load_job:
            stmt = stmt.options(sqlalchemy.orm.joinedload(JobRun.job))

        run = session.execute(stmt).scalar_one()
        logger.debug("Got JobRun by job_id %s", job_id)
    return run

get_runs_by_status staticmethod

get_runs_by_status(
    *,
    status: JobRunStatus | list[JobRunStatus],
    load_job: bool = False,
) -> Sequence[JobRun]

Return job runs filtered by status.

Parameters:

Name	Type	Description	Default
status	JobRunStatus | list[JobRunStatus]	Single or list of run statuses to filter on.	required
load_job	bool	If True, eagerly load the related Job.	False

Returns:

Type	Description
Sequence[JobRun]	All matching runs.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def get_runs_by_status(
    *,
    status: JobRunStatus | list[JobRunStatus],
    load_job: bool = False,
) -> Sequence[JobRun]:
    """Return job runs filtered by status.

    Args:
        status: Single or list of run statuses to filter on.
        load_job: If True, eagerly load the related `Job`.

    Returns:
        All matching runs.
    """
    if not isinstance(status, list):
        status = [status]

    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            select(JobRun)
            .where(JobRun.status.in_(status))
            .order_by(JobRun.scheduled_time.asc())
        )

        if load_job:
            stmt = stmt.options(sqlalchemy.orm.joinedload(JobRun.job))

        return session.execute(stmt).scalars().all()

get_scheduled_time_by_job_id staticmethod

get_scheduled_time_by_job_id(*, job_id: int) -> datetime

Return the scheduled time of a run by job ID.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job.	required

Returns:

Type	Description
datetime	The scheduled start time of the run.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def get_scheduled_time_by_job_id(*, job_id: int) -> datetime:
    """Return the scheduled time of a run by job ID.

    Args:
        job_id: ID of the job.

    Returns:
        The scheduled start time of the run.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            select(JobRun.scheduled_time)
            .where(JobRun.job_id == job_id)
            .order_by(JobRun.scheduled_time.desc())
            .limit(1)
        )

        scheduled_time = session.execute(stmt).scalar_one()
        logger.debug("Got scheduled time for job_id %s", job_id)
    return scheduled_time

insert_run staticmethod

insert_run(*, run: JobRun) -> JobRun

Insert a new job run and emit a creation event.

Parameters:

Name	Type	Description	Default
run	JobRun	The job run instance to persist.	required

Returns:

Type	Description
JobRun	The persisted job run with generated fields populated.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def insert_run(*, run: JobRun) -> JobRun:
    """Insert a new job run and emit a creation event.

    Args:
        run: The job run instance to persist.

    Returns:
        The persisted job run with generated fields populated.
    """
    with sqlalchemy.orm.Session(engine) as session:
        session.add(run)
        session.commit()
        session.refresh(run)
        logger.debug("Created new run %s", run)

    emit_queue.put(
        {
            "event": "job_run.new",
            "data": {
                "job_run": SQLAlchemyDictEncoder.encode(obj=run),
            },
        }
    )

    return run

set_parameter_update_timestamp staticmethod

set_parameter_update_timestamp(
    *, run_id: int, timestamp: datetime
) -> None

Set the paramter update timestamp.

Parameters:

Name	Type	Description	Default
run_id	int	ID of the job.	required
timestamp	datetime	New parameter update timestamp.	required

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def set_parameter_update_timestamp(*, run_id: int, timestamp: datetime) -> None:
    """Set the paramter update timestamp.

    Args:
        run_id: ID of the job.
        timestamp: New parameter update timestamp.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            update(JobRun)
            .where(JobRun.id == run_id)
            .values(parameter_update_timestamp=timestamp.astimezone(UTC))
            .returning(JobRun)
        )

        run = session.execute(stmt).scalar_one()
        session.commit()

        logger.debug("Updated parameter update timestam for run %s", run)

update_run_by_id staticmethod

update_run_by_id(
    *,
    run_id: int,
    status: JobRunStatus,
    log: str | None = None,
) -> JobRun

Update a job run by ID and emit an update event.

Parameters:

Name	Type	Description	Default
run_id	int	The ID of the job run to update.	required
status	JobRunStatus	New status of the run.	required
log	str | None	Optional log message (e.g. failure reason).	None

Returns:

Type	Description
JobRun	The updated job run.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

@staticmethod
def update_run_by_id(
    *,
    run_id: int,
    status: JobRunStatus,
    log: str | None = None,
) -> JobRun:
    """Update a job run by ID and emit an update event.

    Args:
        run_id: The ID of the job run to update.
        status: New status of the run.
        log: Optional log message (e.g. failure reason).

    Returns:
        The updated job run.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            update(JobRun)
            .where(JobRun.id == run_id)
            .values(status=status, log=log)
            .returning(JobRun)
        )
        run = session.execute(stmt).scalar_one()
        session.commit()

        logger.debug("Updated run %s", run)

    emit_queue.put(
        {
            "event": "job_run.update",
            "data": {
                "run_id": run_id,
                "updated_properties": {
                    "status": status.value,
                    "log": log,
                },
            },
        }
    )

    return run

job_run_cancelled_or_failed

job_run_cancelled_or_failed(job_id: int) -> bool

Check if a job’s run was cancelled or failed.

Parameters:

Name	Type	Description	Default
job_id	int	ID of the job whose run should be checked.	required

Returns:

Type	Description
bool	True if the run status is CANCELLED or FAILED, False otherwise.

Source code in src/icon/server/data_access/repositories/job_run_repository.py

def job_run_cancelled_or_failed(job_id: int) -> bool:
    """Check if a job's run was cancelled or failed.

    Args:
        job_id: ID of the job whose run should be checked.

    Returns:
        True if the run status is CANCELLED or FAILED, False otherwise.
    """
    job_run = JobRunRepository.get_run_by_job_id(job_id=job_id)
    if job_run.status in (JobRunStatus.CANCELLED, JobRunStatus.FAILED):
        logger.info(
            "JobRun with id %s %s.",
            job_run.id,
            job_run.status.value,
        )
        return True
    return False

parameters_repository

NotInitialisedError

Bases: Exception

Raised when repository methods are called before initialization.

ParametersRepository

Repository for parameter values and metadata.

Provides methods to read and update shared parameter state (via a multiprocessing.Manager dict) and to persist/retrieve parameters from InfluxDB. Emits Socket.IO events on updates.

get_influxdb_parameter_by_id staticmethod

get_influxdb_parameter_by_id(
    parameter_id: str,
) -> DatabaseValueType | None

Return a single parameter value from InfluxDB.

Parameters:

Name	Type	Description	Default
parameter_id	str	ID of the parameter.	required

Returns:

Type	Description
DatabaseValueType | None	The parameter value, or None if not found.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@staticmethod
def get_influxdb_parameter_by_id(parameter_id: str) -> DatabaseValueType | None:
    """Return a single parameter value from InfluxDB.

    Args:
        parameter_id: ID of the parameter.

    Returns:
        The parameter value, or None if not found.
    """
    with InfluxDBv1Session() as influxdb:
        result_dict = influxdb.query(
            measurement=get_config().databases.influxdbv1.measurement,
            field=parameter_id,
        )
        if result_dict is None:
            logger.error(
                "Could not find parameter with id %s in database %s",
                parameter_id,
                get_config().databases.influxdbv1.measurement,
            )
            return None
        return result_dict[parameter_id]

get_influxdb_parameter_keys staticmethod

get_influxdb_parameter_keys() -> list[str]

Return all parameter field keys from InfluxDB v1.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@staticmethod
def get_influxdb_parameter_keys() -> list[str]:
    """Return all parameter field keys from InfluxDB v1."""
    with InfluxDBv1Session() as influxdbv1:
        return influxdbv1.get_field_keys(
            get_config().databases.influxdbv1.measurement
        )

get_influxdb_parameters staticmethod

get_influxdb_parameters(
    *,
    before: str | None = None,
    namespace: str | None = None,
) -> dict[str, DatabaseValueType]

Return the latest parameter values from InfluxDB.

Parameters:

Name	Type	Description	Default
before	str | None	Optional ISO timestamp to query parameters before.	None
namespace	str | None	Optional namespace filter.	None

Returns:

Type	Description
dict[str, DatabaseValueType]	Mapping of parameter IDs to values.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@staticmethod
def get_influxdb_parameters(
    *, before: str | None = None, namespace: str | None = None
) -> dict[str, DatabaseValueType]:
    """Return the latest parameter values from InfluxDB.

    Args:
        before: Optional ISO timestamp to query parameters before.
        namespace: Optional namespace filter.

    Returns:
        Mapping of parameter IDs to values.
    """
    with InfluxDBv1Session() as influxdbv1:
        return influxdbv1.query_last(
            get_config().databases.influxdbv1.measurement,
            before=before,
            namespace=namespace,
        )

get_shared_parameter_by_id classmethod

get_shared_parameter_by_id(
    *, parameter_id: str
) -> DatabaseValueType | None

Return a single parameter value from shared state.

Parameters:

Name	Type	Description	Default
parameter_id	str	ID of the parameter.	required

Returns:

Type	Description
DatabaseValueType | None	The parameter value, or None if not set.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@classmethod
def get_shared_parameter_by_id(
    cls,
    *,
    parameter_id: str,
) -> DatabaseValueType | None:
    """Return a single parameter value from shared state.

    Args:
        parameter_id: ID of the parameter.

    Returns:
        The parameter value, or None if not set.
    """
    cls._check_initialised()

    return cls._shared_parameters.get(parameter_id, None)

get_shared_parameters classmethod

get_shared_parameters() -> DictProxy[
    str, DatabaseValueType
]

Return the full shared parameter dictionary.

Returns:

Type	Description
DictProxy[str, DatabaseValueType]	Proxy dictionary of parameters.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@classmethod
def get_shared_parameters(cls) -> DictProxy[str, DatabaseValueType]:
    """Return the full shared parameter dictionary.

    Returns:
        Proxy dictionary of parameters.
    """
    cls._check_initialised()

    return cls._shared_parameters

initialize classmethod

initialize(
    *, shared_parameters: DictProxy[str, DatabaseValueType]
) -> None

Initialize the repository with a shared parameters dict.

Parameters:

Name	Type	Description	Default
shared_parameters	DictProxy[str, DatabaseValueType]	Proxy dictionary used to store shared state.	required

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@classmethod
def initialize(
    cls, *, shared_parameters: DictProxy[str, DatabaseValueType]
) -> None:
    """Initialize the repository with a shared parameters dict.

    Args:
        shared_parameters: Proxy dictionary used to store shared state.
    """
    cls._shared_parameters = shared_parameters
    cls.initialised = True

update_parameters classmethod

update_parameters(
    *, parameter_mapping: dict[str, DatabaseValueType]
) -> None

Update parameters in both shared state and InfluxDB.

Parameters:

Name	Type	Description	Default
parameter_mapping	dict[str, DatabaseValueType]	Mapping of parameter IDs to values.	required

Source code in src/icon/server/data_access/repositories/parameters_repository.py

@classmethod
def update_parameters(
    cls,
    *,
    parameter_mapping: dict[str, DatabaseValueType],
) -> None:
    """Update parameters in both shared state and InfluxDB.

    Args:
        parameter_mapping: Mapping of parameter IDs to values.
    """
    for key, value in parameter_mapping.items():
        if (
            isinstance(value, int)
            and not isinstance(value, bool)
            and "ParameterTypes.INT" not in key
        ):
            parameter_mapping[key] = float(value)

    cls._update_shared_parameters(parameter_mapping=parameter_mapping)
    cls._update_influxdb_parameters(parameter_mapping=parameter_mapping)

get_specifiers_from_parameter_identifier

get_specifiers_from_parameter_identifier(
    parameter_identifier: str,
) -> dict[str, str]

Extract specifiers from a parameter identifier string.

Parameter identifiers encode metadata as key='value' pairs. This helper parses them into a dictionary.

Parameters:

Name	Type	Description	Default
parameter_identifier	str	Identifier string to parse.	required

Returns:

Type	Description
dict[str, str]	Mapping of specifier keys to values.

Source code in src/icon/server/data_access/repositories/parameters_repository.py

def get_specifiers_from_parameter_identifier(
    parameter_identifier: str,
) -> dict[str, str]:
    """Extract specifiers from a parameter identifier string.

    Parameter identifiers encode metadata as `key='value'` pairs. This helper parses
    them into a dictionary.

    Args:
        parameter_identifier: Identifier string to parse.

    Returns:
        Mapping of specifier keys to values.
    """
    pattern = re.compile(r"(\w+)='([^']*)'")
    matches = pattern.findall(parameter_identifier)

    return dict(matches)

icon.server.hardware_processing

Modules:

Name	Description
devices
hardware_controller
task
utils
worker
zedboard_controller

devices

Classes:

Name	Description
Devices
Hardware

Functions:

Name	Description
load

Devices

Devices()

Methods:

Name	Description
__getitem__
items
main_device
reload

Source code in src/icon/server/hardware_processing/devices.py

def __init__(self) -> None:
    self.__devices: dict[str, Hardware] = {}
    self.__reloader = DictReloader(
        initial_objs=self.__devices,
        obj_factory=load,
        subconfig=lambda config: {
            dev["id"]: dev for dev in config.hardware.model_dump()["devices"]
        },
    )

__devices instance-attribute

__devices: dict[str, Hardware] = {}

__reloader instance-attribute

__reloader = DictReloader(initial_objs=__devices, obj_factory=load, subconfig=lambda config: {(dev['id']): _wefor dev in (model_dump()['devices'])})

__getitem__

__getitem__(dev_id: str) -> Hardware

Source code in src/icon/server/hardware_processing/devices.py

def __getitem__(self, dev_id: str) -> Hardware:
    self.reload()
    return self.__devices[dev_id]

items

items() -> Iterable[tuple[str, Hardware]]

Source code in src/icon/server/hardware_processing/devices.py

def items(self) -> Iterable[tuple[str, Hardware]]:
    self.reload()
    return self.__devices.items()

main_device

main_device() -> HardwareController

Source code in src/icon/server/hardware_processing/devices.py

def main_device(self) -> HardwareController:
    self.reload()
    try:
        return next(
            dev.controller
            for dev in self.__devices.values()
            if dev.enabled and isinstance(dev.controller, HardwareController)
        )
    except StopIteration:
        return FallbackHardwareController()

reload

reload() -> None

Source code in src/icon/server/hardware_processing/devices.py

def reload(self) -> None:
    reloaded_devices = self.__reloader.reload_changed()
    py_ids = {id(dev) for dev in reloaded_devices}
    # Reconnect changed / new / disconnected:
    for dev in self.__devices.values():
        if isinstance(dev.controller, HardwareController) and (
            id(dev) in py_ids or not dev.controller.connected
        ):
            dev.controller.connect()

Hardware dataclass

Hardware(
    controller: HardwareController | ReloadError,
    enabled: bool,
)

Attributes:

Name	Type	Description
controller	HardwareController | ReloadError
enabled	bool

controller instance-attribute

controller: HardwareController | ReloadError

enabled instance-attribute

enabled: bool

load

load(
    controller_module: str,
    controller_class: str,
    id: str,
    args: dict[str, Any],
    *,
    enabled: bool,
) -> Hardware

Source code in src/icon/server/hardware_processing/devices.py

def load(
    controller_module: str,
    controller_class: str,
    id: str,
    args: dict[str, Any],
    *,
    enabled: bool,
) -> Hardware:
    try:
        dev_module = importlib.import_module(controller_module)
        dev_class = getattr(dev_module, controller_class)
        return Hardware(controller=dev_class(**args), enabled=enabled)
    except (ImportError, AttributeError) as e:
        return Hardware(
            controller=ReloadError(
                f"Configuration for device {id} is invalid.\n"
                f"Error message: {e}\n"
                "Please reconfigure!"
            ),
            enabled=enabled,
        )

hardware_controller

Classes:

Name	Description
FallbackHardwareController	Noop hardware controller.
HardwareController
StatusFlag

Attributes:

Name	Type	Description
logger

logger module-attribute

logger = getLogger(__name__)

FallbackHardwareController

Bases: HardwareController

Noop hardware controller.

Methods:

Name	Description
connect
receive
run
send
status

Attributes:

Name	Type	Description
connected	bool

connected property

connected: bool

connect

connect() -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def connect(self) -> None:
    pass

receive

receive() -> Readouts

Source code in src/icon/server/hardware_processing/hardware_controller.py

def receive(self) -> Readouts:
    return Readouts(result_channels={}, vector_channels={}, shot_channels={})

run

run() -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def run(self) -> None:
    pass

send

send(data: str) -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def send(self, data: str) -> None:
    pass

status

status() -> tuple[StatusFlag, str, Any]

Source code in src/icon/server/hardware_processing/hardware_controller.py

def status(self) -> tuple[StatusFlag, str, Any]:
    return (StatusFlag.SUCCESS, "OK", ...)

HardwareController

Methods:

Name	Description
connect
receive
run
send
status

Attributes:

Name	Type	Description
connected	bool

connected property

connected: bool

connect

connect() -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def connect(self) -> None:
    raise NotImplementedError("Must be implemented by a derived class")

receive

receive() -> Readouts

Source code in src/icon/server/hardware_processing/hardware_controller.py

def receive(self) -> Readouts:
    raise NotImplementedError("Must be implemented by a derived class")

run

run() -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def run(self) -> None:
    raise NotImplementedError("Must be implemented by a derived class")

send

send(data: str) -> None

Source code in src/icon/server/hardware_processing/hardware_controller.py

def send(self, data: str) -> None:
    raise NotImplementedError("Must be implemented by a derived class")

status

status() -> tuple[StatusFlag, str, Any]

Source code in src/icon/server/hardware_processing/hardware_controller.py

def status(self) -> tuple[StatusFlag, str, Any]:
    raise NotImplementedError("Must be implemented by a derived class")

StatusFlag

Bases: Enum

Attributes:

Name	Type	Description
ERROR
SUCCESS
UNKNOWN

ERROR class-attribute instance-attribute

ERROR = auto()

SUCCESS class-attribute instance-attribute

SUCCESS = auto()

UNKNOWN class-attribute instance-attribute

UNKNOWN = auto()

task

Classes:

Name	Description
HardwareProcessingTask

HardwareProcessingTask

Bases: BaseModel

Methods:

Name	Description
__lt__

Attributes:

Name	Type	Description
created	datetime
data_point_index	int
data_points_to_process	Queue[tuple[int, dict[str, DatabaseValueType]]]
global_parameter_timestamp	datetime
hardware_instructions	str
model_config
outdated_tasks	PriorityQueue[HardwareProcessingTask]
pre_processing_task	PreProcessingTask
priority	int
processed_data_points	Queue[HardwareProcessingTask]
scanned_params	dict[str, DatabaseValueType]
src_dir	str | None

created instance-attribute

created: datetime

data_point_index instance-attribute

data_point_index: int

data_points_to_process instance-attribute

data_points_to_process: Queue[
    tuple[int, dict[str, DatabaseValueType]]
]

global_parameter_timestamp instance-attribute

global_parameter_timestamp: datetime

hardware_instructions instance-attribute

hardware_instructions: str

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

outdated_tasks instance-attribute

outdated_tasks: PriorityQueue[HardwareProcessingTask]

pre_processing_task instance-attribute

pre_processing_task: PreProcessingTask

priority instance-attribute

priority: int

processed_data_points instance-attribute

processed_data_points: Queue[HardwareProcessingTask]

scanned_params instance-attribute

scanned_params: dict[str, DatabaseValueType]

src_dir instance-attribute

src_dir: str | None

__lt__

__lt__(other: HardwareProcessingTask) -> bool

Source code in src/icon/server/hardware_processing/task.py

def __lt__(self, other: HardwareProcessingTask) -> bool:
    return (self.priority, self.created) < (other.priority, other.created)

utils

Functions:

Name	Description
extract_hardware_error_message	Extract the error message from a hardware exception.

extract_hardware_error_message

extract_hardware_error_message(exception: Exception) -> str

Extract the error message from a hardware exception.

tiqi_rpc wraps hardware errors as::

RPCResponseError("Server reported msgid <N> error: [<code>, '<message>']")

This strips the RPC wrapper and the error-code list, returning just the hardware message. For any other exception the full string is returned.

Source code in src/icon/server/hardware_processing/utils.py

def extract_hardware_error_message(exception: Exception) -> str:
    """Extract the error message from a hardware exception.

    ``tiqi_rpc`` wraps hardware errors as::

        RPCResponseError("Server reported msgid <N> error: [<code>, '<message>']")

    This strips the RPC wrapper and the error-code list, returning just the
    hardware message.  For any other exception the full string is returned.
    """
    msg = str(exception)
    # Strip "Server reported msgid <N> error: " prefix
    match = re.search(r"error: (.+)$", msg)
    if not match:
        return msg
    payload = match.group(1)
    # The RPC error payload is a repr of a Python object (typically a list
    # like [0, 'message']).  Try to parse it structurally so we don't depend
    # on regex for every possible error shape.
    try:
        parsed = ast.literal_eval(payload)
    except (ValueError, SyntaxError):
        return payload
    if isinstance(parsed, list | tuple):
        # Return the first string element (the human-readable message).
        for item in parsed:
            if isinstance(item, str):
                return item
    return payload

worker

Classes:

Name	Description
HardwareProcessingWorker

Functions:

Name	Description
parse_parameter_id	Parses a parameter ID string into a device name and variable ID.
should_divert_task	Whether the hardware worker should divert a task back to pre-processing.

Attributes:

Name	Type	Description
logger
timezone

logger module-attribute

logger = getLogger(__name__)

timezone module-attribute

timezone = timezone(timezone)

HardwareProcessingWorker

HardwareProcessingWorker(
    hardware_processing_queue: PriorityQueue[
        HardwareProcessingTask
    ],
    post_processing_queue: Queue[PostProcessingTask],
    manager: SharedResourceManager,
    devices: Devices,
)

Bases: Process

Methods:

Name	Description
run

Source code in src/icon/server/hardware_processing/worker.py

def __init__(
    self,
    hardware_processing_queue: queue.PriorityQueue[HardwareProcessingTask],
    post_processing_queue: multiprocessing.Queue[PostProcessingTask],
    manager: SharedResourceManager,
    devices: Devices,
) -> None:
    super().__init__()
    self._queue = hardware_processing_queue
    self._post_processing_queue = post_processing_queue
    self._manager = manager
    self._pydase_clients: dict[str, pydase.Client] = {}

    self._devices = devices

run

run() -> None

Source code in src/icon/server/hardware_processing/worker.py

@handle_keyboard_interrupt(logger)
def run(self) -> None:
    self._pydase_clients = {
        device.name: pydase.Client(
            url=device.url, block_until_connected=False, auto_update_proxy=False
        )
        for device in DeviceRepository.get_devices_by_status(
            status=DeviceStatus.ENABLED
        )
    }

    while True:
        task = self._queue.get()

        # One fetch covers both checks: the run carries the current status
        # (cancel/pause) and the parameter-update timestamp.
        job_run = JobRunRepository.get_run_by_job_id(
            job_id=task.pre_processing_task.job.id,
        )
        if job_run.status in (JobRunStatus.CANCELLED, JobRunStatus.FAILED):
            task.processed_data_points.put(task)
            continue

        if should_divert_task(
            task,
            job_run.parameter_update_timestamp,
            job_run.status,
        ):
            task.outdated_tasks.put(task)
            continue
        try:
            self._set_pydase_service_values(scanned_params=task.scanned_params)

            timestamp = datetime.now(timezone)
            hardware_controller = self._devices.main_device()
            hardware_controller.send(data=task.hardware_instructions)
            hardware_controller.run()
            readouts = hardware_controller.receive()

            experiment_data_point = ExperimentDataPoint(
                index=task.data_point_index,
                scan_params=task.scanned_params,
                readouts=readouts,
                timestamp=timestamp.isoformat(),
                hardware_instructions=task.hardware_instructions,
            )

            post_processing_task = PostProcessingTask(
                priority=task.priority,
                pre_processing_task=task.pre_processing_task,
                data_point=experiment_data_point,
                src_dir=task.src_dir,
                created=task.created,
            )

            self._post_processing_queue.put(post_processing_task)
        except Exception as e:
            logger.exception("pydase error")
            JobRunRepository.update_run_by_id(
                run_id=task.pre_processing_task.job_run.id,
                status=JobRunStatus.FAILED,
                log=extract_hardware_error_message(e),
            )
        finally:
            task.processed_data_points.put(task)

parse_parameter_id

parse_parameter_id(param_id: str) -> tuple[str | None, str]

Parses a parameter ID string into a device name and variable ID.

If the input string is in the format “Device(device_name) variable_id”, the device name and variable ID are returned as a tuple.

Parameters:

Name	Type	Description	Default
param_id	str	The parameter identifier string.	required

Returns:

Type	Description
str | None	A tuple (device_name, variable_id). If the input does not match the expected
str	format, device_name is None and the entire param_id is returned as the
tuple[str | None, str]	variable_id.

Examples:

>>> parse_parameter_id("Device(my_device) my_param")
('my_device', 'my_param')

>>> parse_parameter_id("bare_param")
(None, 'bare_param')

Source code in src/icon/server/hardware_processing/worker.py

def parse_parameter_id(param_id: str) -> tuple[str | None, str]:
    """Parses a parameter ID string into a device name and variable ID.

    If the input string is in the format "Device(device_name) variable_id",
    the device name and variable ID are returned as a tuple.

    Parameters:
        param_id: The parameter identifier string.

    Returns:
        A tuple (device_name, variable_id). If the input does not match the expected
        format, device_name is None and the entire param_id is returned as the
        variable_id.

    Examples:
        >>> parse_parameter_id("Device(my_device) my_param")
        ('my_device', 'my_param')

        >>> parse_parameter_id("bare_param")
        (None, 'bare_param')
    """
    match = re.match(r"^Device\(([^)]+)\) (.*)$", param_id)
    if match:
        return match[1], match[2]
    return None, param_id

should_divert_task

should_divert_task(
    task: HardwareProcessingTask,
    parameter_update_timestamp: datetime | None,
    job_run_status: JobRunStatus,
) -> bool

Whether the hardware worker should divert a task back to pre-processing.

A paused job always diverts. Otherwise a task is diverted when its parameters went stale (it was built before the last parameter update) – except for realtime scans, whose sequences the realtime handler regenerates in place, so diverting a stale realtime task would just bounce it back and forth in a tight loop.

parameter_update_timestamp is stored without timezone info (as UTC), so it is made timezone-aware before comparing with the task’s timezone-aware created.

Source code in src/icon/server/hardware_processing/worker.py

def should_divert_task(
    task: HardwareProcessingTask,
    parameter_update_timestamp: datetime | None,
    job_run_status: JobRunStatus,
) -> bool:
    """Whether the hardware worker should divert a task back to pre-processing.

    A paused job always diverts. Otherwise a task is diverted when its parameters
    went stale (it was built before the last parameter update) -- except for realtime
    scans, whose sequences the realtime handler regenerates in place, so diverting a
    stale realtime task would just bounce it back and forth in a tight loop.

    ``parameter_update_timestamp`` is stored without timezone info (as UTC), so it is
    made timezone-aware before comparing with the task's timezone-aware ``created``.
    """
    if job_run_status == JobRunStatus.PAUSED:
        return True
    if contains_realtime_parameter(task.pre_processing_task.scan_parameters):
        return False
    return (
        parameter_update_timestamp is not None
        and task.created < parameter_update_timestamp.replace(tzinfo=UTC)
    )

zedboard_controller

Classes:

Name	Description
ZedboardController

Attributes:

Name	Type	Description
logger

logger module-attribute

logger = getLogger(__name__)

ZedboardController

ZedboardController(*, host: str, port: int)

Bases: HardwareController

Methods:

Name	Description
connect
receive
run
send
status

Attributes:

Name	Type	Description
connected	bool

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def __init__(self, *, host: str, port: int) -> None:
    self._host = host
    self._port = port
    self._zedboard: tiqi_zedboard.zedboard.Zedboard | None = None

connected property

connected: bool

connect

connect() -> None

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def connect(self) -> None:
    logger.info("Connecting to the Zedboard")
    self._zedboard = tiqi_zedboard.zedboard.Zedboard(
        hostname=self._host, port=self._port
    )
    if not self.connected:
        logger.warning("Failed to connect to the Zedboard")

receive

receive() -> Readouts

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def receive(self) -> Readouts:
    results: tiqi_zedboard.zedboard.Result = self._zedboard.sequence_JSON_parser()  # type: ignore

    return Readouts(
        result_channels=results.result_channels,
        vector_channels=results.vector_channels
        if results.vector_channels is not None
        else {},
        shot_channels=results.shot_channels,
    )

run

run() -> None

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def run(self) -> None:
    self._zedboard.sequence_JSON_parser.Parse_JSON_Header()  # type: ignore

send

send(data: str) -> None

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def send(self, data: str) -> None:
    if not self.connected:
        self.connect()
    if not self.connected:
        raise RuntimeError("Could not connect to the Zedboard")
    self._update_zedboard_sequence(sequence=data)

status

status() -> tuple[StatusFlag, str, Any]

Source code in src/icon/server/hardware_processing/zedboard_controller.py

def status(self) -> tuple[StatusFlag, str, Any]:
    return (StatusFlag.UNKNOWN, "", None)

icon.server.post_processing

Modules:

Name	Description
task
worker

task

Classes:

Name	Description
PostProcessingTask

PostProcessingTask

Bases: BaseModel

Methods:

Name	Description
__lt__

Attributes:

Name	Type	Description
created	datetime
data_point	ExperimentDataPoint
pre_processing_task	PreProcessingTask
priority	int
src_dir	str | None

created instance-attribute

created: datetime

data_point instance-attribute

data_point: ExperimentDataPoint

pre_processing_task instance-attribute

pre_processing_task: PreProcessingTask

priority instance-attribute

priority: int

src_dir instance-attribute

src_dir: str | None

__lt__

__lt__(other: PostProcessingTask) -> bool

Source code in src/icon/server/post_processing/task.py

def __lt__(self, other: PostProcessingTask) -> bool:
    return self.priority < other.priority

worker

Classes:

Name	Description
PostProcessingWorker

Attributes:

Name	Type	Description
logger

logger module-attribute

logger = getLogger(__name__)

PostProcessingWorker

PostProcessingWorker(
    post_processing_queue: Queue[PostProcessingTask],
)

Bases: Process

Methods:

Name	Description
run

Source code in src/icon/server/post_processing/worker.py

def __init__(
    self,
    post_processing_queue: multiprocessing.Queue[PostProcessingTask],
) -> None:
    super().__init__()
    self._post_processing_queue = post_processing_queue

run

run() -> None

Source code in src/icon/server/post_processing/worker.py

@handle_keyboard_interrupt(logger)
def run(self) -> None:
    logger.info("Pre-processing worker started")

    while True:
        task = self._post_processing_queue.get()

        if job_run_cancelled_or_failed(
            job_id=task.pre_processing_task.job.id,
        ):
            continue

        ExperimentDataRepository.write_experiment_data_by_job_id(
            job_id=task.pre_processing_task.job.id,
            data_point=task.data_point,
        )

icon.server.pre_processing

Modules:

Name	Description
task
worker

task

Classes:

Name	Description
PreProcessingTask

PreProcessingTask

Bases: BaseModel

Methods:

Name	Description
__lt__

Attributes:

Name	Type	Description
auto_calibration	bool
debug_mode	bool
git_commit_hash	str | None
job	Job
job_run	JobRun
local_parameters_timestamp	str
model_config
priority	int
repetitions	int
scan_parameters	list[ScanParameter]

auto_calibration instance-attribute

auto_calibration: bool

debug_mode class-attribute instance-attribute

debug_mode: bool = False

git_commit_hash class-attribute instance-attribute

git_commit_hash: str | None = None

job instance-attribute

job: Job

job_run instance-attribute

job_run: JobRun

local_parameters_timestamp instance-attribute

local_parameters_timestamp: str

model_config class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True)

priority class-attribute instance-attribute

priority: int = Field(ge=0, le=20)

repetitions class-attribute instance-attribute

repetitions: int = 1

scan_parameters instance-attribute

scan_parameters: list[ScanParameter]

__lt__

__lt__(other: PreProcessingTask) -> bool

Source code in src/icon/server/pre_processing/task.py

def __lt__(self, other: "PreProcessingTask") -> bool:
    return self.priority < other.priority

worker

Classes:

Name	Description
ExperimentIdentifier
ParamUpdateMode
PreProcessingWorker

Functions:

Name	Description
change_process_priority	Changes process priority.
consume_queue
create_hardware_instructions
freeze_dict
get_scan_combinations	Generates all combinations of scan parameters for a given job.
parse_experiment_identifier	Parses an experiment identifier.

Attributes:

Name	Type	Description
ScanCombination
T
logger
timezone

ScanCombination module-attribute

ScanCombination = frozenset[tuple[str, DatabaseValueType]]

T module-attribute

T = TypeVar('T')

logger module-attribute

logger = getLogger(__name__)

timezone module-attribute

timezone = timezone(timezone)

ExperimentIdentifier dataclass

ExperimentIdentifier(
    module_name: str, class_name: str, instance_name: str
)

Methods:

Name	Description
__str__
from_str	Parses an experiment identifier.

Attributes:

Name	Type	Description
class_name	str	Experiment class name (e.g. ‘ClassName’)
instance_name	str	Experiment instance name (e.g. ‘Instance name’)
module_name	str	Module path (e.g. ‘experiment_library.experiments.exp_name’)

class_name instance-attribute

class_name: str

Experiment class name (e.g. ‘ClassName’)

instance_name instance-attribute

instance_name: str

Experiment instance name (e.g. ‘Instance name’)

module_name instance-attribute

module_name: str

Module path (e.g. ‘experiment_library.experiments.exp_name’)

__str__

__str__() -> str

Source code in src/icon/server/pre_processing/worker.py

def __str__(self) -> str:
    return f"{self.module_name}.{self.class_name}.{self.instance_name}"

from_str classmethod

from_str(identifier_str: str) -> Self

Parses an experiment identifier.

Returns: - the module path (e.g. ‘experiment_library.experiments.exp_name’) - the experiment class name (e.g. ‘ClassName’) - the experiment instance name (e.g. ‘Instance name’)

Example

“experiment_library.experiments.exp_name.ClassName (Instance name)” -> (“experiment_library.experiments.exp_name”, “ClassName”, “Instance name”)

Source code in src/icon/server/pre_processing/worker.py

@classmethod
def from_str(cls, identifier_str: str) -> Self:
    """Parses an experiment identifier.

    Returns:
    - the module path (e.g. 'experiment_library.experiments.exp_name')
    - the experiment class name (e.g. 'ClassName')
    - the experiment instance name (e.g. 'Instance name')

    Example:
        "experiment_library.experiments.exp_name.ClassName (Instance name)"
        -> ("experiment_library.experiments.exp_name", "ClassName", "Instance name")
    """
    match = re.match(r"^(.*)\.([^. ]+) \(([^)]+)\)$", identifier_str)
    if not match:
        raise ValueError(
            "Unexpected format of experiment identifier: ", identifier_str
        )
    return cls(match.group(1), match.group(2), match.group(3))

ParamUpdateMode

Bases: str, Enum

Attributes:

Name	Type	Description
ALL_FROM_TIMESTAMP
ALL_UP_TO_DATE
LOCALS_FROM_TS_GLOBALS_LATEST
ONLY_NEW_PARAMETERS

ALL_FROM_TIMESTAMP class-attribute instance-attribute

ALL_FROM_TIMESTAMP = 'all_from_timestamp'

ALL_UP_TO_DATE class-attribute instance-attribute

ALL_UP_TO_DATE = 'all_up_to_date'

LOCALS_FROM_TS_GLOBALS_LATEST class-attribute instance-attribute

LOCALS_FROM_TS_GLOBALS_LATEST = 'locals_ts_globals_now'

ONLY_NEW_PARAMETERS class-attribute instance-attribute

ONLY_NEW_PARAMETERS = 'only_new_parameters'

PreProcessingWorker

PreProcessingWorker(
    worker_number: int,
    pre_processing_queue: PriorityQueue[PreProcessingTask],
    update_queue: Queue[UpdateQueue],
    hardware_processing_queue: PriorityQueue[
        HardwareProcessingTask
    ],
    manager: SharedResourceManager,
    experiment_library_client: ExperimentLibraryClient,
)

Bases: Process

Methods:

Name	Description
run

Source code in src/icon/server/pre_processing/worker.py

def __init__(
    self,
    worker_number: int,
    pre_processing_queue: queue.PriorityQueue[PreProcessingTask],
    update_queue: multiprocessing.Queue[UpdateQueue],
    hardware_processing_queue: queue.PriorityQueue[HardwareProcessingTask],
    manager: SharedResourceManager,
    experiment_library_client: ExperimentLibraryClient,
) -> None:
    super().__init__()
    self._queue = pre_processing_queue
    self._update_queue = update_queue
    self._hw_processing_queue = hardware_processing_queue
    self._worker_number = worker_number
    self._manager = manager
    # Queues to communicate with the hardware worker:
    self._data_points_to_process: queue.Queue[
        tuple[int, dict[str, DatabaseValueType]]
    ]
    self._processed_data_points: queue.Queue[HardwareProcessingTask]
    self._parameter_dict: dict[str, DatabaseValueType] = {}
    self._outdated_tasks: queue.PriorityQueue[HardwareProcessingTask] = (
        manager.PriorityQueue()
    )
    self._experiment_library_client = experiment_library_client

run

run() -> None

Source code in src/icon/server/pre_processing/worker.py

@handle_keyboard_interrupt(logger)
def run(self) -> None:
    with self._experiment_library_client.isolated() as isolated_lib_client:
        logger.debug(
            "Created isolated experiment library client: %s",
            isolated_lib_client.checkout_revision(None),
        )

        while True:
            pre_processing_task = self._queue.get()

            self._data_points_to_process = self._manager.Queue()
            self._processed_data_points = self._manager.Queue()

            try:
                self._process_task(
                    pre_processing_task, isolated_lib_client=isolated_lib_client
                )

                logger.info(
                    "JobRun with id '%s' finished", pre_processing_task.job_run.id
                )

                if JobRunRepository.get_run_by_job_id(
                    job_id=pre_processing_task.job.id
                ).status in (JobRunStatus.PROCESSING, JobRunStatus.PAUSED):
                    JobRunRepository.update_run_by_id(
                        run_id=pre_processing_task.job_run.id,
                        status=JobRunStatus.DONE,
                    )

                try_auto_fit(
                    job_id=pre_processing_task.job.id,
                    experiment_source_id=pre_processing_task.job.experiment_source_id,
                )
            except Exception as e:
                logger.exception(
                    "JobRun with id '%s' failed", pre_processing_task.job_run.id
                )

                if JobRunRepository.get_run_by_job_id(
                    job_id=pre_processing_task.job.id
                ).status in (JobRunStatus.PROCESSING, JobRunStatus.PAUSED):
                    JobRunRepository.update_run_by_id(
                        run_id=pre_processing_task.job_run.id,
                        status=JobRunStatus.FAILED,
                        log=str(e),
                    )
            finally:
                JobRepository.update_job_status(
                    job=pre_processing_task.job, status=JobStatus.PROCESSED
                )

change_process_priority

change_process_priority(priority: int) -> None

Changes process priority.

Only superusers can decrease the niceness of a process.

Source code in src/icon/server/pre_processing/worker.py

def change_process_priority(priority: int) -> None:
    """Changes process priority.

    Only superusers can decrease the niceness of a process.
    """
    if os.getuid() == 0:
        p = psutil.Process(os.getpid())

        p.nice(priority)

consume_queue

consume_queue(q: Queue[T] | Queue[T]) -> Iterator[T]

Source code in src/icon/server/pre_processing/worker.py

def consume_queue(q: multiprocessing.Queue[T] | queue.Queue[T]) -> Iterator[T]:
    while True:
        try:
            yield q.get(block=False)
        except queue.Empty:
            return

create_hardware_instructions

create_hardware_instructions(
    client: ExperimentLibraryClient,
    n_shots: int,
    parameter_dict: dict[str, DatabaseValueType],
    namespace: ExperimentIdentifier,
) -> str

Source code in src/icon/server/pre_processing/worker.py

def create_hardware_instructions(
    client: ExperimentLibraryClient,
    n_shots: int,
    parameter_dict: dict[str, DatabaseValueType],
    namespace: ExperimentIdentifier,
) -> str:
    return asyncio.run(
        client.create_hardware_instructions(
            n_shots=n_shots,
            parameter_dict=parameter_dict,
            exp_module_name=namespace.module_name,
            exp_instance_name=namespace.instance_name,
        )
    )

freeze_dict

freeze_dict(
    combination: dict[str, DatabaseValueType],
) -> ScanCombination

Source code in src/icon/server/pre_processing/worker.py

def freeze_dict(combination: dict[str, DatabaseValueType]) -> ScanCombination:
    return frozenset(combination.items())

get_scan_combinations

get_scan_combinations(
    job: Job,
) -> list[dict[str, DatabaseValueType]]

Generates all combinations of scan parameters for a given job.

Repeats each combination job.repetitions times.

Parameters:

Name	Type	Description	Default
job	Job	The job containing scan parameters.	required

Returns:

Type	Description
list[dict[str, DatabaseValueType]]	A list of dictionaries, where each dictionary represents a combination of
list[dict[str, DatabaseValueType]]	parameter values.

Source code in src/icon/server/pre_processing/worker.py

def get_scan_combinations(job: Job) -> list[dict[str, DatabaseValueType]]:
    """Generates all combinations of scan parameters for a given job.

    Repeats each combination `job.repetitions` times.

    Args:
        job:
            The job containing scan parameters.

    Returns:
        A list of dictionaries, where each dictionary represents a combination of
        parameter values.
    """
    # Extract variable IDs and their scan values from the job's scan parameters
    parameter_values = {
        scan_param.unique_id(): scan_param.scan_values
        for scan_param in job.scan_parameters
        if not scan_param.realtime
    }

    if not parameter_values:
        return [{}] * job.repetitions

    # Generate combinations using itertools.product
    keys, values = zip(*parameter_values.items(), strict=True)

    combinations = itertools.product(*values)

    # Map each combination back to variable IDs
    return [
        dict(zip(keys, combination, strict=True)) for combination in combinations
    ] * job.repetitions

parse_experiment_identifier

parse_experiment_identifier(
    identifier: str,
) -> tuple[str, str, str]

Parses an experiment identifier.

Returns: - the module path (e.g. ‘experiment_library.experiments.exp_name’) - the experiment class name (e.g. ‘ClassName’) - the experiment instance name (e.g. ‘Instance name’)

Example

“experiment_library.experiments.exp_name.ClassName (Instance name)” -> (“experiment_library.experiments.exp_name”, “ClassName”, “Instance name”)

Source code in src/icon/server/pre_processing/worker.py

def parse_experiment_identifier(identifier: str) -> tuple[str, str, str]:
    """Parses an experiment identifier.

    Returns:
    - the module path (e.g. 'experiment_library.experiments.exp_name')
    - the experiment class name (e.g. 'ClassName')
    - the experiment instance name (e.g. 'Instance name')

    Example:
        "experiment_library.experiments.exp_name.ClassName (Instance name)"
        -> ("experiment_library.experiments.exp_name", "ClassName", "Instance name")
    """
    match = re.match(r"^(.*)\.([^. ]+) \(([^)]+)\)$", identifier)
    if match:
        return match.group(1), match.group(2), match.group(3)
    raise ValueError("Unexpected format of experiment identifier: ", identifier)

icon.server.scheduler

Modules:

Name	Description
scheduler

scheduler

Classes:

Name	Description
Scheduler

Functions:

Name	Description
initialise_job_tables
should_exit

Attributes:

Name	Type	Description
logger

logger module-attribute

logger = getLogger(__name__)

Scheduler

Scheduler(
    pre_processing_queue: PriorityQueue[PreProcessingTask],
    **kwargs: Any,
)

Bases: Process

Methods:

Name	Description
run

Attributes:

Name	Type	Description
kwargs

Source code in src/icon/server/scheduler/scheduler.py

def __init__(
    self,
    pre_processing_queue: queue.PriorityQueue[PreProcessingTask],
    **kwargs: Any,
) -> None:
    super().__init__()
    self.kwargs = kwargs
    self._pre_processing_queue = pre_processing_queue

kwargs instance-attribute

kwargs = kwargs

run

run() -> None

Source code in src/icon/server/scheduler/scheduler.py

@handle_keyboard_interrupt(logger)
def run(self) -> None:
    initialise_job_tables()
    while not should_exit():
        try:
            jobs = JobRepository.get_jobs_by_status_and_timeframe(
                status=JobStatus.SUBMITTED
            )
            for job_ in jobs:
                try:
                    job = JobRepository.update_job_status(
                        job=job_, status=JobStatus.PROCESSING
                    )
                    timestamp = now()
                    timezone = timestamp.tzinfo
                    run = JobRun(job_id=job.id, scheduled_time=timestamp)
                    run = JobRunRepository.insert_run(run=run)

                    self._pre_processing_queue.put(
                        PreProcessingTask(
                            job=job,
                            job_run=run,
                            git_commit_hash=job.git_commit_hash,
                            scan_parameters=job.scan_parameters,
                            local_parameters_timestamp=job.local_parameters_timestamp.astimezone(
                                tz=timezone
                            ).isoformat(),
                            priority=job.priority,
                            auto_calibration=job.auto_calibration,
                            debug_mode=job.debug_mode,
                            repetitions=job.repetitions,
                        )
                    )
                except Exception:
                    logger.exception(
                        "Failed to dispatch job %s, reverting to SUBMITTED",
                        job_.id,
                    )
                    try:
                        JobRepository.update_job_status(
                            job=job_, status=JobStatus.SUBMITTED
                        )
                    except Exception:
                        logger.exception(
                            "Failed to revert job %s back to SUBMITTED", job_.id
                        )
        except Exception:
            logger.exception("Unexpected error in scheduler loop")
        time.sleep(0.1)

initialise_job_tables

initialise_job_tables() -> None

Source code in src/icon/server/scheduler/scheduler.py

def initialise_job_tables() -> None:
    # update job_runs table
    job_runs = JobRunRepository.get_runs_by_status(
        status=[
            JobRunStatus.PENDING,
            JobRunStatus.PROCESSING,
            JobRunStatus.PAUSED,
        ]
    )
    for job_run in job_runs:
        JobRunRepository.update_run_by_id(
            run_id=job_run.id,
            status=JobRunStatus.CANCELLED,
            log="Cancelled during scheduler initialization.",
        )

    # update jobs table
    jobs = JobRepository.get_jobs_by_status_and_timeframe(status=JobStatus.PROCESSING)
    for job in jobs:
        logger.warning(
            "Job '%s' was left in PROCESSING state and is being marked as PROCESSED "
            "during scheduler initialization (likely abandoned due to a server restart).",
            job.id,
        )
        JobRepository.update_job_status(job=job, status=JobStatus.PROCESSED)

should_exit

should_exit() -> bool

Source code in src/icon/server/scheduler/scheduler.py

def should_exit() -> bool:
    return False

icon.server.utils.types

Classes:

Name	Description
UpdateQueue

UpdateQueue

Bases: TypedDict

Attributes:

Name	Type	Description
event	Literal['update_parameters', 'calibration']
job_id	NotRequired[int | None]
new_parameters	NotRequired[dict[str, DatabaseValueType]]

event instance-attribute

event: Literal['update_parameters', 'calibration']

job_id instance-attribute

job_id: NotRequired[int | None]

new_parameters instance-attribute

new_parameters: NotRequired[dict[str, DatabaseValueType]]

icon.server.web_server

Modules:

Name	Description
icon_server
sio_setup
socketio_emit_queue

icon_server

Classes:

Name	Description
IconServer

Attributes:

Name	Type	Description
logger

logger module-attribute

logger = getLogger(__name__)

IconServer

Bases: Server

Methods:

Name	Description
post_startup

post_startup async

post_startup() -> None

Source code in src/icon/server/web_server/icon_server.py

async def post_startup(self) -> None:
    sio = self._web_server._sio

    async def emit_worker() -> None:
        while not self.should_exit:
            try:
                emit_event = await asyncio.to_thread(emit_queue.get, timeout=1.0)
            except queue.Empty:
                continue
            await sio.emit(
                event=emit_event["event"],
                data=emit_event.get("data", None),
                room=emit_event.get("room", None),
            )

    asyncio.create_task(emit_worker())

    def devices_callback(
        full_access_path: str, value: Any, cached_value_dict: SerializedObject
    ) -> None:
        """This callback handles structural changes of devices.

        If the structure of
        a device changes, it will re-calculate the scannable parameters and emit
        them to the interested clients.
        """
        if full_access_path.startswith("devices.device_proxies"):
            emit_scannable_device_params_change(
                self._observer, full_access_path, value, cached_value_dict
            )

    self._observer.add_notification_callback(devices_callback)

sio_setup

Classes:

Name	Description
AsyncServer

Functions:

Name	Description
log_id
patch_sio_setup
setup_sio_events

Attributes:

Name	Type	Description
logger
pydase_setup_sio_events

logger module-attribute

logger = getLogger(__name__)

pydase_setup_sio_events module-attribute

pydase_setup_sio_events = setup_sio_events

AsyncServer

Bases: AsyncServer

Attributes:

Name	Type	Description
controlling_sid	str | None	Socketio SID of the client controlling the frontend.

controlling_sid class-attribute instance-attribute

controlling_sid: str | None = None

Socketio SID of the client controlling the frontend.

log_id

log_id(headers: Any, sid: str) -> str

Source code in src/icon/server/web_server/sio_setup.py

def log_id(headers: Any, sid: str) -> str:
    client_id_header = headers.get("HTTP_X_CLIENT_ID", None)
    remote_username_header = headers.get("HTTP_REMOTE_USER", None)

    if remote_username_header is not None:
        return f"user={click.style(remote_username_header, fg='cyan')}"
    if client_id_header is not None:
        return f"id={click.style(client_id_header, fg='cyan')}"
    return f"sid={click.style(sid, fg='cyan')}"

patch_sio_setup

patch_sio_setup() -> None

Source code in src/icon/server/web_server/sio_setup.py

def patch_sio_setup() -> None:
    import pydase.server.web_server.sio_setup  # noqa: PLC0415

    pydase.server.web_server.sio_setup.setup_sio_events = setup_sio_events

setup_sio_events

setup_sio_events(
    sio: AsyncServer, state_manager: StateManager
) -> None

Source code in src/icon/server/web_server/sio_setup.py

def setup_sio_events(
    sio: AsyncServer,
    state_manager: pydase.data_service.state_manager.StateManager,
) -> None:
    pydase_setup_sio_events(sio, state_manager)

    sio.controlling_sid = None

    @sio.event
    async def connect(sid: str, environ: Any) -> None:
        # send current controlling state to the newly connected client
        await sio.emit(
            "control_state", {"controlling_sid": sio.controlling_sid}, to=sid
        )

        async with sio.session(sid) as session:
            session["client_id"] = log_id(environ, sid)
            logger.info("Client [%s] connected", session["client_id"])

    @sio.event
    async def disconnect(sid: str) -> None:
        if sid == sio.controlling_sid:
            sio.controlling_sid = None
            await sio.emit("control_state", {"controlling_sid": None})

        async with sio.session(sid) as session:
            logger.info("Client [%s] disconnected", session["client_id"])

    @sio.event
    async def take_control(sid: str) -> None:
        sio.controlling_sid = sid
        await sio.emit("control_state", {"controlling_sid": sio.controlling_sid})

    @sio.event
    async def release_control(sid: str) -> None:
        if sio.controlling_sid == sid:
            sio.controlling_sid = None
            await sio.emit("control_state", {"controlling_sid": None})

socketio_emit_queue

Classes:

Name	Description
EmitEvent

Attributes:

Name	Type	Description
emit_queue	Queue[EmitEvent]

emit_queue module-attribute

emit_queue: Queue[EmitEvent] = Queue()

EmitEvent

Bases: TypedDict

Attributes:

Name	Type	Description
data	Any
event	str
room	NotRequired[str]

data instance-attribute

data: Any

event instance-attribute

event: str

room instance-attribute

room: NotRequired[str]