Skip to content

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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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=self.devices,
    parameters_controller=self.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 
21
22
23
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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 
43
44
45
46
47
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
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.
    """
    timeout = get_config().devices.set_value_timeout_seconds
    try:
        return await asyncio.to_thread(
            client_call_with_timeout,
            client=self._devices[name],
            event="get_value",
            data=parameter_id,
            timeout=timeout,
        )
    except BadNamespaceError:
        logger.warning(
            'Could not get %r. Device %r at ("%s") is not connected.',
            parameter_id,
            name,
            self._devices[name]._url,
        )
    except socketio.exceptions.TimeoutError:
        logger.warning(
            "Timed out after %s s while getting %r of device %r.",
            timeout,
            parameter_id,
            name,
        )
    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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
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

    timeout = get_config().devices.set_value_timeout_seconds
    try:
        await asyncio.to_thread(
            client_call_with_timeout,
            client=self._devices[name],
            event="update_value",
            data={"access_path": parameter_id, "value": dump(new_value)},
            timeout=timeout,
        )
    except BadNamespaceError:
        logger.warning(
            'Could not set %r. Device %r at ("%s") is not connected.',
            parameter_id,
            name,
            self._devices[name]._url,
        )
    except socketio.exceptions.TimeoutError:
        logger.warning(
            "Timed out after %s s while setting %r of device %r.",
            timeout,
            parameter_id,
            name,
        )
    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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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 = DEFAULT_MAX_TRANSFER_BYTES,
    *,
    include_hardware_instructions: bool = False,
    include_all_shots: bool = False,
) -> str

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 DEFAULT_MAX_TRANSFER_BYTES.

 DEFAULT_MAX_TRANSFER_BYTES
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
include_all_shots bool

If True, include the raw shots of every data point. Defaults to False, which returns only the newest data point’s shots because one array per point per channel dominates the payload; live updates still carry the shots of each new data point.

 False

Returns:

Type Description
str

The experiment data linked to the job as a JSON string representing an
str

[ExperimentData][icon.server.data_access.repositories.experiment_data_repository.ExperimentData]
str

instance.
Source code in src/icon/server/api/experiment_data_controller.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def get_experiment_data_by_job_id(
    self,
    job_id: int,
    max_transfer_bytes: int = DEFAULT_MAX_TRANSFER_BYTES,
    *,
    include_hardware_instructions: bool = False,
    include_all_shots: bool = False,
) -> str:
    """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
            ``DEFAULT_MAX_TRANSFER_BYTES``.
        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.
        include_all_shots: If True, include the raw shots of every data
            point.  Defaults to False, which returns only the newest data
            point's shots because one array per point per channel dominates
            the payload; live updates still carry the shots of each new
            data point.

    Returns:
        The experiment data linked to the job as a JSON string representing 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,
        include_all_shots=include_all_shots,
    )
    # TODO: workaround for avoiding the costly serialization which stalls the event loop for large objects.
    #   Packing as JSON string makes it opaque to the serializer.
    return json.dumps(asdict(result))
get_hardware_instructions async 
get_hardware_instructions(
    job_id: int | None = None, index: int | None = None
) -> str | None

Return stored hardware instructions (the serialized sequence JSON).

Used by the sequence visualizer to display the pulse sequence of a specific data point, a job, or the most recent experiment run.

Parameters:

Name Type Description Default
job_id int | None

Job to read from. Defaults to the most recent job with stored hardware instructions.

 None
index int | None

Data point index within the job. Defaults to the last stored entry.

 None

Returns:

Type Description
str | None

The serialized hardware instructions, or None when nothing is
str | None

stored for the requested scope.
Source code in src/icon/server/api/experiment_data_controller.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
async def get_hardware_instructions(
    self,
    job_id: int | None = None,
    index: int | None = None,
) -> str | None:
    """Return stored hardware instructions (the serialized sequence JSON).

    Used by the sequence visualizer to display the pulse sequence of a
    specific data point, a job, or the most recent experiment run.

    Args:
        job_id: Job to read from. Defaults to the most recent job with
            stored hardware instructions.
        index: Data point index within the job. Defaults to the last
            stored entry.

    Returns:
        The serialized hardware instructions, or None when nothing is
        stored for the requested scope.
    """
    return await asyncio.to_thread(
        ExperimentDataRepository.get_hardware_instructions,
        job_id=job_id,
        index=index,
    )
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 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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 
24
25
26
27
def __init__(self) -> None:
    super().__init__()
    self._experiments: ExperimentDict = {}
    self.hardware_description: str = ""
get_experiments 
get_experiments() -> dict[str, dict[str, Any]]

Return the current experiment metadata.

Returns:

Type Description
dict[str, dict[str, Any]]

Mapping of experiment IDs to their metadata.
Source code in src/icon/server/api/experiments_controller.py 
29
30
31
32
33
34
35
def get_experiments(self) -> dict[str, dict[str, Any]]:
    """Return the current experiment metadata.

    Returns:
        Mapping of experiment IDs to their metadata.
    """
    return {key: vars(val) for key, val in self._experiments.items()}
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 
68
69
70
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) -> dict[str, Any]

Serve experiment metadata for experiment id experiment_id.

Source code in src/icon/server/api/experiments_controller.py 
37
38
39
def get_metadata(self, experiment_id: str) -> dict[str, Any]:
    """Serve experiment metadata for experiment id `experiment_id`."""
    return vars(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 
56
57
58
59
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 
81
82
83
84
85
86
87
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 
89
90
91
92
93
94
95
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 
61
62
63
64
65
66
67
68
69
70
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
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 
72
73
74
75
76
77
78
79
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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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 
23
24
25
26
27
28
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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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

JOB_LIST_PAGE_SIZE module-attribute 

JOB_LIST_PAGE_SIZE = 100

Default number of jobs returned in a get_job_list page.

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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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 job.

The following status updates are performed:

  • Job: → PROCESSED
  • JobRun: PENDING/PROCESSING/PAUSED → 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 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def cancel_job(self, *, job_id: int) -> None:
    """Cancel a job.

    The following status updates are performed:

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

    Args:
        job_id: ID of the job to cancel.
    """
    job_transactions.cancel_job(
        job_id=job_id, log="Cancelled through user interaction."
    )
get_active_jobs 
get_active_jobs(
    *, limit: int = JOB_LIST_PAGE_SIZE
) -> list[JobListItemDict]

Return the queued and currently running jobs, newest first.

Parameters:

Name Type Description Default
limit int

Maximum number of jobs to return.

 JOB_LIST_PAGE_SIZE

Returns:

Type Description
list[JobListItemDict]

List of job-list entries ordered by descending job ID.
Source code in src/icon/server/api/scheduler_controller.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def get_active_jobs(
    self, *, limit: int = JOB_LIST_PAGE_SIZE
) -> list[JobListItemDict]:
    """Return the queued and currently running jobs, newest first.

    Args:
        limit: Maximum number of jobs to return.

    Returns:
        List of job-list entries ordered by descending job ID.
    """
    return JobRepository.get_job_list(
        statuses=[JobStatus.PROCESSING, JobStatus.SUBMITTED], limit=limit
    )
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 
340
341
342
343
344
345
346
347
348
349
350
351
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_list 
get_job_list(
    *,
    before_id: int | None = None,
    limit: int = JOB_LIST_PAGE_SIZE,
) -> list[JobListItemDict]

Return one page of finished jobs, newest first.

Only the fields the job list renders are returned, which keeps a page roughly 5x smaller on the wire than the equivalent get_scheduled_jobs response.

Parameters:

Name Type Description Default
before_id int | None

Exclusive upper bound on the job ID. Pass the lowest ID of the previous page to fetch the next one; omit it for the first page.

 None
limit int

Maximum number of jobs to return.

 JOB_LIST_PAGE_SIZE

Returns:

Type Description
list[JobListItemDict]

List of job-list entries ordered by descending job ID. A result
list[JobListItemDict]

shorter than limit means the end of the list has been reached.
Source code in src/icon/server/api/scheduler_controller.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
def get_job_list(
    self,
    *,
    before_id: int | None = None,
    limit: int = JOB_LIST_PAGE_SIZE,
) -> list[JobListItemDict]:
    """Return one page of finished jobs, newest first.

    Only the fields the job list renders are returned, which keeps a page
    roughly 5x smaller on the wire than the equivalent `get_scheduled_jobs`
    response.

    Args:
        before_id: Exclusive upper bound on the job ID. Pass the lowest ID of
            the previous page to fetch the next one; omit it for the first page.
        limit: Maximum number of jobs to return.

    Returns:
        List of job-list entries ordered by descending job ID. A result
        shorter than `limit` means the end of the list has been reached.
    """
    return JobRepository.get_job_list(
        statuses=[JobStatus.PROCESSED], before_id=before_id, limit=limit
    )
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 
353
354
355
356
357
358
359
360
361
362
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 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
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)
    JobRunRepository.update_run_by_id(
        run_id=job_run.id,
        status=JobRunStatus.PAUSED,
        log="Paused through user interaction.",
        only_if_status=(JobRunStatus.PROCESSING,),
    )
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 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
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)
    JobRunRepository.update_run_by_id(
        run_id=job_run.id,
        status=JobRunStatus.PROCESSING,
        only_if_status=(JobRunStatus.PAUSED,),
    )
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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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
    ]
    self._check_scan_values_within_bounds(scan_parameters=concretized_params)
    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 
32
33
34
35
36
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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.retry_disconnected)

    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 
52
53
54
55
56
57
58
59
60
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 
38
39
40
41
42
43
44
45
46
47
48
49
50
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.datetime: sqlalchemy.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] = sqlalchemy.orm.mapped_column(
    default=now
)

Timestamp when the device entry was created.

description class-attribute instance-attribute 

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

Optional human-readable description of the device.

id class-attribute instance-attribute 

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

Primary key identifier for the device.

name class-attribute instance-attribute 

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

Unique name of the device.

retry_attempts class-attribute instance-attribute 

retry_attempts: Mapped[int] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.mapped_column(
        default=0.0, nullable=False
    )
)

Delay in seconds between retry attempts

scan_parameters class-attribute instance-attribute 

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

Relationship to scan parameters linked to this device.

status class-attribute instance-attribute 

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

Current status of the device (enabled or disabled).

url class-attribute instance-attribute 

url: Mapped[str] = sqlalchemy.orm.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] = sqlalchemy.orm.mapped_column()

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

id class-attribute instance-attribute 

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

Primary key identifier for the experiment source.

jobs class-attribute instance-attribute 

jobs: Mapped[list[Job]] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.mapped_column(default=False)
)

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

created class-attribute instance-attribute 

created: Mapped[datetime] = sqlalchemy.orm.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] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.relationship(back_populates="jobs")
)

Relationship to the experiment source.

experiment_source_id class-attribute instance-attribute 

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

Foreign key referencing the associated experiment source.

git_commit_hash class-attribute instance-attribute 

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

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

id class-attribute instance-attribute 

id: Mapped[int] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.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] = sqlalchemy.orm.mapped_column(
    default=50
)

Number of shots per repetition.

parent_job class-attribute instance-attribute 

parent_job: Mapped[Job | None] = (
    sqlalchemy.orm.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] = (
    sqlalchemy.orm.mapped_column(
        sqlalchemy.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] = sqlalchemy.orm.mapped_column(
    default=20
)

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

repetitions class-attribute instance-attribute 

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

Number of times the experiment should be repeated.

resubmitted_jobs class-attribute instance-attribute 

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

List of jobs resubmitted from this job.

run class-attribute instance-attribute 

run: Mapped[JobRun] = sqlalchemy.orm.relationship(
    back_populates="job"
)

Relationship to the job run associated with this job.

scan_parameters class-attribute instance-attribute 

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

List of scan parameters associated with this job.

status class-attribute instance-attribute 

status: Mapped[JobStatus] = sqlalchemy.orm.mapped_column(
    default=JobStatus.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] = sqlalchemy.orm.mapped_column(
    primary_key=True, autoincrement=True
)

Primary key identifier for the job run.

job class-attribute instance-attribute 

job: Mapped[Job] = sqlalchemy.orm.relationship(
    back_populates="run"
)

Relationship to the job associated with this run.

job_id class-attribute instance-attribute 

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

Foreign key referencing the job being executed.

log class-attribute instance-attribute 

log: Mapped[str | None] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.mapped_column(default=None)
)

Timestamp of the last parameter update.

scheduled_time class-attribute instance-attribute 

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

Time when the run was scheduled to start.

status class-attribute instance-attribute 

status: Mapped[JobRunStatus] = sqlalchemy.orm.mapped_column(
    default=JobRunStatus.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] = sqlalchemy.orm.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] = (
    sqlalchemy.orm.mapped_column(
        sqlalchemy.ForeignKey("devices.id"), nullable=True
    )
)

Foreign key referencing the associated device, if any.

id class-attribute instance-attribute 

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

Primary key identifier for the scan parameter.

job class-attribute instance-attribute 

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

Relationship to the job.

job_id class-attribute instance-attribute 

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

Foreign key referencing the job this parameter belongs to.

name class-attribute instance-attribute 

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

Human-friendly display name persisted at submission time.

scan_values class-attribute instance-attribute 

scan_values: Mapped[list[DatabaseValueType]] = (
    sqlalchemy.orm.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] = sqlalchemy.orm.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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@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 
190
191
192
193
194
195
196
197
198
199
@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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@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 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@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 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@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

DEFAULT_MAX_TRANSFER_BYTES module-attribute 

DEFAULT_MAX_TRANSFER_BYTES = 4000000

Approximate cap on the serialised payload of one data request.

MOST_RECENT_JOB_RUNS module-attribute 

MOST_RECENT_JOB_RUNS = 10

How many of the newest job runs to search when no job is specified.

ExperimentDataRepository

Repository for HDF5-based experiment data.

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

Initialize the data container for a new job by calling :meth:initialize_for_job_id. Initialization is required before any read/write operation is triggered.

