Beamline and Connection¶

The Beamline class is the high-level facade for beamline hardware control. Connection holds BCS server settings from the environment.

Bases: BaseSettings

Connection settings loaded from environment variables.

Reads BCS_SERVER_ADDRESS and BCS_SERVER_PORT from the environment or a .env file. Used by Beamline.create().

Parameters:

addr (str) – BCS server hostname or IP address (env: BCS_SERVER_ADDRESS). Default: “localhost”.
port (int) – BCS server port (env: BCS_SERVER_PORT). Default: 5577.

addr: str¶

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

port: int¶

class resonance.api.core.beamline.Beamline(conn: bcs.BCSz.BCSServer)[source]¶

Bases: object

High-level interface for beamline hardware control.

Composes a BCSz server connection with typed accessors for motors, analog inputs, and digital I/O, plus a scan executor for running DataFrame-defined scans. Does not subclass BCSz.

Parameters:: conn (BCSz.BCSServer) – Connected BCS server instance. Use Beamline.create() to construct with automatic connection from environment variables.

ai¶

Read and acquire analog input channels.

Type:: AIAccessor

motors¶

Read, move, and wait for motors.

Type:: MotorAccessor

dio¶

Read and set digital I/O channels (e.g. shutter).

Type:: DIOAccessor

Examples

>>> bl = await Beamline.create()
>>> data = await bl.ai.trigger_and_read(["Photodiode"], acquisition_time=1.0)
>>> await bl.motors.set("Sample X", 10.5)
>>> results = await bl.scan_from_dataframe(scan_df, ai_channels=["Photodiode"])

async abort_scan() → None[source]¶

Request an abort of the currently running scan.

Sets the internal abort flag. The scan stops after the current point completes and returns partial results.

Notes

Safe to call even when no scan is running (no-op). From Jupyter: run the scan as an asyncio.Task and call this method from another cell while it executes. Programmatic: call from any async context.

async classmethod create() → Beamline[source]¶

Create and connect a Beamline from environment variables.

Reads BCS_SERVER_ADDRESS and BCS_SERVER_PORT from the environment or a .env file, creates a BCSz server, and connects.

Returns:: A connected, ready-to-use Beamline instance.
Return type:: Beamline
Raises:: ConnectionError – If the BCS server is unreachable or connection fails.

property is_scanning: bool¶

Whether a scan is currently running.

Returns:: True if a scan is in progress, False otherwise.
Return type:: bool

async scan_from_dataframe(df: pd.DataFrame, ai_channels: list[str] | None = None, default_delay: float = 0.1, shutter: str = 'Shutter Output', motor_timeout: float = 30.0, progress: bool = True, actuate_every: bool = False, writer: RunWriter | None = None, with_detector: bool = False) → pd.DataFrame[source]¶

Execute a scan defined by a DataFrame.

Each row defines one scan point: motor columns set motor positions, and an optional exposure column sets per-point acquisition time.

Parameters:

df (pd.DataFrame) – Scan definition. Motor columns must match valid motor names. An optional “exposure” (or “exp”, “count_time”) column sets per-point acquisition time.
ai_channels (list[str] or None, optional) – AI channels to acquire at each point. If None, uses [“Photodiode”, “TEY signal”, “AI 3 Izero”].
default_delay (float, optional) – Settle delay after each motor move in seconds (default: 0.1).
shutter (str, optional) – DIO channel name for the shutter (default: “Shutter Output”).
motor_timeout (float, optional) – Maximum wait time for motor moves in seconds (default: 30.0).
progress (bool, optional) – Show a tqdm progress bar (default: True).
actuate_every (bool, optional) – If True, open/close the shutter per point. If False (default), open the shutter once for the entire scan.
writer (RunWriter or None, optional) – If provided, scan data are persisted to the beamtime SQLite database via the writer. The caller is responsible for constructing and opening the writer before passing it here.
with_detector (bool, optional) – If True, acquire a 2D detector image at each scan point using the beamline’s AreaDetector. Requires a writer for image persistence. Shutter actuation is hardware-driven (default: False).

Returns:

Results with columns: motor_position, channel_mean, channel_std, exposure, timestamp per row.

Return type:

pd.DataFrame

Notes

To abort a running scan from another Jupyter cell, call await bl.abort_scan() while the scan task is running.

Examples¶

Creating a beamline from environment variables (BCS_SERVER_ADDRESS, BCS_SERVER_PORT):

bl = await Beamline.create()
data = await bl.ai.trigger_and_read(["Photodiode"], acquisition_time=1.0)
await bl.motors.set("Sample X", 10.5)