resonance.api.data.catalog

resonance.api.data.catalog

Catalog(db_path)

Read-only catalog over a per-beamtime SQLite database.

The catalog reads from a .db SQLite file and an adjacent .zarr directory store that holds detector image arrays.

Parameters:

Name Type Description Default
db_path Path

Path to the beamtime SQLite database file. The Zarr store is expected at db_path.with_suffix(".zarr").

required

Attributes:

Name Type Description
_conn Connection

Read-only connection to the database.

_db_path Path

Resolved path passed at construction time.

_zarr_path Path

Path to the adjacent Zarr store directory.

Source code in src/resonance/api/data/catalog.py
def __init__(self, db_path: Path) -> None:
    self._db_path = db_path
    self._zarr_path = db_path.with_suffix(".zarr")
    self._conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    self._conn.row_factory = sqlite3.Row

recent(n=10)

Return the n most recently started runs.

Parameters:

Name Type Description Default
n int

Maximum number of runs to return. Defaults to 10.

10

Returns:

Type Description
list[RunSummary]

Run summaries ordered newest-first.

Source code in src/resonance/api/data/catalog.py
def recent(self, n: int = 10) -> list[RunSummary]:
    """Return the n most recently started runs.

    Parameters
    ----------
    n : int, optional
        Maximum number of runs to return. Defaults to 10.

    Returns
    -------
    list[RunSummary]
        Run summaries ordered newest-first.
    """
    rows = self._conn.execute(_SQL_RECENT, (n,)).fetchall()
    return [_row_to_run_summary(r) for r in rows]

by_sample(name)

Return all runs associated with a sample by name.

Parameters:

Name Type Description Default
name str

Exact sample name as stored in the samples table.

required

Returns:

Type Description
list[RunSummary]

Run summaries for the given sample, ordered newest-first.

Source code in src/resonance/api/data/catalog.py
def by_sample(self, name: str) -> list[RunSummary]:
    """Return all runs associated with a sample by name.

    Parameters
    ----------
    name : str
        Exact sample name as stored in the ``samples`` table.

    Returns
    -------
    list[RunSummary]
        Run summaries for the given sample, ordered newest-first.
    """
    rows = self._conn.execute(_SQL_BY_SAMPLE, (name,)).fetchall()
    return [_row_to_run_summary(r) for r in rows]

__getitem__(uid)

Return the full Run object for the given UID.

Parameters:

Name Type Description Default
uid str

Hex UUID of the run.

required

Returns:

Type Description
Run

Fully-featured run accessor backed by the open connection.

Raises:

Type Description
KeyError

If no run with the given UID exists in the database.

Source code in src/resonance/api/data/catalog.py
def __getitem__(self, uid: str) -> Run:
    """Return the full Run object for the given UID.

    Parameters
    ----------
    uid : str
        Hex UUID of the run.

    Returns
    -------
    Run
        Fully-featured run accessor backed by the open connection.

    Raises
    ------
    KeyError
        If no run with the given UID exists in the database.
    """
    row = self._conn.execute(_SQL_RUN_BY_UID, (uid,)).fetchone()
    if row is None:
        raise KeyError(uid)
    return Run(self._conn, dict(row), zarr_path=self._zarr_path)

close()

Close the underlying database connection.

Source code in src/resonance/api/data/catalog.py
def close(self) -> None:
    """Close the underlying database connection."""
    self._conn.close()

Run(conn, row, zarr_path)

Full accessor for a single run, including scalar event data and sample metadata.

Parameters:

Name Type Description Default
conn Connection

Active connection to the beamtime database.

required
row dict[str, Any]

Deserialized row from the runs table.

required
zarr_path Path

Path to the adjacent Zarr store directory.

required

Attributes:

Name Type Description
_conn Connection

Shared database connection from the parent Catalog.

_row dict[str, Any]

Raw column values for this run.

_zarr_path Path

Path to the Zarr store for detector images.

Source code in src/resonance/api/data/catalog.py
def __init__(
    self, conn: sqlite3.Connection, row: dict[str, Any], zarr_path: Path
) -> None:
    self._conn = conn
    self._row = row
    self._zarr_path = zarr_path

uid property

str: Hex UUID of the run.

plan_name property

str: Name of the scan plan.

time_start property

float: Unix timestamp of run start.

time_stop property

float or None: Unix timestamp of run stop.

exit_status property

str or None: Final run status, e.g. 'success', 'failed'.

num_events property

int: Total number of events recorded in the primary stream.

sample property

Return the associated sample metadata, or None if not set.

Returns:

Type Description
SampleMetadata or None

Populated from the samples table row, with tags and extra deserialized from JSON. Returns None if sample_id is null or the referenced row does not exist.

table(stream='primary')

Load scalar event data from a named stream as a DataFrame.

Parameters:

Name Type Description Default
stream str

Name of the stream to load. Defaults to "primary".

'primary'

Returns:

Type Description
DataFrame

Columns are seq_num, time, followed by all scalar fields present in the event data JSON. Returns an empty DataFrame with columns ["seq_num", "time"] if the stream has no events.