get_experiment_data_by_job_id staticmethod 
get_experiment_data_by_job_id(
    *,
    job_id: int,
    max_transfer_bytes: int = DEFAULT_MAX_TRANSFER_BYTES,
    include_hardware_instructions: bool = False,
    include_all_shots: 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 4 MB.

 DEFAULT_MAX_TRANSFER_BYTES
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
include_all_shots bool

If True, return the raw shots of every data point. Defaults to False, which returns only the newest data point’s shots.

 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 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
@staticmethod
def get_experiment_data_by_job_id(
    *,
    job_id: int,
    max_transfer_bytes: int = DEFAULT_MAX_TRANSFER_BYTES,
    include_hardware_instructions: bool = False,
    include_all_shots: 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 4 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.
        include_all_shots: If True, return the raw shots of every data point.
            Defaults to False, which returns only the newest data point's
            shots.

    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, HDF5FileMode.READ_ONLY) as h5file:
        return load_experiment_data(
            h5file,
            max_transfer_bytes,
            include_hardware_instructions=include_hardware_instructions,
            include_all_shots=include_all_shots,
        )
get_hardware_instructions staticmethod 
get_hardware_instructions(
    *, job_id: int | None = None, index: int | None = None
) -> str | None

Return stored hardware instructions (the serialized sequence JSON).

Parameters:

Name Type Description Default
job_id int | None

Job to read from. Defaults to the most recent job with stored hardware instructions, looking no further back than the MOST_RECENT_JOB_RUNS most recent runs.

 None
index int | None

Data point index within the job. Defaults to the last stored entry. Instructions are stored deduplicated (one entry per change), so the entry active at index is returned.

 None

Returns:

Type Description
str | None

The serialized hardware instructions, or None when nothing is
str | None

stored for the requested scope.
Source code in src/icon/server/data_access/repositories/experiment_data_repository.py 
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
@staticmethod
def get_hardware_instructions(
    *,
    job_id: int | None = None,
    index: int | None = None,
) -> str | None:
    """Return stored hardware instructions (the serialized sequence JSON).

    Args:
        job_id: Job to read from. Defaults to the most recent job with
            stored hardware instructions, looking no further back than the
            ``MOST_RECENT_JOB_RUNS`` most recent runs.
        index: Data point index within the job. Defaults to the last stored
            entry. Instructions are stored deduplicated (one entry per
            change), so the entry active at *index* is returned.

    Returns:
        The serialized hardware instructions, or None when nothing is
        stored for the requested scope.
    """
    results_dir = Path(get_config().data.results_dir)
    if job_id is not None:
        try:
            paths = [results_dir / get_filename_by_job_id(job_id)]
        except NoResultFound:
            return None
    else:
        paths = _recent_result_paths(results_dir)

    for path in paths:
        if not path.is_file():
            continue
        instructions = _read_hardware_instructions(path, index=index)
        if instructions is not None:
            return instructions
    return None
initialize_for_job_id staticmethod 
initialize_for_job_id(*, job_id: int) -> None

Create the file.

Parameters:

Name Type Description Default
job_id int

Job identifier.

 required
Source code in src/icon/server/data_access/repositories/experiment_data_repository.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
@staticmethod
def initialize_for_job_id(*, job_id: int) -> None:
    """Create the file.

    Args:
        job_id: Job identifier.
    """
    filename = get_filename_by_job_id(job_id)
    h5_path = Path(get_config().data.results_dir) / filename
    with h5_open(
        h5_path,
        HDF5FileMode.CREATE_OR_FAIL,
        fs_strategy="page",
        fs_persist=True,
        fs_page_size=65536,
    ):
        pass
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 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
@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, HDF5FileMode.READ_WRITE_OR_FAIL) 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 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
@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, HDF5FileMode.READ_WRITE_OR_FAIL) 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.

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 
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
@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.

    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, HDF5FileMode.READ_WRITE_OR_FAIL) 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 = cast("h5py.Dataset", parameters_group[param_id])
                if _parameter_value_unchanged(ds, value):
                    continue

                index = ds.shape[0]
                if ds.chunks is None:
                    # ds is fixed-size. Replace it with resizeable copy of itself.
                    ds = _make_parameter_dataset_extensible(
                        parameters_group, param_id
                    )
                else:
                    resize_dataset(ds, next_index=index, axis=0)
            else:
                # create fixed sized dataset which gets replaced with resizeable dataset on demand.
                ds = parameters_group.create_dataset(
                    param_id,
                    shape=(1,),
                    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()
            },
        }
    )

HDF5FileMode

Bases: StrEnum

HDF5 File modes - see https://docs.h5py.org/en/stable/high/file.html#opening-creating-files.

CREATE_OR_FAIL class-attribute instance-attribute 
CREATE_OR_FAIL = 'w-'

Create file, fail if exists

CREATE_OR_TRUNCATE class-attribute instance-attribute 
CREATE_OR_TRUNCATE = 'w'

Create file, truncate if exists

READ_ONLY class-attribute instance-attribute 
READ_ONLY = 'r'

Read-only, file must exist (default)

READ_WRITE_OR_CREATE class-attribute instance-attribute 
READ_WRITE_OR_CREATE = 'a'

Read/write if exists, create otherwise

READ_WRITE_OR_FAIL class-attribute instance-attribute 
READ_WRITE_OR_FAIL = 'r+'

Read/write, fail if not exists

OSFileLockError

Bases: OSError

Raised when an HDF5 file is locked by another process.

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 
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
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, HDF5FileMode.READ_WRITE_OR_FAIL) 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 
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
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
    )

    total_vector_bytes = 0
    for channel_group in (vector_channels_group or {}).values():
        vectors = cast("h5py.Group", channel_group)
        sample_name = next(iter(vectors), None)
        if sample_name is None:
            continue
        sample = cast("h5py.Dataset", vectors[sample_name])
        total_vector_bytes += sample.shape[0] * sample.dtype.itemsize * len(vectors)
    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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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").
    """
    return _result_filename(
        JobRunRepository.get_scheduled_time_by_job_id(job_id=job_id)
    )

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 
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
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, HDF5FileMode.READ_ONLY) 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 
914
915
916
917
918
919
920
921
922
923
924
925
926
927
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)}")

get_result_channels_dataset 

get_result_channels_dataset(
    h5file: File,
    result_channels: list[str],
    number_of_data_points: int = 0,
) -> Dataset

Return the ‘result_channels’ dataset, creating it if it does not exist yet.

Source code in src/icon/server/data_access/repositories/experiment_data_repository.py 
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
def get_result_channels_dataset(
    h5file: h5py.File, result_channels: list[str], number_of_data_points: int = 0
) -> h5py.Dataset:
    """Return the 'result_channels' dataset, creating it if it does not exist yet."""
    sorted_result_channels = sorted(result_channels)
    result_dtype = np.dtype([(key, np.float64) for key in sorted_result_channels])

    return h5file.require_dataset(
        "result_channels",
        shape=(number_of_data_points,),
        maxshape=(None,),
        dtype=result_dtype,
        chunks=True,
        **_common_hdf5_dataset_params,
    )

h5_open 

h5_open(
    path: Path,
    mode: HDF5FileMode,
    *,
    timeout: float | None = None,
    **kwargs: Any,
) -> Generator[File]

Open an HDF5 file under a process-wide per-file lock.

Opens of the same file are serialised within this process by a per-file lock; different files proceed in parallel. The OS-native HDF5 lock may still be held by another process, in which case the open is retried with a capped backoff. Every other failure is permanent and raised immediately.

Parameters:

Name Type Description Default
path Path

Path of the HDF5 file.

 required
mode HDF5FileMode

Mode passed to h5py.File.

 required
timeout float | None

Seconds to wait in total, for the in-process lock and the file lock together. Defaults to data.h5_open_timeout_seconds from the configuration.

 None
kwargs Any

Additional arguments passed to h5py.File.

 {}

Yields:

Type Description
Generator[File]

The open h5py.File.

Raises:

Type Description
TimeoutError

The file could not be opened within timeout due to locking.
OSError

The file could not be opened for any other reason.
Source code in src/icon/server/data_access/repositories/experiment_data_repository.py 
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
@contextmanager
def h5_open(
    path: Path, mode: HDF5FileMode, *, timeout: float | None = None, **kwargs: Any
) -> Generator[h5py.File]:
    """Open an HDF5 file under a process-wide per-file lock.

    Opens of the same file are serialised within this process by a per-file
    lock; different files proceed in parallel. The OS-native HDF5 lock may still
    be held by another process, in which case the open is retried with a capped
    backoff. Every other failure is permanent and raised immediately.

    Args:
        path: Path of the HDF5 file.
        mode: Mode passed to `h5py.File`.
        timeout: Seconds to wait in total, for the in-process lock and the file
            lock together. Defaults to `data.h5_open_timeout_seconds` from the
            configuration.
        kwargs: Additional arguments passed to `h5py.File`.

    Yields:
        The open `h5py.File`.

    Raises:
        TimeoutError: The file could not be opened within `timeout` due to locking.
        OSError: The file could not be opened for any other reason.
    """
    if timeout is None:
        timeout = get_config().data.h5_open_timeout_seconds
    deadline = time.monotonic() + timeout

    with (
        _in_process_lock(path, timeout=timeout),
        _h5_open_with_retry(path, mode, deadline=deadline, **kwargs) as h5file,
    ):
        yield h5file

load_experiment_data 

load_experiment_data(
    h5file: File,
    max_transfer_bytes: int = DEFAULT_MAX_TRANSFER_BYTES,
    *,
    include_hardware_instructions: bool = False,
    include_all_shots: 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 4 MB.

 DEFAULT_MAX_TRANSFER_BYTES
include_hardware_instructions bool

Whether to include hardware instructions.

 False
include_all_shots bool

If True, return the raw shots of every data point. Defaults to False, which returns only the newest data point’s shots.

 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 
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
def load_experiment_data(
    h5file: h5py.File,
    max_transfer_bytes: int = DEFAULT_MAX_TRANSFER_BYTES,
    *,
    include_hardware_instructions: bool = False,
    include_all_shots: 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 4 MB.
        include_hardware_instructions: Whether to include hardware instructions.
        include_all_shots: If True, return the raw shots of every data point.
            Defaults to False, which returns only the newest data point's
            shots.

    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 if include_all_shots else None,
        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 [])
        ]
        shot_start_index = (
            start_index if include_all_shots else max(start_index, total - 1)
        )
        data.readouts.shot_channels = {
            key: dict(
                enumerate(  # type: ignore[call-overload]
                    value[shot_start_index:].tolist(), start=shot_start_index
                )
            )
            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(name): cast("h5py.Dataset", vector_group[name])[:].tolist()
                for name in vector_group
                if int(name) >= start_index
            }
            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 
118
119
120
121
122
123
124
125
126
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 
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
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, HDF5FileMode.READ_WRITE_OR_FAIL) 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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
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,
        **_common_hdf5_dataset_params,
    )

    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 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
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 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
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,
        **_common_hdf5_dataset_params,
    )

    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 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
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),
            dtype=np.float64,
            chunks=True,
            **_common_hdf5_dataset_params,
        )

        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 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
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)
        # Don't create a dataset for empty vector data.
        if str(data_point_index) not in channel_group and vector:
            channel_group.create_dataset(
                str(data_point_index),
                data=vector,
                **_common_hdf5_dataset_params,
            )

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 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@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 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
@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 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
@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_job_list staticmethod 
get_job_list(
    *,
    statuses: Sequence[JobStatus] | None = None,
    before_id: int | None = None,
    limit: int | None = None,
) -> list[JobListItemDict]

List jobs as lightweight JobListItemDict items.

Pagination is cursor-based on Job.id.

Parameters:

Name Type Description Default
statuses Sequence[JobStatus] | None

Optional status filter; jobs matching any of the given statuses are returned.

 None
before_id int | None

Exclusive upper bound on the job ID; pass the lowest ID of the previous page to fetch the next one.

 None
limit int | None

Maximum number of jobs to return.

 None

Returns:

Type Description
list[JobListItemDict]

Job list entries ordered by descending job ID.
Source code in src/icon/server/data_access/repositories/job_repository.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
@staticmethod
def get_job_list(
    *,
    statuses: Sequence[JobStatus] | None = None,
    before_id: int | None = None,
    limit: int | None = None,
) -> list[JobListItemDict]:
    """List jobs as lightweight `JobListItemDict` items.

    Pagination is cursor-based on `Job.id`.

    Args:
        statuses: Optional status filter; jobs matching any of the given
            statuses are returned.
        before_id: Exclusive upper bound on the job ID; pass the lowest ID of
            the previous page to fetch the next one.
        limit: Maximum number of jobs to return.

    Returns:
        Job list entries ordered by descending job ID.
    """
    stmt = (
        select(
            Job.id,
            Job.created,
            Job.status,
            ExperimentSource.experiment_id,
        )
        .join(ExperimentSource, Job.experiment_source_id == ExperimentSource.id)
        .order_by(Job.id.desc())
    )

    if statuses is not None:
        stmt = stmt.where(Job.status.in_(statuses))
    if before_id is not None:
        stmt = stmt.where(Job.id < before_id)
    if limit is not None:
        stmt = stmt.limit(limit)

    with sqlalchemy.orm.Session(engine) as session:
        rows = session.execute(stmt).all()
        if not rows:
            return []

        job_ids = [row.id for row in rows]

        scan_parameter_counts: dict[int, int] = dict(
            session.execute(
                select(ScanParameter.job_id, func.count(ScanParameter.id))
                .where(ScanParameter.job_id.in_(job_ids))
                .group_by(ScanParameter.job_id)
            ).all()  # type: ignore[arg-type]
        )

        latest_runs: dict[int, tuple[int, JobRunStatus]] = {
            run.job_id: (run.id, run.status)
            for run in session.execute(
                select(JobRun.job_id, JobRun.id, JobRun.status)
                .where(JobRun.job_id.in_(job_ids))
                .order_by(JobRun.scheduled_time.asc())
            ).all()
        }

    return [
        _job_list_item(
            job_id=row.id,
            created=row.created,
            status=row.status,
            experiment_id=row.experiment_id,
            num_scan_parameters=scan_parameter_counts.get(row.id, 0),
            run=latest_runs.get(row.id),
        )
        for row in rows
    ]
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 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@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_id: int, status: JobStatus) -> Job

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

Parameters:

Name Type Description Default
job_id int

ID of the job to update.

 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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@staticmethod
def update_job_status(*, job_id: int, status: JobStatus) -> Job:
    """Update a job's status and emit an update event.

    Args:
        job_id: ID of the job to update.
        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 update and query job runs from the database. Runs are created by job_transactions.dispatch_job. Emits Socket.IO events when job runs are updated.

get_parameter_update_timestamp staticmethod 
get_parameter_update_timestamp(
    *, run_id: int
) -> datetime | None

Get the paramter update timestamp.

Parameters:

Name Type Description Default
run_id int

ID of the job.

 required

Returns:

Type Description
datetime | None

The parameter update timestamp, or None if no parameter update has
datetime | None

been recorded for this run yet.
Source code in src/icon/server/data_access/repositories/job_run_repository.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
@staticmethod
def get_parameter_update_timestamp(*, run_id: int) -> datetime | None:
    """Get the paramter update timestamp.

    Args:
        run_id: ID of the job.

    Returns:
        The parameter update timestamp, or None if no parameter update has
        been recorded for this run yet.
    """
    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)

    if timestamp is None:
        return None
    return timestamp.replace(tzinfo=UTC)
get_recent_scheduled_times staticmethod 
get_recent_scheduled_times(
    *, limit: int
) -> Sequence[datetime]

Return the scheduled times of the most recent runs, newest first.

Parameters:

Name Type Description Default
limit int

Maximum number of scheduled times to return.

 required

Returns:

Type Description
Sequence[datetime]

Scheduled times ordered from newest to oldest.
Source code in src/icon/server/data_access/repositories/job_run_repository.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@staticmethod
def get_recent_scheduled_times(*, limit: int) -> Sequence[datetime]:
    """Return the scheduled times of the most recent runs, newest first.

    Args:
        limit: Maximum number of scheduled times to return.

    Returns:
        Scheduled times ordered from newest to oldest.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = (
            select(JobRun.scheduled_time)
            .where(JobRun.status != JobRunStatus.PENDING)
            .order_by(JobRun.scheduled_time.desc())
            .limit(limit)
        )

        scheduled_times = session.execute(stmt).scalars().all()
        logger.debug("Got the %s most recent scheduled times", limit)
    return scheduled_times
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@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 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@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
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 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
@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,
    only_if_status: Sequence[JobRunStatus] | None = None,
) -> JobRun | None

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
only_if_status Sequence[JobRunStatus] | None

Apply the update only if current status is in the list.

 None

Returns:

Type Description
JobRun | None

The updated job run, or None when the current status is non of the
JobRun | None

only_if_status statuses.
Source code in src/icon/server/data_access/repositories/job_run_repository.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@staticmethod
def update_run_by_id(
    *,
    run_id: int,
    status: JobRunStatus,
    log: str | None = None,
    only_if_status: Sequence[JobRunStatus] | None = None,
) -> JobRun | None:
    """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).
        only_if_status: Apply the update only if current status is in the list.

    Returns:
        The updated job run, or None when the current status is non of the
        `only_if_status` statuses.
    """
    with sqlalchemy.orm.Session(engine) as session:
        stmt = update(JobRun).where(JobRun.id == run_id)
        if only_if_status is not None:
            stmt = stmt.where(JobRun.status.in_(only_if_status))

        run = session.execute(
            stmt.values(status=status, log=log).returning(JobRun)
        ).scalar_one_or_none()

        if run is None:
            if only_if_status is None:
                raise NoResultFound(f"No job run with id {run_id}")
            return None

        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 
36
37
38
39
40
41
42
43
44
45
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.
    """
    return run_cancelled_or_failed(JobRunRepository.get_run_by_job_id(job_id=job_id))

run_cancelled_or_failed 

run_cancelled_or_failed(job_run: JobRun) -> bool

Check if an already fetched run was cancelled or failed.

Parameters:

Name Type Description Default
job_run JobRun

The run to check.

 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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def run_cancelled_or_failed(job_run: JobRun) -> bool:
    """Check if an already fetched run was cancelled or failed.

    Args:
        job_run: The run to check.

    Returns:
        True if the run status is CANCELLED or FAILED, False otherwise.
    """
    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

try_update_run_by_id 

try_update_run_by_id(
    *,
    run_id: int,
    status: JobRunStatus,
    log: str | None = None,
    only_if_status: Sequence[JobRunStatus] | None = None,
) -> JobRun | None

Update a job run, logging any failure instead of raising.

Parameters:

Name Type Description Default
run_id int

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
only_if_status Sequence[JobRunStatus] | None

Apply the update only while the stored status is one of these; see JobRunRepository.update_run_by_id.

 None

Returns:

Type Description
JobRun | None

The updated job run, or None when the update failed or did not apply.
Source code in src/icon/server/data_access/repositories/job_run_repository.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def try_update_run_by_id(
    *,
    run_id: int,
    status: JobRunStatus,
    log: str | None = None,
    only_if_status: Sequence[JobRunStatus] | None = None,
) -> JobRun | None:
    """Update a job run, logging any failure instead of raising.

    Args:
        run_id: ID of the job run to update.
        status: New status of the run.
        log: Optional log message (e.g. failure reason).
        only_if_status: Apply the update only while the stored status is one of
            these; see `JobRunRepository.update_run_by_id`.

    Returns:
        The updated job run, or None when the update failed or did not apply.
    """
    try:
        return JobRunRepository.update_run_by_id(
            run_id=run_id, status=status, log=log, only_if_status=only_if_status
        )
    except Exception:
        logger.exception(
            "Failed to update run '%s' to status '%s'", run_id, status.value
        )
        return None

job_transactions

LIVE_RUN_STATUSES module-attribute 

LIVE_RUN_STATUSES = (
    JobRunStatus.PENDING,
    JobRunStatus.PROCESSING,
    JobRunStatus.PAUSED,
)

Run states in which a pre-processing worker still owns the job.

cancel_job 

cancel_job(
    *,
    job_id: int,
    log: str | None = None,
    run_status: JobRunStatus = CANCELLED,
) -> None

Cancel a job and move its run to run_status.

Parameters:

Name Type Description Default
job_id int

ID of the job to retire.

 required
run_status JobRunStatus

Terminal status to give the run.

 CANCELLED
log str | None

Reason recorded on the run, if there is one.

 None
Source code in src/icon/server/data_access/repositories/job_transactions.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def cancel_job(
    *,
    job_id: int,
    log: str | None = None,
    run_status: JobRunStatus = JobRunStatus.CANCELLED,
) -> None:
    """Cancel a job and move its run to `run_status`.

    Args:
        job_id: ID of the job to retire.
        run_status: Terminal status to give the run.
        log: Reason recorded on the run, if there is one.
    """
    with (
        sqlalchemy.orm.Session(engine, expire_on_commit=False) as session,
        session.begin(),
    ):
        updated_job_id = session.execute(
            update(Job)
            .where(Job.id == job_id)
            .values(status=JobStatus.PROCESSED)
            .returning(Job.id)
        ).scalar_one_or_none()

        run = session.execute(
            update(JobRun)
            .where(JobRun.id == _latest_run_id(job_id))
            .where(JobRun.status.in_(LIVE_RUN_STATUSES))
            .values(status=run_status, log=log)
            .returning(JobRun)
        ).scalar_one_or_none()

    logger.debug(
        "Cancelled job %s, run %s is %s",
        job_id,
        run.id if run is not None else None,
        run_status.value,
    )

    if updated_job_id is not None:
        emit_queue.put(
            {
                "event": "job.update",
                "data": {
                    "job_id": job_id,
                    "updated_properties": {"status": JobStatus.PROCESSED.value},
                },
            }
        )

    if run is not None:
        emit_queue.put(
            {
                "event": "job_run.update",
                "data": {
                    "run_id": run.id,
                    "updated_properties": {
                        "status": run_status.value,
                        "log": log,
                    },
                },
            }
        )

dispatch_job 

dispatch_job(*, job_id: int) -> tuple[Job, JobRun] | None

Insert a run for a SUBMITTED job and progress it to PROCESSING.

Parameters:

Name Type Description Default
job_id int

ID of the job to dispatch.

 required

Returns:

Type Description
tuple[Job, JobRun] | None

The claimed job with its relationships loaded and its new run, or None
tuple[Job, JobRun] | None

when the job is no longer SUBMITTED, i.e. cancelled or processing.
Source code in src/icon/server/data_access/repositories/job_transactions.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def dispatch_job(*, job_id: int) -> tuple[Job, JobRun] | None:
    """Insert a run for a SUBMITTED job and progress it to PROCESSING.

    Args:
        job_id: ID of the job to dispatch.

    Returns:
        The claimed job with its relationships loaded and its new run, or None
        when the job is no longer SUBMITTED, i.e. cancelled or processing.
    """
    with (
        sqlalchemy.orm.Session(engine, expire_on_commit=False) as session,
        session.begin(),
    ):
        claimed_id = session.execute(
            update(Job)
            .where(Job.id == job_id, Job.status == JobStatus.SUBMITTED)
            .values(status=JobStatus.PROCESSING)
            .returning(Job.id)
        ).scalar_one_or_none()

        if claimed_id is None:
            return None

        run = JobRun(job_id=job_id, scheduled_time=now())
        session.add(run)
        session.flush()

        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()
        )

    logger.debug("Dispatched job %s as run %s", job_id, run.id)

    emit_queue.put(
        {
            "event": "job.update",
            "data": {
                "job_id": job_id,
                "updated_properties": {"status": JobStatus.PROCESSING.value},
            },
        }
    )
    emit_queue.put(
        {
            "event": "job_run.new",
            "data": {"job_run": SQLAlchemyDictEncoder.encode(obj=run)},
        }
    )

    return job, run

fail_job 

fail_job(*, job_id: int, log: str | None = None) -> None

Cancel job and fail the run it may own.

Parameters:

Name Type Description Default
job_id int

ID of the job to fail.

 required
log str | None

Optional reason recorded on the run.

 None
Source code in src/icon/server/data_access/repositories/job_transactions.py 
107
108
109
110
111
112
113
114
def fail_job(*, job_id: int, log: str | None = None) -> None:
    """Cancel job and fail the run it may own.

    Args:
        job_id: ID of the job to fail.
        log: Optional reason recorded on the run.
    """
    cancel_job(job_id=job_id, run_status=JobRunStatus.FAILED, log=log)

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 classmethod 
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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@classmethod
def get_influxdb_parameter_by_id(
    cls, 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.
    """
    backend = cls._get_backend()
    value = backend.get_influxdb_parameter_by_id(parameter_id)
    if value is None:
        logger.error(
            "Could not find parameter with id %s in measurement %s",
            parameter_id,
            backend.measurement,
        )
        return None
    return value
get_influxdb_parameter_keys classmethod 
get_influxdb_parameter_keys() -> list[str]

Return all known parameter identifiers from InfluxDB.

Source code in src/icon/server/data_access/repositories/parameters_repository.py 
157
158
159
160
@classmethod
def get_influxdb_parameter_keys(cls) -> list[str]:
    """Return all known parameter identifiers from InfluxDB."""
    return cls._get_backend().get_influxdb_parameter_keys()
get_influxdb_parameters classmethod 
get_influxdb_parameters(
    *,
    before: str | None = None,
    namespace: str | None = None,
    measurement: 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
measurement str | None

Optional measurement override; defaults to the active backend’s measurement.

 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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@classmethod
def get_influxdb_parameters(
    cls,
    *,
    before: str | None = None,
    namespace: str | None = None,
    measurement: 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.
        measurement: Optional measurement override; defaults to the active backend's
            measurement.

    Returns:
        Mapping of parameter IDs to values.
    """
    return cls._get_backend().get_influxdb_parameters(
        before=before, namespace=namespace, measurement=measurement
    )
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 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@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 
146
147
148
149
150
151
152
153
154
155
@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 
50
51
52
53
54
55
56
57
58
59
60
@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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@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)

icon.server.hardware_processing

Modules:

Name Description
devices
hardware_controller
rpc
task
tiqizedboard_controller
utils
worker
zedboard_controller

devices

Classes:

Name Description
Devices
Hardware

Functions:

Name Description
load

Devices 

Devices()

Methods:

Name Description
__getitem__
items
main_device
reload
retry_disconnected
Source code in src/icon/server/hardware_processing/devices.py 
43
44
45
46
47
48
49
50
51
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=self.__devices,
    obj_factory=load,
    subconfig=lambda config: {
        dev["id"]: dev
        for dev in config.hardware.model_dump()["devices"]
    },
)
__getitem__ 
__getitem__(dev_id: str) -> Hardware
Source code in src/icon/server/hardware_processing/devices.py 
67
68
69
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 
71
72
73
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 
75
76
77
78
79
80
81
82
83
84
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(*, retry_disconnected: bool = False) -> None
Source code in src/icon/server/hardware_processing/devices.py 
53
54
55
56
57
58
59
60
61
62
def reload(self, *, retry_disconnected: bool = False) -> 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 (retry_disconnected and not dev.controller.connected)
        ):
            dev.controller.connect()
retry_disconnected 
retry_disconnected() -> None
Source code in src/icon/server/hardware_processing/devices.py 
64
65
def retry_disconnected(self) -> None:
    self.reload(retry_disconnected=True)

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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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 = logging.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 
40
41
def connect(self) -> None:
    pass
receive 
receive() -> Readouts
Source code in src/icon/server/hardware_processing/hardware_controller.py 
56
57
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 
50
51
def run(self) -> None:
    pass
send 
send(data: str) -> None
Source code in src/icon/server/hardware_processing/hardware_controller.py 
47
48
def send(self, data: str) -> None:
    pass
status 
status() -> tuple[StatusFlag, str, Any]
Source code in src/icon/server/hardware_processing/hardware_controller.py 
53
54
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 
17
18
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 
33
34
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 
27
28
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 
24
25
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 
30
31
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()

rpc

Modules:

Name Description
client

Blocking client for lock-step msgpack-rpc. Only one .call() can be in flight.
connection
errors

Exception hierarchy for the RPC client.
zedboard

client

Blocking client for lock-step msgpack-rpc. Only one .call() can be in flight.

REQUEST := [0, msgid_u32, method: str, params: array] RESPONSE := [1, msgid_u32, error, result] NOTIFICATION := [2, method: str, params]

Classes:

Name Description
MessageType

First element of every msgpack-rpc message array.
MsgPackRPCClient