Source code in src/resonance/api/data/catalog.py
def table(self, stream: str = "primary") -> pd.DataFrame:
    """Load scalar event data from a named stream as a DataFrame.

    Parameters
    ----------
    stream : str, optional
        Name of the stream to load. Defaults to "primary".

    Returns
    -------
    pd.DataFrame
        Columns are ``seq_num``, ``time``, followed by all scalar fields
        present in the event ``data`` JSON. Returns an empty DataFrame
        with columns ``["seq_num", "time"]`` if the stream has no events.
    """
    rows = self._conn.execute(_SQL_EVENTS, (self._row["uid"], stream)).fetchall()
    if not rows:
        return pd.DataFrame(columns=pd.Index(["seq_num", "time"]))
    records: list[dict[str, Any]] = []
    for row in rows:
        record: dict[str, Any] = {"seq_num": row["seq_num"], "time": row["time"]}
        record.update(json.loads(row["data"]))
        records.append(record)
    df = pd.DataFrame(records)
    leading = ["seq_num", "time"]
    remaining = [c for c in df.columns if c not in leading]
    return pd.DataFrame(df[leading + remaining])

images(field='detector_image')

Return a lazy accessor for detector images in this run.

Images are not loaded until explicitly indexed. Each frame is stored as a slice of a Zarr array on the filesystem.

Parameters:

Name Type Description Default
field str

The image field name to load (default: "detector_image").

'detector_image'

Returns:

Type Description
LazyImageSequence

Lazy accessor with length equal to the number of images in this run.

Notes

Returns an empty LazyImageSequence if no images are stored for the given field or if the Zarr store does not exist.

Source code in src/resonance/api/data/catalog.py
def images(self, field: str = "detector_image") -> LazyImageSequence:
    """Return a lazy accessor for detector images in this run.

    Images are not loaded until explicitly indexed. Each frame is stored
    as a slice of a Zarr array on the filesystem.

    Parameters
    ----------
    field : str, optional
        The image field name to load (default: "detector_image").

    Returns
    -------
    LazyImageSequence
        Lazy accessor with length equal to the number of images in this run.

    Notes
    -----
    Returns an empty LazyImageSequence if no images are stored for the given
    field or if the Zarr store does not exist.
    """
    rows = self._conn.execute(_SQL_IMAGE_REFS, (self._row["uid"], field)).fetchall()
    refs = [dict(r) for r in rows]
    return LazyImageSequence(self._conn, refs, zarr_store_path=self._zarr_path)

LazyImageSequence(conn, refs, zarr_store_path)

Lazy accessor for detector images referenced via Zarr.

Images are not loaded until explicitly indexed. Each image is stored as a frame in a Zarr array on the filesystem; this class holds the reference metadata and defers loading.

Parameters:

Name Type Description Default
conn Connection

Active connection to the beamtime database.

required
refs list[dict[str, Any]]

List of deserialized rows from the image_refs table.

required
zarr_store_path Path

Path to the Zarr store directory containing detector arrays.

required

Attributes:

Name Type Description
_conn Connection

Shared database connection from the parent Catalog.

_refs list[dict[str, Any]]

Image reference metadata, one entry per frame.

_zarr_store_path Path

Path to the Zarr store directory.

Source code in src/resonance/api/data/catalog.py
def __init__(
    self,
    conn: sqlite3.Connection,
    refs: list[dict[str, Any]],
    zarr_store_path: Path,
) -> None:
    self._conn = conn
    self._refs = refs
    self._zarr_store_path = zarr_store_path

shape property

tuple[int, int, int]: (n_frames, shape_x, shape_y).

__getitem__(idx)

Load one or more detector images from the Zarr store.

Parameters:

Name Type Description Default
idx int or slice

Frame index or slice.

required

Returns:

Type Description
ndarray

A single (shape_x, shape_y) array for int index, or a stacked (n, shape_x, shape_y) array for slice index.

Raises:

Type Description
IndexError

If idx is out of range.

TypeError

If idx is not int or slice.

Source code in src/resonance/api/data/catalog.py
def __getitem__(self, idx: int | slice) -> np.ndarray:
    """Load one or more detector images from the Zarr store.

    Parameters
    ----------
    idx : int or slice
        Frame index or slice.

    Returns
    -------
    np.ndarray
        A single (shape_x, shape_y) array for int index, or a stacked
        (n, shape_x, shape_y) array for slice index.

    Raises
    ------
    IndexError
        If idx is out of range.
    TypeError
        If idx is not int or slice.
    """
    if isinstance(idx, int):
        if idx < -len(self._refs) or idx >= len(self._refs):
            raise IndexError(
                f"index {idx} out of range for LazyImageSequence of length {len(self._refs)}"
            )
        ref = self._refs[idx] if idx >= 0 else self._refs[len(self._refs) + idx]
        store = zarr.open_group(str(self._zarr_store_path), mode="r")
        arr = store[ref["zarr_group"]]
        return np.asarray(arr[ref["index_in_stack"]])
    if isinstance(idx, slice):
        indices = range(*idx.indices(len(self._refs)))
        return (
            np.stack([self[i] for i in indices])
            if indices
            else np.empty((0,), dtype=np.int32)
        )
    raise TypeError(f"indices must be int or slice, not {type(idx).__name__}")