Blocking msgpack-rpc client, with one request in flight at a time.
RPCNotification

A message the server pushed of its own accord, answering no request.
RPCResponse

A reply to one request, tied to it by :attr:msgid rather than by arrival order.

Attributes:

Name Type Description
DEFAULT_LOCK_ACQUSITION_TIME Final

Default time for a RPC round-trip lock to become available.
DEFAULT_NOTIFICATION_BUFFER Final

Notifications retained while nobody polls. Past this the oldest are dropped, which is
DEFAULT_NOTIFICATION_LIMIT Final

Default consumption batch size when calling :meth:MsgPackRPCClient.consume_notifications
logger
DEFAULT_LOCK_ACQUSITION_TIME module-attribute 
DEFAULT_LOCK_ACQUSITION_TIME: Final = 1.0

Default time for a RPC round-trip lock to become available.

DEFAULT_NOTIFICATION_BUFFER module-attribute 
DEFAULT_NOTIFICATION_BUFFER: Final = 1000

Notifications retained while nobody polls. Past this the oldest are dropped, which is reported once per poll rather than once per message.

DEFAULT_NOTIFICATION_LIMIT module-attribute 
DEFAULT_NOTIFICATION_LIMIT: Final = 100

Default consumption batch size when calling :meth:MsgPackRPCClient.consume_notifications

logger module-attribute 
logger = logging.getLogger(__name__)
MessageType

Bases: IntEnum

First element of every msgpack-rpc message array.

Attributes:

Name Type Description
NOTIFICATION
REQUEST
RESPONSE
NOTIFICATION class-attribute instance-attribute 
NOTIFICATION = 2
REQUEST class-attribute instance-attribute 
REQUEST = 0
RESPONSE class-attribute instance-attribute 
RESPONSE = 1
MsgPackRPCClient 
MsgPackRPCClient(
    hostname: str,
    port: int,
    timeout: float | None = None,
    *,
    framed: bool = True,
    lock_timeout: float = DEFAULT_LOCK_ACQUSITION_TIME,
    notification_buffer: int = DEFAULT_NOTIFICATION_BUFFER,
)

Blocking msgpack-rpc client, with one request in flight at a time.

Call :meth:connect to open the connection.

For example::

client = MsgPackRPCClient("localhost", 1234)
client.connect()
print(client.call("hello"))

Parameters:

Name Type Description Default
hostname str

Host to connect to.

 required
port int

TCP port to connect to.

 required
timeout float | None

Default timeout for a call, in seconds, counted from when the request is sent. None waits indefinitely.

 None
framed bool

Whether to use framed transport, where every message is preceeded with a 4-byte length header.

 True
lock_timeout float

How long a call waits for another thread’s round trip to finish.

 DEFAULT_LOCK_ACQUSITION_TIME
notification_buffer int

How many pushed notifications to retain in the buffer. Oldest are dropped first.

 DEFAULT_NOTIFICATION_BUFFER

Methods:

Name Description
__repr__
call

Invoke a blocking remote function call and return its result.
connect
consume_notifications

Return up to limit received notifications.
disconnect
notify

Send a one-way notification, for which the server sends no reply.

Attributes:

Name Type Description
is_connected bool
Source code in src/icon/server/hardware_processing/rpc/client.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def __init__(
    self,
    hostname: str,
    port: int,
    timeout: float | None = None,
    *,
    framed: bool = True,
    lock_timeout: float = DEFAULT_LOCK_ACQUSITION_TIME,
    notification_buffer: int = DEFAULT_NOTIFICATION_BUFFER,
) -> None:
    self._timeout = timeout
    self._msgid = 0
    self._notification_buffer = notification_buffer
    self._notifications: deque[tuple[float, Any]] = deque(
        maxlen=notification_buffer
    )
    self._dropped = 0

    self._connection = (FramedConnection if framed else Connection)(
        hostname, port, timeout=timeout, lock_timeout=lock_timeout
    )
is_connected property 
is_connected: bool
__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/client.py 
126
127
def __repr__(self) -> str:
    return f"MsgPackRPCClient[conn={self._connection}, timeout={self._timeout}, msgid={self._msgid}, pending_notify={len(self._notifications)}]"
call 
call(
    method_name: str,
    *args: Any,
    timeout: float | None = None,
) -> Any

Invoke a blocking remote function call and return its result.

Parameters:

Name Type Description Default
method_name str

The name of the RPC method to call.

 required
*args Any

Positional arguments passed to the RPC method.

 ()
timeout float | None

Maximum time to wait for a response, in seconds. Uses the client’s default timeout when omitted.

 None

Returns:

Type Description
Any

The result field of the server’s reply, unpacked.

Raises:

Type Description
RPCResponseError

If the server answers with an error.
ProtocolError

If the server sends something unreadable.
TimeoutError

If no response arrives within timeout.
ConnectionBusyError

If another thread holds the connection. Nothing was sent, so retrying is safe.
ConnectionError

If the connection is not open, or the peer closed it.
Source code in src/icon/server/hardware_processing/rpc/client.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def call(self, method_name: str, *args: Any, timeout: float | None = None) -> Any:
    """Invoke a blocking remote function call and return its result.

    Args:
        method_name: The name of the RPC method to call.
        *args: Positional arguments passed to the RPC method.
        timeout: Maximum time to wait for a response, in seconds. Uses the
            client's default timeout when omitted.

    Returns:
        The ``result`` field of the server's reply, unpacked.

    Raises:
        RPCResponseError: If the server answers with an error.
        ProtocolError: If the server sends something unreadable.
        TimeoutError: If no response arrives within ``timeout``.
        ConnectionBusyError: If another thread holds the connection. Nothing was
            sent, so retrying is safe.
        ConnectionError: If the connection is not open, or the peer closed it.
    """
    _timeout = self._timeout if timeout is None else timeout

    with self._connection.transaction(_timeout) as transaction:
        msgid = self._next_msgid
        logger.debug("< CALL (%d): %s%s", msgid, method_name, _Abbreviated(args))
        transaction.send((MessageType.REQUEST, msgid, method_name, args))

        while True:
            received = self._route(transaction.receive())
            if not isinstance(received, RPCResponse):
                continue  # a notification, already buffered
            if received.msgid != msgid:
                logger.warning(
                    "discarding response for msgid %d while awaiting %d",
                    received.msgid,
                    msgid,
                )
                continue
            if received.error is not None:
                raise RPCResponseError(msgid, received.error)
            return received.result
connect 
connect() -> None
Source code in src/icon/server/hardware_processing/rpc/client.py 
129
130
def connect(self) -> None:
    self._connection.connect()
consume_notifications 
consume_notifications(
    limit: int = DEFAULT_NOTIFICATION_LIMIT,
) -> list[tuple[float, Any]]

Return up to limit received notifications.

Reads the socket for more only if fewer than limit are already buffered, and only what has already arrived. A connection busy with a call is not an error here: that call is reading the same socket and buffers every notification it passes, so the poll gives up its turn and returns what it has.

Source code in src/icon/server/hardware_processing/rpc/client.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def consume_notifications(
    self, limit: int = DEFAULT_NOTIFICATION_LIMIT
) -> list[tuple[float, Any]]:
    """Return up to ``limit`` received notifications.

    Reads the socket for more only if fewer than ``limit`` are already buffered, and
    only what has already arrived. A connection busy with a call is not an error here:
    that call is reading the same socket and buffers every notification it passes, so
    the poll gives up its turn and returns what it has.
    """
    limit = min(limit, self._notification_buffer)
    if len(self._notifications) < limit:
        # Notification queue holds less than requested. Check if we can drain the
        # connection for more.
        try:
            with self._connection.transaction() as transaction:
                while (
                    len(self._notifications) < limit
                    and (message := transaction.try_receive()) is not None
                ):
                    _ = self._route(message)
        except ConnectionBusyError:
            logger.debug("connection busy; returning what is already buffered")

    if self._dropped:
        logger.warning(
            "dropped %d notification(s): the buffer holds %d and was not polled in time",
            self._dropped,
            self._notification_buffer,
        )
        self._dropped = 0
    return [
        self._notifications.popleft()
        for _ in range(min(limit, len(self._notifications)))
    ]
disconnect 
disconnect() -> None
Source code in src/icon/server/hardware_processing/rpc/client.py 
132
133
def disconnect(self) -> None:
    self._connection.disconnect()
notify 
notify(method_name: str, *args: Any) -> None

Send a one-way notification, for which the server sends no reply.

Source code in src/icon/server/hardware_processing/rpc/client.py 
216
217
218
219
220
def notify(self, method_name: str, *args: Any) -> None:
    """Send a one-way notification, for which the server sends no reply."""
    logger.debug("notify: %s%s", method_name, _Abbreviated(args))
    with self._connection.transaction() as transaction:
        transaction.send((MessageType.NOTIFICATION, method_name, args))
RPCNotification dataclass 
RPCNotification(method: str, params: Any)

A message the server pushed of its own accord, answering no request.

Attributes:

Name Type Description
method str
params Any
method instance-attribute 
method: str
params instance-attribute 
params: Any
RPCResponse dataclass 
RPCResponse(msgid: int, error: Any, result: Any)

A reply to one request, tied to it by :attr:msgid rather than by arrival order.

Attributes:

Name Type Description
error Any
msgid int
result Any
error instance-attribute 
error: Any
msgid instance-attribute 
msgid: int
result instance-attribute 
result: Any

connection

Classes:

Name Description
Connection

Blocking connection with a raw msgpack stream.
FramedConnection

Connection with framed transport: single msgpack message following a 4-byte length header.
Transaction

Exclusive use of a connection for one exchange, under a single deadline.

Functions:

Name Description
deadline_from

Compute absolute deadline from timeout.
time_left

Compute time left until deadline is reached.

Attributes:

Name Type Description
DEFAULT_KEEPALIVE_COUNT Final
DEFAULT_KEEPALIVE_IDLE Final
DEFAULT_KEEPALIVE_INTERVAL Final
DEFAULT_MAX_MESSAGE_SIZE Final

Msgpack maximum message size.
HEADER_SIZE Final
HEADER_STRUCT Final
MsgPackRecord
RECV_CHUNK_SIZE Final

Socket reads are done in chunks of this size.
logger
DEFAULT_KEEPALIVE_COUNT module-attribute 
DEFAULT_KEEPALIVE_COUNT: Final = 3
DEFAULT_KEEPALIVE_IDLE module-attribute 
DEFAULT_KEEPALIVE_IDLE: Final = 10
DEFAULT_KEEPALIVE_INTERVAL module-attribute 
DEFAULT_KEEPALIVE_INTERVAL: Final = 5
DEFAULT_MAX_MESSAGE_SIZE module-attribute 
DEFAULT_MAX_MESSAGE_SIZE: Final = 256 * 1024 * 1024

Msgpack maximum message size.

HEADER_SIZE module-attribute 
HEADER_SIZE: Final = HEADER_STRUCT.size
HEADER_STRUCT module-attribute 
HEADER_STRUCT: Final = struct.Struct('<I')
MsgPackRecord module-attribute 
MsgPackRecord = Any
RECV_CHUNK_SIZE module-attribute 
RECV_CHUNK_SIZE: Final = 256 * 1024

Socket reads are done in chunks of this size.

logger module-attribute 
logger = logging.getLogger(__name__)
Connection 
Connection(
    hostname: str,
    port: int,
    *,
    timeout: float | None = None,
    lock_timeout: float = 1.0,
    max_message_size: int = DEFAULT_MAX_MESSAGE_SIZE,
    keepalive: bool = True,
    keepalive_idle: int = DEFAULT_KEEPALIVE_IDLE,
    keepalive_interval: int = DEFAULT_KEEPALIVE_INTERVAL,
    keepalive_count: int = DEFAULT_KEEPALIVE_COUNT,
    user_timeout: float | None = None,
)

Blocking connection with a raw msgpack stream.

The connection is opened by :meth:connect. Every I/O operation must be wrapped in a :meth:transaction to be thread-safe.

On any error, the connection is closed. The caller must explicitly re-open with :meth:connect.

Parameters:

Name Type Description Default
hostname str

Host to connect to.

 required
port int

TCP port to connect to.

 required
timeout float | None

Default bound in seconds, used for connecting and for any :meth:receive or :meth:transaction whose caller passes no timeout. None blocks indefinitely.

 None
lock_timeout float

How long :meth:transaction waits for another thread’s round trip to finish before giving up acquiring the lock.

 1.0
max_message_size int

Messages larger than this will not be accepted and the connection closed.

 DEFAULT_MAX_MESSAGE_SIZE
keepalive bool

Enable OS-level TCP keepalive.

 True
keepalive_idle int

Seconds of inactivity before the first probe.

 DEFAULT_KEEPALIVE_IDLE
keepalive_interval int

Seconds between probes once they start.

 DEFAULT_KEEPALIVE_INTERVAL
keepalive_count int

Unanswered probes before the connection is declared dead.

 DEFAULT_KEEPALIVE_COUNT
user_timeout float | None

Seconds until sent data may go unacknowledged before the kernel declares the connection dead. Defaults to the full keepalive budget. Applied on Linux whether or not keepalive is set; ignored elsewhere.

 None

Methods:

Name Description
__repr__
connect

Open the connection, replacing any socket this instance still holds.
disconnect

Close the connection.
locked

Hold the connection without starting an exchange.
receive

Wait for the next message.
send

Encode and write one message.
transaction

Context manager for owning the connection for one exchange.
try_receive

Non-blocking variant of :meth:receive. Suitable for polling.

Attributes:

Name Type Description
is_connected bool

Check if the socket is up.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def __init__(
    self,
    hostname: str,
    port: int,
    *,
    timeout: float | None = None,
    lock_timeout: float = 1.0,
    max_message_size: int = DEFAULT_MAX_MESSAGE_SIZE,
    keepalive: bool = True,
    keepalive_idle: int = DEFAULT_KEEPALIVE_IDLE,
    keepalive_interval: int = DEFAULT_KEEPALIVE_INTERVAL,
    keepalive_count: int = DEFAULT_KEEPALIVE_COUNT,
    user_timeout: float | None = None,
) -> None:
    self._hostname = hostname
    self._port = port
    self._timeout = timeout
    self._lock_timeout = lock_timeout
    self._max_message_size = max_message_size
    self._keepalive = keepalive
    self._keepalive_idle = keepalive_idle
    self._keepalive_interval = keepalive_interval
    self._keepalive_count = keepalive_count
    self._user_timeout = (
        keepalive_idle + keepalive_interval * keepalive_count
        if user_timeout is None
        else user_timeout
    )

    self._lock = threading.Lock()
    self._packer = msgpack.Packer(use_bin_type=True)
    self._unpacker = self._new_unpacker()
    self._socket: socket.socket | None = None
is_connected property 
is_connected: bool

Check if the socket is up.

Reports False if FIN or RST was received. Idle or dead connections will report as connected until keepalive timeout is reached (if configured).

__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/connection.py 
172
173
def __repr__(self) -> str:
    return f"msgpack(stream)://{self._hostname}:{self._port}"
connect 
connect() -> None

Open the connection, replacing any socket this instance still holds.

Idempotent, and the only way a socket is ever installed. Any previous socket is closed and the decoder is reset first, so no bytes left over from a dead stream can be spliced onto the new one and no file descriptor is orphaned.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def connect(self) -> None:
    """Open the connection, replacing any socket this instance still holds.

    Idempotent, and the only way a socket is ever installed. Any previous socket is
    closed and the decoder is reset first, so no bytes left over from a dead stream
    can be spliced onto the new one and no file descriptor is orphaned.
    """
    with self._lock:
        self._close()
        self._reset_decoder()
        sock = socket.create_connection(
            (self._hostname, self._port), timeout=self._timeout
        )
        self._configure_socket(sock)
        self._socket = sock
disconnect 
disconnect() -> None

Close the connection.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
205
206
207
208
def disconnect(self) -> None:
    """Close the connection."""
    with self._lock:
        self._close()
locked 
locked() -> Generator[None]

Hold the connection without starting an exchange.

For operations that must not interleave with a round trip but send nothing themselves, such as reconnecting.

Raises:

Type Description
ConnectionBusyError

if the connection lock could not be acquired within lock_timeout.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@contextmanager
def locked(self) -> Generator[None]:
    """Hold the connection without starting an exchange.

    For operations that must not interleave with a round trip but send nothing
    themselves, such as reconnecting.

    Raises:
        ConnectionBusyError: if the connection lock could not be acquired within
            ``lock_timeout``.
    """
    if not self._lock.acquire(timeout=self._lock_timeout):
        raise ConnectionBusyError(
            f"connection still in use after {self._lock_timeout:g}s"
        )
    try:
        yield
    finally:
        self._lock.release()
receive 
receive(timeout: float | None = None) -> MsgPackRecord

Wait for the next message.

Blocks until full message is received, potentially over multiple reads.

Parameters:

Name Type Description Default
timeout float | None

Seconds to wait for a whole message, potentially over multiple reads. None falls back to the connection’s own timeout.

 None

Returns:

Type Description
MsgPackRecord

The unpacked opaque message.

Raises:

Type Description
ProtocolError

If the bytes cannot belong to a valid msgpack stream.
TimeoutError

If no complete message arrives in time.
ConnectionError

If the peer closed the connection.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def receive(self, timeout: float | None = None) -> MsgPackRecord:
    """Wait for the next message.

    Blocks until full message is received, potentially over multiple reads.

    Args:
        timeout: Seconds to wait for a whole message, potentially over multiple reads.
            ``None`` falls back to the connection's own ``timeout``.

    Returns:
        The unpacked opaque message.

    Raises:
        ProtocolError: If the bytes cannot belong to a valid msgpack stream.
        TimeoutError: If no complete message arrives in time.
        ConnectionError: If the peer closed the connection.
    """
    return self._read_message(
        deadline_from(self._timeout if timeout is None else timeout)
    )
send 
send(message: MsgPackRecord) -> None

Encode and write one message.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
305
306
307
def send(self, message: MsgPackRecord) -> None:
    """Encode and write one message."""
    self._sock().sendall(self._packer.pack(message))
transaction 
transaction(
    timeout: float | None = None,
) -> Generator[Transaction]

Context manager for owning the connection for one exchange.

Blocks at most lock_timeout seconds on the connection lock.

Parameters:

Name Type Description Default
timeout float | None

Total time limit for the transaction to complete. None falls back to the connection’s own timeout.

 None

Yields:

Name Type Description
A Generator[Transaction]

class:Transaction object for send/receive operations.

A failed exchange closes the connection but never re-opens it. Once a read has timed out or the framing has desynced, the bytes still in flight belong to a request nobody is waiting for, so the stream cannot be reused – but reconnecting here would do it silently, behind a caller that may have session state to re-establish (discovered ids, subscriptions) and no way to notice it must. So the socket is dropped, :attr:is_connected goes false, and reconnecting is the caller’s decision.

Raises:

Type Description
ConnectionBusyError

if the connection lock could not be acquired within lock_timeout. Nothing was sent, so the connection stays healthy.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
@contextmanager
def transaction(self, timeout: float | None = None) -> Generator[Transaction]:
    """Context manager for owning the connection for one exchange.

    Blocks at most ``lock_timeout`` seconds on the connection lock.

    Args:
        timeout: Total time limit for the transaction to complete.
            ``None`` falls back to the connection's own ``timeout``.

    Yields:
        A :class:`Transaction` object for send/receive operations.

    A failed exchange closes the connection but never re-opens it. Once a read has
    timed out or the framing has desynced, the bytes still in flight belong to a
    request nobody is waiting for, so the stream cannot be reused -- but reconnecting
    here would do it silently, behind a caller that may have session state to
    re-establish (discovered ids, subscriptions) and no way to notice it must. So the
    socket is dropped, :attr:`is_connected` goes false, and reconnecting is the
    caller's decision.

    Raises:
        ConnectionBusyError: if the connection lock could not be acquired within
            ``lock_timeout``. Nothing was sent, so the connection stays healthy.
    """
    with self.locked():
        try:
            deadline = deadline_from(self._timeout if timeout is None else timeout)
            yield Transaction(self, deadline)
        except (OSError, ProtocolError):
            logger.debug("dropping %r after a failed exchange", self, exc_info=True)
            self._close()
            raise
try_receive 
try_receive() -> MsgPackRecord | None

Non-blocking variant of :meth:receive. Suitable for polling.

Consumes what the decoder already holds, plus whatever bytes are readable right now. A message that has only partly arrived stays buffered for the next call, so this never waits on the peer.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def try_receive(self) -> MsgPackRecord | None:
    """Non-blocking variant of :meth:`receive`. Suitable for polling.

    Consumes what the decoder already holds, plus whatever bytes are readable right
    now. A message that has only partly arrived stays buffered for the next call, so
    this never waits on the peer.
    """
    message = self._next_buffered()
    if message is not _NOTHING:
        return message
    chunk = self._recv_available(RECV_CHUNK_SIZE)
    if not chunk:
        return None
    self._feed(chunk)
    message = self._next_buffered()
    return None if message is _NOTHING else message
FramedConnection 
FramedConnection(*args: Any, **kwargs: Any)

Bases: Connection

Connection with framed transport: single msgpack message following a 4-byte length header.

Bytes are accumulated in one buffer that both the blocking and the polling read path draw on, so a frame that arrives in pieces is never half-consumed by a poll.

Methods:

Name Description
__repr__
connect

Open the connection, replacing any socket this instance still holds.
disconnect

Close the connection.
locked

Hold the connection without starting an exchange.
receive

Wait for the next message.
send

Override for the framed decode case. Prepend message with length header.
transaction

Context manager for owning the connection for one exchange.
try_receive

Attributes:

Name Type Description
is_connected bool

Check if the socket is up.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
406
407
408
def __init__(self, *args: Any, **kwargs: Any) -> None:
    self._buffer = bytearray()
    super().__init__(*args, **kwargs)
is_connected property 
is_connected: bool

Check if the socket is up.

Reports False if FIN or RST was received. Idle or dead connections will report as connected until keepalive timeout is reached (if configured).

__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/connection.py 
410
411
def __repr__(self) -> str:
    return f"msgpack(framed)://{self._hostname}:{self._port}"
connect 
connect() -> None

Open the connection, replacing any socket this instance still holds.

Idempotent, and the only way a socket is ever installed. Any previous socket is closed and the decoder is reset first, so no bytes left over from a dead stream can be spliced onto the new one and no file descriptor is orphaned.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def connect(self) -> None:
    """Open the connection, replacing any socket this instance still holds.

    Idempotent, and the only way a socket is ever installed. Any previous socket is
    closed and the decoder is reset first, so no bytes left over from a dead stream
    can be spliced onto the new one and no file descriptor is orphaned.
    """
    with self._lock:
        self._close()
        self._reset_decoder()
        sock = socket.create_connection(
            (self._hostname, self._port), timeout=self._timeout
        )
        self._configure_socket(sock)
        self._socket = sock
disconnect 
disconnect() -> None

Close the connection.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
205
206
207
208
def disconnect(self) -> None:
    """Close the connection."""
    with self._lock:
        self._close()
locked 
locked() -> Generator[None]

Hold the connection without starting an exchange.

For operations that must not interleave with a round trip but send nothing themselves, such as reconnecting.

Raises:

Type Description
ConnectionBusyError

if the connection lock could not be acquired within lock_timeout.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@contextmanager
def locked(self) -> Generator[None]:
    """Hold the connection without starting an exchange.

    For operations that must not interleave with a round trip but send nothing
    themselves, such as reconnecting.

    Raises:
        ConnectionBusyError: if the connection lock could not be acquired within
            ``lock_timeout``.
    """
    if not self._lock.acquire(timeout=self._lock_timeout):
        raise ConnectionBusyError(
            f"connection still in use after {self._lock_timeout:g}s"
        )
    try:
        yield
    finally:
        self._lock.release()
receive 
receive(timeout: float | None = None) -> MsgPackRecord

Wait for the next message.

Blocks until full message is received, potentially over multiple reads.

Parameters:

Name Type Description Default
timeout float | None

Seconds to wait for a whole message, potentially over multiple reads. None falls back to the connection’s own timeout.

 None

Returns:

Type Description
MsgPackRecord

The unpacked opaque message.

Raises:

Type Description
ProtocolError

If the bytes cannot belong to a valid msgpack stream.
TimeoutError

If no complete message arrives in time.
ConnectionError

If the peer closed the connection.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def receive(self, timeout: float | None = None) -> MsgPackRecord:
    """Wait for the next message.

    Blocks until full message is received, potentially over multiple reads.

    Args:
        timeout: Seconds to wait for a whole message, potentially over multiple reads.
            ``None`` falls back to the connection's own ``timeout``.

    Returns:
        The unpacked opaque message.

    Raises:
        ProtocolError: If the bytes cannot belong to a valid msgpack stream.
        TimeoutError: If no complete message arrives in time.
        ConnectionError: If the peer closed the connection.
    """
    return self._read_message(
        deadline_from(self._timeout if timeout is None else timeout)
    )
send 
send(message: MsgPackRecord) -> None

Override for the framed decode case. Prepend message with length header.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
420
421
422
423
def send(self, message: MsgPackRecord) -> None:
    """Override for the framed decode case. Prepend message with length header."""
    body: bytes = self._packer.pack(message)
    self._sock().sendall(HEADER_STRUCT.pack(len(body)) + body)
transaction 
transaction(
    timeout: float | None = None,
) -> Generator[Transaction]

Context manager for owning the connection for one exchange.

Blocks at most lock_timeout seconds on the connection lock.

Parameters:

Name Type Description Default
timeout float | None

Total time limit for the transaction to complete. None falls back to the connection’s own timeout.

 None

Yields:

Name Type Description
A Generator[Transaction]

class:Transaction object for send/receive operations.

A failed exchange closes the connection but never re-opens it. Once a read has timed out or the framing has desynced, the bytes still in flight belong to a request nobody is waiting for, so the stream cannot be reused – but reconnecting here would do it silently, behind a caller that may have session state to re-establish (discovered ids, subscriptions) and no way to notice it must. So the socket is dropped, :attr:is_connected goes false, and reconnecting is the caller’s decision.

Raises:

Type Description
ConnectionBusyError

if the connection lock could not be acquired within lock_timeout. Nothing was sent, so the connection stays healthy.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
@contextmanager
def transaction(self, timeout: float | None = None) -> Generator[Transaction]:
    """Context manager for owning the connection for one exchange.

    Blocks at most ``lock_timeout`` seconds on the connection lock.

    Args:
        timeout: Total time limit for the transaction to complete.
            ``None`` falls back to the connection's own ``timeout``.

    Yields:
        A :class:`Transaction` object for send/receive operations.

    A failed exchange closes the connection but never re-opens it. Once a read has
    timed out or the framing has desynced, the bytes still in flight belong to a
    request nobody is waiting for, so the stream cannot be reused -- but reconnecting
    here would do it silently, behind a caller that may have session state to
    re-establish (discovered ids, subscriptions) and no way to notice it must. So the
    socket is dropped, :attr:`is_connected` goes false, and reconnecting is the
    caller's decision.

    Raises:
        ConnectionBusyError: if the connection lock could not be acquired within
            ``lock_timeout``. Nothing was sent, so the connection stays healthy.
    """
    with self.locked():
        try:
            deadline = deadline_from(self._timeout if timeout is None else timeout)
            yield Transaction(self, deadline)
        except (OSError, ProtocolError):
            logger.debug("dropping %r after a failed exchange", self, exc_info=True)
            self._close()
            raise
try_receive 
try_receive() -> MsgPackRecord | None
Source code in src/icon/server/hardware_processing/rpc/connection.py 
433
434
435
436
437
438
439
440
441
def try_receive(self) -> MsgPackRecord | None:
    while True:
        message = self._next_buffered()
        if message is not _NOTHING:
            return message
        chunk = self._recv_available(RECV_CHUNK_SIZE)
        if not chunk:
            return None
        self._buffer += chunk
Transaction 
Transaction(connection: Connection, deadline: float | None)

Exclusive use of a connection for one exchange, under a single deadline.

Handed out by :meth:Connection.transaction, which holds the connection’s lock for as long as this object is in scope. Every read draws on the same budget, so a reply arriving behind a burst of notifications still cannot outlive the caller’s timeout – and no call site has to do the arithmetic. That is the point of the object: the deadline is derived once, here, and no absolute time ever crosses the boundary.

Methods:

Name Description
receive

Wait for the next message, within what is left of the budget.
send

Encode and write one message.
try_receive

Return the next message if one is already available, else None.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
76
77
78
def __init__(self, connection: Connection, deadline: float | None) -> None:
    self._connection = connection
    self._deadline = deadline
receive 
receive() -> MsgPackRecord

Wait for the next message, within what is left of the budget.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
84
85
86
def receive(self) -> MsgPackRecord:
    """Wait for the next message, within what is left of the budget."""
    return self._connection._read_message(self._deadline)
send 
send(message: MsgPackRecord) -> None

Encode and write one message.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
80
81
82
def send(self, message: MsgPackRecord) -> None:
    """Encode and write one message."""
    self._connection.send(message)
try_receive 
try_receive() -> MsgPackRecord | None

Return the next message if one is already available, else None.

Spends none of the budget, because it never waits.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
88
89
90
91
92
93
def try_receive(self) -> MsgPackRecord | None:
    """Return the next message if one is already available, else ``None``.

    Spends none of the budget, because it never waits.
    """
    return self._connection.try_receive()
deadline_from 
deadline_from(timeout: float | None) -> float | None

Compute absolute deadline from timeout.

Source code in src/icon/server/hardware_processing/rpc/connection.py 
44
45
46
def deadline_from(timeout: float | None) -> float | None:
    """Compute absolute deadline from timeout."""
    return None if timeout is None else time.monotonic() + timeout
time_left 
time_left(deadline: float | None) -> float | None

Compute time left until deadline is reached.

Returns:

Type Description
float | None

Remaining time in seconds. Always positive.

Raises:

Type Description
TimeoutError

If deadline has already passed.
Source code in src/icon/server/hardware_processing/rpc/connection.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def time_left(deadline: float | None) -> float | None:
    """Compute time left until deadline is reached.

    Returns:
        Remaining time in seconds. Always positive.

    Raises:
        TimeoutError: If ``deadline`` has already passed.
    """
    if deadline is None:
        return None
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        raise TimeoutError("timeout waiting for a response")
    return remaining

errors

Exception hierarchy for the RPC client.

Classes:

Name Description
ConnectionBusyError

The thread lock could not be acquired within lock_timeout.
ProtocolError

A server message could not be decoded.
RPCError

Base class for every error this package defines.
RPCResponseError

The server answered a request with a non-nil error field.
ConnectionBusyError

Bases: RPCError, TimeoutError

The thread lock could not be acquired within lock_timeout.

ProtocolError

Bases: RPCError

A server message could not be decoded.

This error is fatal to the connection as it otherwise lead to desynchronization between client and server. Recover by opening the connection again.

RPCError

Bases: Exception

Base class for every error this package defines.

RPCResponseError 
RPCResponseError(msgid: int, error: Any)

Bases: RPCError

The server answered a request with a non-nil error field.

The payload has the form [code, "message"].

:func:icon.server.hardware_processing.utils.extract_hardware_error_message parses the error: prefix to pull the hardware message out for hardware_processing/worker.py, so it must not change without updating that helper.

Attributes:

Name Type Description
error
msgid
Source code in src/icon/server/hardware_processing/rpc/errors.py 
22
23
24
25
def __init__(self, msgid: int, error: Any) -> None:
    super().__init__(f"Server reported msgid {msgid:d} error: {error}")
    self.msgid = msgid
    self.error = error
error instance-attribute 
error = error
msgid instance-attribute 
msgid = msgid

zedboard

Classes:

Name Description
ExperimentResult
Zedboard

Direct RPC interface to the legacy experiment runtime on the Zedboard.
ZedboardError

The device is configured differently than the client expects.
ZedboardSeqRunner

Representation of the Sequence Runner Experiment on the Zedboard.
ZedboardSeqRunnerCached

Cached variant of ZedboardSeqRunner.

Attributes:

Name Type Description
PageDescr
ParamDescr
ParamVal
logger
PageDescr module-attribute 
PageDescr = tuple[str, int, list[int]]
ParamDescr module-attribute 
ParamDescr = tuple[
    tuple[int, ParamVal],
    tuple[str, int, int, str, list[Any]],
]
ParamVal module-attribute 
ParamVal = bool | int | float | str
logger module-attribute 
logger = logging.getLogger(__name__)
ExperimentResult dataclass 
ExperimentResult(
    result_channels: dict[str, float] = dict(),
    vector_channels: dict[str, list[float]] = dict(),
    shot_channels: dict[str, list[int]] = dict(),
)

Attributes:

Name Type Description
result_channels dict[str, float]
shot_channels dict[str, list[int]]
vector_channels dict[str, list[float]]
result_channels class-attribute instance-attribute 
result_channels: dict[str, float] = field(
    default_factory=dict
)
shot_channels class-attribute instance-attribute 
shot_channels: dict[str, list[int]] = field(
    default_factory=dict
)
vector_channels class-attribute instance-attribute 
vector_channels: dict[str, list[float]] = field(
    default_factory=dict
)
Zedboard 
Zedboard(
    hostname: str = "zedboard.lab",
    port: int = 6007,
    timeout: float | None = 5,
)

Direct RPC interface to the legacy experiment runtime on the Zedboard.

Provides typed stateless methods for the following operations * read pages, parameters, channels, and remote action. * write parameters and invoke remote actions. * execute experiment with channel name resolution

Intended to serve as a base class for specialized clients on top of the experiment runtime.

Note: This is a compatibility module for the legacy Zedboard controller interface. The interface is deprecated and will be replaced in the future.

Methods:

Name Description
__enter__
__exit__
__repr__
call_remote_action

Call remote action by page id and action id. See :meth:get_remote_actions.
connect

Establish a connection to the Zedboard.
disconnect
get_data_channels

Returns list of data channel names for the page.
get_pages

Returns page list. Position in the list is the page’s id.
get_params

Returns global parameter list.
get_remote_actions

Returns list of remote action names for the page. Position in the list is the action’s id.
get_shot_channels

Returns list of shot channel names for the page.
get_vector_channels

Returns list of vector channel names for the page.
run_experiment

Run the experiment corresponding to the page.
set_param
set_params

Write several parameters in one round trip.
set_ttl_mask

Set TTL mask, state.

Attributes:

Name Type Description
is_connected bool

Whether the underlying socket is up.
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    hostname: str = "zedboard.lab",
    port: int = 6007,
    timeout: float | None = 5,
) -> None:
    self._client = MsgPackRPCClient(
        hostname=hostname, port=port, timeout=timeout, framed=True
    )
is_connected property 
is_connected: bool

Whether the underlying socket is up.

__enter__ 
__enter__() -> Self
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
61
62
63
def __enter__(self) -> Self:
    self.connect()
    return self
__exit__ 
__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
65
66
67
68
69
70
71
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    self.disconnect()
__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
58
59
def __repr__(self) -> str:
    return f"{self.__class__.__name__}[{self._client}]"
call_remote_action 
call_remote_action(page_id: int, action_id: int) -> Any

Call remote action by page id and action id. See :meth:get_remote_actions.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
133
134
135
def call_remote_action(self, page_id: int, action_id: int) -> Any:
    """Call remote action by page id and action id. See :meth:`get_remote_actions`."""
    return self._invoke("callRemoteAction", page_id, action_id)
connect 
connect() -> None

Establish a connection to the Zedboard.

Raises:

Type Description
ConnectionRefusedError

if the peer does not accept the connection. Probably not listening.
OSError

if the host cannot be reached or resolved.
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
73
74
75
76
77
78
79
80
81
def connect(self) -> None:
    """Establish a connection to the Zedboard.

    Raises:
        ConnectionRefusedError: if the peer does not accept the connection. Probably
            not listening.
        OSError: if the host cannot be reached or resolved.
    """
    self._client.connect()
disconnect 
disconnect() -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
83
84
def disconnect(self) -> None:
    self._client.disconnect()
get_data_channels 
get_data_channels(page_id: int) -> list[str]

Returns list of data channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
106
107
108
def get_data_channels(self, page_id: int) -> list[str]:
    """Returns list of data channel names for the page."""
    return self._get_channels("dataChannels", page_id)
get_pages 
get_pages() -> list[PageDescr]

Returns page list. Position in the list is the page’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
95
96
97
def get_pages(self) -> list[PageDescr]:
    """Returns page list. Position in the list is the page's id."""
    return self._invoke("pages")
get_params 
get_params() -> list[ParamDescr]

Returns global parameter list.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
 99
100
101
def get_params(self) -> list[ParamDescr]:
    """Returns global parameter list."""
    return self._invoke("params")
get_remote_actions 
get_remote_actions(page_id: int) -> list[str]

Returns list of remote action names for the page. Position in the list is the action’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
118
119
120
def get_remote_actions(self, page_id: int) -> list[str]:
    """Returns list of remote action names for the page. Position in the list is the action's id."""
    return self._invoke("remoteActions", page_id)
get_shot_channels 
get_shot_channels(page_id: int) -> list[str]

Returns list of shot channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
110
111
112
def get_shot_channels(self, page_id: int) -> list[str]:
    """Returns list of shot channel names for the page."""
    return self._get_channels("shotChannels", page_id)
get_vector_channels 
get_vector_channels(page_id: int) -> list[str]

Returns list of vector channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
114
115
116
def get_vector_channels(self, page_id: int) -> list[str]:
    """Returns list of vector channel names for the page."""
    return self._get_channels("vectorChannels", page_id)
run_experiment 
run_experiment(page_id: int) -> ExperimentResult

Run the experiment corresponding to the page.

Channel names are resolved dynamically. This is required for experiments which build their channels dynamically (like the sequence runner experiment).

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
def run_experiment(self, page_id: int) -> ExperimentResult:
    """Run the experiment corresponding to the page.

    Channel names are resolved dynamically. This is required for experiments which
    build their channels dynamically (like the sequence runner experiment).
    """
    data_res, shot_res, vec_res = self._invoke("runExperiment", page_id)
    data_names, shot_names, vec_names = self._result_channel_names(page_id)
    return ExperimentResult(
        result_channels=self._zip_channels("data", page_id, data_names, data_res),
        shot_channels=self._zip_channels("shot", page_id, shot_names, shot_res),
        vector_channels=self._zip_channels("vector", page_id, vec_names, vec_res),
    )
set_param 
set_param(param_id: int, value: ParamVal) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
122
123
def set_param(self, param_id: int, value: ParamVal) -> None:
    self.set_params([(param_id, value)])
set_params 
set_params(params: Iterable[tuple[int, ParamVal]]) -> None

Write several parameters in one round trip.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
125
126
127
def set_params(self, params: Iterable[tuple[int, ParamVal]]) -> None:
    """Write several parameters in one round trip."""
    self._invoke("setParams", [[param_id, value] for param_id, value in params])
set_ttl_mask 
set_ttl_mask(mask: int, state: int) -> None

Set TTL mask, state.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
129
130
131
def set_ttl_mask(self, mask: int, state: int) -> None:
    """Set TTL `mask`, `state`."""
    return self._invoke("setTTLMasks", mask & 0xFFFFFFFF, state & 0xFFFFFFFF)
ZedboardError

Bases: RPCError

The device is configured differently than the client expects.

When reading and evaluating server state like page, parameter or remote action this error indicates unexpected results mich may indicate misconfiguration.

ZedboardSeqRunner 
ZedboardSeqRunner(
    hostname: str = "zedboard.lab",
    port: int = 6007,
    timeout: float | None = 5,
)

Bases: Zedboard

Representation of the Sequence Runner Experiment on the Zedboard.

On connect, the necessary configuration values for a sequence run are discovered once.

Usage:

zedboard = ZedboardSeqRunner(hostname="localhost", port=6000)
zedboard.connect()

The following steps are required for each sequence run:

zedboard.load_sequence(seq_json)
res = zedboard.run_sequence()

Methods:

Name Description
__enter__
__exit__
__repr__
call_remote_action

Call remote action by page id and action id. See :meth:get_remote_actions.
connect

Connect to the device and discover configuration values.
disconnect
get_data_channels

Returns list of data channel names for the page.
get_pages

Returns page list. Position in the list is the page’s id.
get_params

Returns global parameter list.
get_remote_actions

Returns list of remote action names for the page. Position in the list is the action’s id.
get_shot_channels

Returns list of shot channel names for the page.
get_vector_channels

Returns list of vector channel names for the page.
load_sequence

Transmit the sequence description to the device.
run_experiment

Run the experiment corresponding to the page.
run_sequence

Execute the experiment, fetch the channel names and map the result data to the channels.
set_param
set_params

Write several parameters in one round trip.
set_ttl_mask

Set TTL mask, state.

Attributes:

Name Type Description
is_connected bool

Whether the underlying socket is up.
is_ready bool

Whether the socket is up and discovery has completed.
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    hostname: str = "zedboard.lab",
    port: int = 6007,
    timeout: float | None = 5,
) -> None:
    self._client = MsgPackRPCClient(
        hostname=hostname, port=port, timeout=timeout, framed=True
    )
is_connected property 
is_connected: bool

Whether the underlying socket is up.

is_ready property 
is_ready: bool

Whether the socket is up and discovery has completed.

__enter__ 
__enter__() -> Self
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
61
62
63
def __enter__(self) -> Self:
    self.connect()
    return self
__exit__ 
__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
65
66
67
68
69
70
71
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    self.disconnect()
__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
58
59
def __repr__(self) -> str:
    return f"{self.__class__.__name__}[{self._client}]"
call_remote_action 
call_remote_action(page_id: int, action_id: int) -> Any

Call remote action by page id and action id. See :meth:get_remote_actions.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
133
134
135
def call_remote_action(self, page_id: int, action_id: int) -> Any:
    """Call remote action by page id and action id. See :meth:`get_remote_actions`."""
    return self._invoke("callRemoteAction", page_id, action_id)
connect 
connect() -> None

Connect to the device and discover configuration values.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
222
223
224
225
def connect(self) -> None:
    """Connect to the device and discover configuration values."""
    super().connect()
    self._discover()
disconnect 
disconnect() -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
83
84
def disconnect(self) -> None:
    self._client.disconnect()
get_data_channels 
get_data_channels(page_id: int) -> list[str]

Returns list of data channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
106
107
108
def get_data_channels(self, page_id: int) -> list[str]:
    """Returns list of data channel names for the page."""
    return self._get_channels("dataChannels", page_id)
get_pages 
get_pages() -> list[PageDescr]

Returns page list. Position in the list is the page’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
95
96
97
def get_pages(self) -> list[PageDescr]:
    """Returns page list. Position in the list is the page's id."""
    return self._invoke("pages")
get_params 
get_params() -> list[ParamDescr]

Returns global parameter list.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
 99
100
101
def get_params(self) -> list[ParamDescr]:
    """Returns global parameter list."""
    return self._invoke("params")
get_remote_actions 
get_remote_actions(page_id: int) -> list[str]

Returns list of remote action names for the page. Position in the list is the action’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
118
119
120
def get_remote_actions(self, page_id: int) -> list[str]:
    """Returns list of remote action names for the page. Position in the list is the action's id."""
    return self._invoke("remoteActions", page_id)
get_shot_channels 
get_shot_channels(page_id: int) -> list[str]

Returns list of shot channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
110
111
112
def get_shot_channels(self, page_id: int) -> list[str]:
    """Returns list of shot channel names for the page."""
    return self._get_channels("shotChannels", page_id)
get_vector_channels 
get_vector_channels(page_id: int) -> list[str]

Returns list of vector channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
114
115
116
def get_vector_channels(self, page_id: int) -> list[str]:
    """Returns list of vector channel names for the page."""
    return self._get_channels("vectorChannels", page_id)
load_sequence 
load_sequence(sequence_json: str) -> None

Transmit the sequence description to the device.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
257
258
259
def load_sequence(self, sequence_json: str) -> None:
    """Transmit the sequence description to the device."""
    self.set_param(self._param_id, sequence_json)
run_experiment 
run_experiment(page_id: int) -> ExperimentResult

Run the experiment corresponding to the page.

Channel names are resolved dynamically. This is required for experiments which build their channels dynamically (like the sequence runner experiment).

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
def run_experiment(self, page_id: int) -> ExperimentResult:
    """Run the experiment corresponding to the page.

    Channel names are resolved dynamically. This is required for experiments which
    build their channels dynamically (like the sequence runner experiment).
    """
    data_res, shot_res, vec_res = self._invoke("runExperiment", page_id)
    data_names, shot_names, vec_names = self._result_channel_names(page_id)
    return ExperimentResult(
        result_channels=self._zip_channels("data", page_id, data_names, data_res),
        shot_channels=self._zip_channels("shot", page_id, shot_names, shot_res),
        vector_channels=self._zip_channels("vector", page_id, vec_names, vec_res),
    )
run_sequence 
run_sequence() -> ExperimentResult

Execute the experiment, fetch the channel names and map the result data to the channels.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
261
262
263
def run_sequence(self) -> ExperimentResult:
    """Execute the experiment, fetch the channel names and map the result data to the channels."""
    return self.run_experiment(self._page_id)
set_param 
set_param(param_id: int, value: ParamVal) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
122
123
def set_param(self, param_id: int, value: ParamVal) -> None:
    self.set_params([(param_id, value)])
set_params 
set_params(params: Iterable[tuple[int, ParamVal]]) -> None

Write several parameters in one round trip.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
125
126
127
def set_params(self, params: Iterable[tuple[int, ParamVal]]) -> None:
    """Write several parameters in one round trip."""
    self._invoke("setParams", [[param_id, value] for param_id, value in params])
set_ttl_mask 
set_ttl_mask(mask: int, state: int) -> None

Set TTL mask, state.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
129
130
131
def set_ttl_mask(self, mask: int, state: int) -> None:
    """Set TTL `mask`, `state`."""
    return self._invoke("setTTLMasks", mask & 0xFFFFFFFF, state & 0xFFFFFFFF)
ZedboardSeqRunnerCached 
ZedboardSeqRunnerCached(*args: Any, **kwargs: Any)

Bases: ZedboardSeqRunner

Cached variant of ZedboardSeqRunner.

By default, the run_experiment routine fetches the channel names after every experiment run. This is because the channel names are not known a priori as they are built according to the instructions in the sequence description. This Sequence Runner infers the channel names from the sequence description and allows to retrieve the channel names directly from the cache. This avoids three round trips to the device for querying the individual channel names. The assumption is that three round trips takes longer than decoding the sequence description.

Usage:

Identical to :class:ZedboardSeqRunner

Classes:

Name Description
ChannelTypes

Channel types as defined in the sequence description with the {ChannelTypes}_channel_names key.

Methods:

Name Description
__enter__
__exit__
__repr__
call_remote_action

Call remote action by page id and action id. See :meth:get_remote_actions.
connect

Connect to the device and discover configuration values.
disconnect
get_data_channels

Returns list of data channel names for the page.
get_pages

Returns page list. Position in the list is the page’s id.
get_params

Returns global parameter list.
get_remote_actions

Returns list of remote action names for the page. Position in the list is the action’s id.
get_shot_channels

Returns list of shot channel names for the page.
get_vector_channels

Returns list of vector channel names for the page.
load_sequence

Update the cache with the new sequence description.
run_experiment

Run the experiment corresponding to the page.
run_sequence

Execute the experiment, fetch the channel names and map the result data to the channels.
set_param
set_params

Write several parameters in one round trip.
set_ttl_mask

Set TTL mask, state.

Attributes:

Name Type Description
is_connected bool

Whether the underlying socket is up.
is_ready bool

Whether the socket is up and discovery has completed.
marker
marker_len
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
291
292
293
294
295
296
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self.marker = '"header":'
    self.marker_len = len(self.marker)

    self._channel_names = {t: [] for t in ZedboardSeqRunnerCached.ChannelTypes}
is_connected property 
is_connected: bool

Whether the underlying socket is up.

is_ready property 
is_ready: bool

Whether the socket is up and discovery has completed.

marker instance-attribute 
marker = '"header":'
marker_len instance-attribute 
marker_len = len(self.marker)
ChannelTypes

Bases: StrEnum

Channel types as defined in the sequence description with the {ChannelTypes}_channel_names key.

Attributes:

Name Type Description
READOUT
SHOT
VECTOR
READOUT class-attribute instance-attribute 
READOUT = 'readout'
SHOT class-attribute instance-attribute 
SHOT = 'shot'
VECTOR class-attribute instance-attribute 
VECTOR = 'vector'
__enter__ 
__enter__() -> Self
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
61
62
63
def __enter__(self) -> Self:
    self.connect()
    return self
__exit__ 
__exit__(
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
65
66
67
68
69
70
71
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    self.disconnect()
__repr__ 
__repr__() -> str
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
58
59
def __repr__(self) -> str:
    return f"{self.__class__.__name__}[{self._client}]"
call_remote_action 
call_remote_action(page_id: int, action_id: int) -> Any

Call remote action by page id and action id. See :meth:get_remote_actions.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
133
134
135
def call_remote_action(self, page_id: int, action_id: int) -> Any:
    """Call remote action by page id and action id. See :meth:`get_remote_actions`."""
    return self._invoke("callRemoteAction", page_id, action_id)
connect 
connect() -> None

Connect to the device and discover configuration values.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
222
223
224
225
def connect(self) -> None:
    """Connect to the device and discover configuration values."""
    super().connect()
    self._discover()
disconnect 
disconnect() -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
83
84
def disconnect(self) -> None:
    self._client.disconnect()
get_data_channels 
get_data_channels(page_id: int) -> list[str]

Returns list of data channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
106
107
108
def get_data_channels(self, page_id: int) -> list[str]:
    """Returns list of data channel names for the page."""
    return self._get_channels("dataChannels", page_id)
get_pages 
get_pages() -> list[PageDescr]

Returns page list. Position in the list is the page’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
95
96
97
def get_pages(self) -> list[PageDescr]:
    """Returns page list. Position in the list is the page's id."""
    return self._invoke("pages")
get_params 
get_params() -> list[ParamDescr]

Returns global parameter list.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
 99
100
101
def get_params(self) -> list[ParamDescr]:
    """Returns global parameter list."""
    return self._invoke("params")
get_remote_actions 
get_remote_actions(page_id: int) -> list[str]

Returns list of remote action names for the page. Position in the list is the action’s id.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
118
119
120
def get_remote_actions(self, page_id: int) -> list[str]:
    """Returns list of remote action names for the page. Position in the list is the action's id."""
    return self._invoke("remoteActions", page_id)
get_shot_channels 
get_shot_channels(page_id: int) -> list[str]

Returns list of shot channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
110
111
112
def get_shot_channels(self, page_id: int) -> list[str]:
    """Returns list of shot channel names for the page."""
    return self._get_channels("shotChannels", page_id)
get_vector_channels 
get_vector_channels(page_id: int) -> list[str]

Returns list of vector channel names for the page.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
114
115
116
def get_vector_channels(self, page_id: int) -> list[str]:
    """Returns list of vector channel names for the page."""
    return self._get_channels("vectorChannels", page_id)
load_sequence 
load_sequence(sequence_json: str) -> None

Update the cache with the new sequence description.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
319
320
321
322
def load_sequence(self, sequence_json: str) -> None:
    """Update the cache with the new sequence description."""
    self._update_channel_names(sequence_json)
    super().load_sequence(sequence_json)
run_experiment 
run_experiment(page_id: int) -> ExperimentResult

Run the experiment corresponding to the page.

Channel names are resolved dynamically. This is required for experiments which build their channels dynamically (like the sequence runner experiment).

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
def run_experiment(self, page_id: int) -> ExperimentResult:
    """Run the experiment corresponding to the page.

    Channel names are resolved dynamically. This is required for experiments which
    build their channels dynamically (like the sequence runner experiment).
    """
    data_res, shot_res, vec_res = self._invoke("runExperiment", page_id)
    data_names, shot_names, vec_names = self._result_channel_names(page_id)
    return ExperimentResult(
        result_channels=self._zip_channels("data", page_id, data_names, data_res),
        shot_channels=self._zip_channels("shot", page_id, shot_names, shot_res),
        vector_channels=self._zip_channels("vector", page_id, vec_names, vec_res),
    )
run_sequence 
run_sequence() -> ExperimentResult

Execute the experiment, fetch the channel names and map the result data to the channels.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
261
262
263
def run_sequence(self) -> ExperimentResult:
    """Execute the experiment, fetch the channel names and map the result data to the channels."""
    return self.run_experiment(self._page_id)
set_param 
set_param(param_id: int, value: ParamVal) -> None
Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
122
123
def set_param(self, param_id: int, value: ParamVal) -> None:
    self.set_params([(param_id, value)])
set_params 
set_params(params: Iterable[tuple[int, ParamVal]]) -> None

Write several parameters in one round trip.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
125
126
127
def set_params(self, params: Iterable[tuple[int, ParamVal]]) -> None:
    """Write several parameters in one round trip."""
    self._invoke("setParams", [[param_id, value] for param_id, value in params])
set_ttl_mask 
set_ttl_mask(mask: int, state: int) -> None

Set TTL mask, state.

Source code in src/icon/server/hardware_processing/rpc/zedboard.py 
129
130
131
def set_ttl_mask(self, mask: int, state: int) -> None:
    """Set TTL `mask`, `state`."""
    return self._invoke("setTTLMasks", mask & 0xFFFFFFFF, state & 0xFFFFFFFF)

task

Classes:

Name Description
HardwareProcessingTask

HardwareProcessingTask

Bases: BaseModel

Methods:

Name Description
__lt__

Attributes:

Name Type Description
created datetime
data_point_index int
global_parameter_timestamp datetime
hardware_instructions str
model_config
outdated_tasks PriorityQueue[HardwareProcessingTask]
pre_processing_task PreProcessingTask
priority int
scan_progress ScanProgress
scanned_params dict[str, DatabaseValueType]
src_dir str | None
created instance-attribute 
created: datetime
data_point_index instance-attribute 
data_point_index: int
global_parameter_timestamp instance-attribute 
global_parameter_timestamp: datetime
hardware_instructions instance-attribute 
hardware_instructions: str
model_config class-attribute instance-attribute 
model_config = pydantic.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
scan_progress instance-attribute 
scan_progress: ScanProgress
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 
37
38
def __lt__(self, other: HardwareProcessingTask) -> bool:
    return (self.priority, self.created) < (other.priority, other.created)

tiqizedboard_controller

Classes:

Name Description
ZedboardController

Zedboard Hardware Controller relying on the tiqi_zedboard client.

Attributes:

Name Type Description
logger

logger module-attribute 

logger = logging.getLogger(__name__)

ZedboardController 

ZedboardController(
    *, host: str, port: int, timeout: int = 5
)

Bases: HardwareController

Zedboard Hardware Controller relying on the tiqi_zedboard client.

Initialise the controller.

Parameters:

Name Type Description Default
host str

Hostname of the Zedboard.

 required
port int

Port the Zedboard RPC server listens on.

 required
timeout int

RPC timeout in seconds for calls such as runExperiment. Configurable in the config file.

 5

Methods:

Name Description
connect
receive
run
send
status

Attributes:

Name Type Description
connected bool
Source code in src/icon/server/hardware_processing/tiqizedboard_controller.py 
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, *, host: str, port: int, timeout: int = 5) -> None:
    """Initialise the controller.

    Args:
        host: Hostname of the Zedboard.
        port: Port the Zedboard RPC server listens on.
        timeout: RPC timeout in seconds for calls such as runExperiment. Configurable in the config file.
    """
    self._host = host
    self._port = port
    self._timeout = timeout
    self._zedboard: tiqi_zedboard.zedboard.Zedboard | None = None
connected property 
connected: bool
connect 
connect() -> None
Source code in src/icon/server/hardware_processing/tiqizedboard_controller.py 
38
39
40
41
42
43
44
def connect(self) -> None:
    logger.info("Connecting to the Zedboard")
    self._zedboard = tiqi_zedboard.zedboard.Zedboard(
        hostname=self._host, port=self._port, timeout=self._timeout
    )
    if not self.connected:
        logger.warning("Failed to connect to the Zedboard")
receive 
receive() -> Readouts
Source code in src/icon/server/hardware_processing/tiqizedboard_controller.py 
68
69
70
71
72
73
74
75
76
77
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/tiqizedboard_controller.py 
65
66
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/tiqizedboard_controller.py 
58
59
60
61
62
63
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/tiqizedboard_controller.py 
79
80
def status(self) -> tuple[StatusFlag, str, Any]:
    return (StatusFlag.UNKNOWN, "", None)

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 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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 = logging.getLogger(__name__)

timezone module-attribute 

timezone = pytz.timezone(get_config().date.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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@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.scan_progress.complete(task.pre_processing_task.job_run.id)
            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("Error in hardware worker.")
            try_update_run_by_id(
                run_id=task.pre_processing_task.job_run.id,
                status=JobRunStatus.FAILED,
                log=extract_hardware_error_message(e),
            )
        finally:
            task.scan_progress.complete(task.pre_processing_task.job_run.id)

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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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

Zedboard Hardware Controller using a stripped-down minimal Zedboard-compatible RPC client.

Attributes:

Name Type Description
logger

logger module-attribute 

logger = logging.getLogger(__name__)

ZedboardController 

ZedboardController(
    *,
    host: str,
    port: int,
    timeout: int = 5,
    cached: bool = True,
)

Bases: HardwareController

Zedboard Hardware Controller using a stripped-down minimal Zedboard-compatible RPC client.

Initialise the controller.

Parameters:

Name Type Description Default
host str

Hostname of the zedboard.

 required
port int

Port the Zedoard RPC server listens on.

 required
timeout int

RPC timeout in seconds for calls such as runExperiment. Configurable in the config file.

 5
cached bool

Whether to read the channel names out of the sequence description instead of asking the device for them after every run. Saves three round trips per data point.

 True

Methods:

Name Description
connect
receive
run

The sequence is executed in the :meth:receive call. Nothing to be done here.
send
status

Attributes:

Name Type Description
connected bool

Zedboard is ready to process sequences.
Source code in src/icon/server/hardware_processing/zedboard_controller.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(
    self, *, host: str, port: int, timeout: int = 5, cached: bool = True
) -> None:
    """Initialise the controller.

    Args:
        host: Hostname of the zedboard.
        port: Port the Zedoard RPC server listens on.
        timeout: RPC timeout in seconds for calls such as runExperiment. Configurable
            in the config file.
        cached: Whether to read the channel names out of the sequence description
            instead of asking the device for them after every run. Saves three round
            trips per data point.
    """
    self._host = host
    self._port = port
    self._timeout = timeout
    self._zedboard = (
        zedboard.ZedboardSeqRunnerCached if cached else zedboard.ZedboardSeqRunner
    )(hostname=self._host, port=self._port, timeout=timeout)
connected property 
connected: bool

Zedboard is ready to process sequences.

connect 
connect() -> None
Source code in src/icon/server/hardware_processing/zedboard_controller.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def connect(self) -> None:
    try:
        self._zedboard.connect()
    except zedboard.ZedboardError as e:
        logger.warning(
            "Connected to %r, but it may not be configured properly sequence running: %s",
            self._zedboard,
            e,
        )
    except (ConnectionResetError, ConnectionRefusedError, OSError) as e:
        logger.warning(
            "Could not connect to the Zedboard: %s (%r)", e, self._zedboard
        )
    else:
        logger.info("Connected to the Zedboard: %s", self._zedboard)
receive 
receive() -> Readouts
Source code in src/icon/server/hardware_processing/zedboard_controller.py 
72
73
74
75
76
77
78
79
def receive(self) -> Readouts:
    results = self._zedboard.run_sequence()

    return Readouts(
        result_channels=results.result_channels,
        vector_channels=results.vector_channels,
        shot_channels=results.shot_channels,
    )
run 
run() -> None

The sequence is executed in the :meth:receive call. Nothing to be done here.

Source code in src/icon/server/hardware_processing/zedboard_controller.py 
69
70
def run(self) -> None:
    """The sequence is executed in the :meth:`receive` call. Nothing to be done here."""
send 
send(data: str) -> None
Source code in src/icon/server/hardware_processing/zedboard_controller.py 
59
60
61
62
63
64
65
66
67
def send(self, data: str) -> None:
    if not self.connected:
        self.connect()
    if not self.connected:
        raise RuntimeError(
            f"Could not connect to the Zedboard at {self._host}:{self._port} "
            f"while trying to run a command"
        )
    self._zedboard.load_sequence(data)
status 
status() -> tuple[StatusFlag, str, Any]
Source code in src/icon/server/hardware_processing/zedboard_controller.py 
81
82
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 
19
20
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 = logging.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 
24
25
26
27
28
29
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@handle_keyboard_interrupt(logger)
def run(self) -> None:
    logger.info("Post-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

        try:
            ExperimentDataRepository.write_experiment_data_by_job_id(
                job_id=task.pre_processing_task.job.id,
                data_point=task.data_point,
            )
        except Exception as e:
            logger.exception(
                "Post-processing of job with id '%s' failed",
                task.pre_processing_task.job.id,
            )
            try_update_run_by_id(
                run_id=task.pre_processing_task.job_run.id,
                status=JobRunStatus.FAILED,
                log=f"Post-processing error: {e}",
            )

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 = pydantic.ConfigDict(
    arbitrary_types_allowed=True
)
priority class-attribute instance-attribute 
priority: int = pydantic.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 
21
22
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.
clear_queue
consume_queue
create_hardware_instructions
freeze_dict
get_scan_combinations

Generates all combinations of scan parameters for a given job.
is_global_parameter

Whether a parameter is a global one rather than scoped to an experiment.
parameter_namespace

Return the namespace a parameter identifier is scoped to (empty if it has none).
parse_experiment_identifier

Parses an experiment identifier.

Attributes:

Name Type Description
GLOBAL_NAMESPACE_SEGMENT

Module path segment identifying the experiment library’s global parameters.
SCAN_COMPLETION_POLL_INTERVAL

Seconds to wait between completion checks once every data point of a regular scan
ScanCombination
T
logger
timezone

GLOBAL_NAMESPACE_SEGMENT module-attribute 

GLOBAL_NAMESPACE_SEGMENT = 'globals'

Module path segment identifying the experiment library’s global parameters.

pycrystal derives a parameter’s namespace from where it is declared: a parameter declared inside an experiment instance gets <module>.<ClassName>.<instance name> Global parameters live in the library’s globals package, e.g. experiment_library.globals.global_parameters.

SCAN_COMPLETION_POLL_INTERVAL module-attribute 

SCAN_COMPLETION_POLL_INTERVAL = 0.1

Seconds to wait between completion checks once every data point of a regular scan has been handed to the hardware worker.

ScanCombination module-attribute 

ScanCombination = frozenset[tuple[str, DatabaseValueType]]

T module-attribute 

T = TypeVar('T')

logger module-attribute 

logger = logging.getLogger(__name__)

timezone module-attribute 

timezone = pytz.timezone(get_config().date.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 
177
178
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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@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 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
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
    self._data_points_to_process: queue.Queue[
        tuple[int, dict[str, DatabaseValueType]]
    ] = queue.Queue()
    self._parameter_dict: dict[str, DatabaseValueType] = {}
    self._scan_progress: ScanProgress = manager.ScanProgress()
    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 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
@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()

            clear_queue(self._data_points_to_process)
            clear_queue(self._outdated_tasks)

            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
                )

                JobRunRepository.update_run_by_id(
                    run_id=pre_processing_task.job_run.id,
                    status=JobRunStatus.DONE,
                    only_if_status=(
                        JobRunStatus.PROCESSING,
                        JobRunStatus.PAUSED,
                    ),
                )

                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
                )

                try_update_run_by_id(
                    run_id=pre_processing_task.job_run.id,
                    status=JobRunStatus.FAILED,
                    log=str(e),
                    only_if_status=(
                        JobRunStatus.PROCESSING,
                        JobRunStatus.PAUSED,
                    ),
                )
            finally:
                JobRepository.update_job_status(
                    job_id=pre_processing_task.job.id, 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 
85
86
87
88
89
90
91
92
93
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)

clear_queue 

clear_queue(q: Queue[T] | Queue[T]) -> None
Source code in src/icon/server/pre_processing/worker.py 
675
676
677
def clear_queue(q: multiprocessing.Queue[T] | queue.Queue[T]) -> None:
    for _ in consume_queue(q):
        pass

consume_queue 

consume_queue(q: Queue[T] | Queue[T]) -> Iterator[T]
Source code in src/icon/server/pre_processing/worker.py 
667
668
669
670
671
672
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 
684
685
686
687
688
689
690
691
692
693
694
695
696
697
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 
680
681
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 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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

is_global_parameter 

is_global_parameter(parameter_id: str) -> bool

Whether a parameter is a global one rather than scoped to an experiment.

Source code in src/icon/server/pre_processing/worker.py 
186
187
188
def is_global_parameter(parameter_id: str) -> bool:
    """Whether a parameter is a global one rather than scoped to an experiment."""
    return GLOBAL_NAMESPACE_SEGMENT in parameter_namespace(parameter_id).split(".")

parameter_namespace 

parameter_namespace(parameter_id: str) -> str

Return the namespace a parameter identifier is scoped to (empty if it has none).

Source code in src/icon/server/pre_processing/worker.py 
181
182
183
def parameter_namespace(parameter_id: str) -> str:
    """Return the namespace a parameter identifier is scoped to (empty if it has none)."""
    return get_specifiers_from_parameter_identifier(parameter_id).get("namespace", "")

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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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 = logging.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 
53
54
55
56
57
58
59
60
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@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:
                    self._dispatch(job_)
                except Exception:
                    logger.exception("Failed to dispatch job %s", 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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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_id=job.id, status=JobStatus.PROCESSED)

should_exit 

should_exit() -> bool
Source code in src/icon/server/scheduler/scheduler.py 
48
49
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
visualiser

Serves the sequence-visualizer build under /visualizer/.

icon_server

Classes:

Name Description
IconServer

Attributes:

Name Type Description
logger

logger module-attribute 

logger = logging.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 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
async def post_startup(self) -> None:
    sio = self._web_server._sio

    _install_device_room_emit(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
device_updates_room

Construct device update room name for device-specific updates. Broadcast room if device_name is None.
log_id
patch_sio_setup
setup_sio_events

Attributes:

Name Type Description
logger
pydase_setup_sio_events

logger module-attribute 

logger = logging.getLogger(__name__)

pydase_setup_sio_events module-attribute 

pydase_setup_sio_events = (
    pydase.server.web_server.sio_setup.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.

device_updates_room 

device_updates_room(device_name: str | None = None) -> str

Construct device update room name for device-specific updates. Broadcast room if device_name is None.

Source code in src/icon/server/web_server/sio_setup.py 
15
16
17
def device_updates_room(device_name: str | None = None) -> str:
    """Construct device update room name for device-specific updates. Broadcast room if device_name is None."""
    return "devices.device_proxies" + (f'["{device_name}"]' if device_name else "")

log_id 

log_id(headers: Any, sid: str) -> str
Source code in src/icon/server/web_server/sio_setup.py 
83
84
85
86
87
88
89
90
91
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 
94
95
96
97
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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})

    _setup_device_update_room_events(sio)

socketio_emit_queue

Classes:

Name Description
EmitEvent

Attributes:

Name Type Description
emit_queue Queue[EmitEvent]

emit_queue module-attribute 

emit_queue: Queue[EmitEvent] = multiprocessing.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]

visualiser

Serves the sequence-visualizer build under /visualizer/.

The files in src/icon/server/frontend_visualizer/ are a build of the ionpulse-sequence-visualiser, produced by the frontend build (see frontend/README.md).

pydase’s WebServer constructs and runs its aiohttp application inside serve() without an extension hook, and its catch-all index route swallows every path. The application object only becomes reachable when serve() hands it to aiohttp.web._run_app, so IconWebServer wraps that call to attach a middleware which serves the visualizer files before route handlers run.

Classes:

Name Description
IconWebServer

pydase WebServer that additionally serves the sequence visualizer.

Functions:

Name Description
patch_web_server

Make pydase.Server instantiate :class:IconWebServer.
visualiser_middleware

Attributes:

Name Type Description
DIST_DIR
URL_PREFIX

DIST_DIR module-attribute 

DIST_DIR = (
    Path(__file__).parent.parent / "frontend_visualizer"
).resolve()

URL_PREFIX module-attribute 

URL_PREFIX = '/visualizer'

IconWebServer

Bases: WebServer

pydase WebServer that additionally serves the sequence visualizer.

Methods:

Name Description
serve
serve async 
serve() -> None
Source code in src/icon/server/web_server/visualiser.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def serve(self) -> None:
    original_run_app = aiohttp.web._run_app

    async def run_app_with_visualiser(
        app: aiohttp.web.Application, **kwargs: Any
    ) -> None:
        app.middlewares.append(visualiser_middleware)
        await original_run_app(app, **kwargs)

    aiohttp.web._run_app = run_app_with_visualiser  # type: ignore[assignment]
    try:
        await super().serve()
    finally:
        aiohttp.web._run_app = original_run_app  # type: ignore[assignment]

patch_web_server 

patch_web_server() -> None

Make pydase.Server instantiate :class:IconWebServer.

Source code in src/icon/server/web_server/visualiser.py 
77
78
79
80
81
def patch_web_server() -> None:
    """Make ``pydase.Server`` instantiate :class:`IconWebServer`."""
    import pydase.server.server  # noqa: PLC0415

    pydase.server.server.WebServer = IconWebServer  # type: ignore[misc]

visualiser_middleware async 

visualiser_middleware(
    request: Request, handler: Handler
) -> StreamResponse
Source code in src/icon/server/web_server/visualiser.py 
44
45
46
47
48
49
50
51
52
53
54
55
@aiohttp.web.middleware
async def visualiser_middleware(
    request: aiohttp.web.Request,
    handler: aiohttp.typedefs.Handler,
) -> aiohttp.web.StreamResponse:
    if request.path == URL_PREFIX:
        # The visualizer is built with relative asset paths ("--base=./"), so
        # it must be served from a URL ending in a slash.
        raise aiohttp.web.HTTPMovedPermanently(f"{URL_PREFIX}/")
    if request.path.startswith(f"{URL_PREFIX}/"):
        return _visualiser_file_response(request.path)
    return await handler(request)