legendmeta package

A package to access legend-metadata in Python.

Submodules

legendmeta.core module

class legendmeta.core.MetadataRepository(path=None, repo_url='', env_var='', default_dir_name='', **kwargs)

Bases: TextDB

Base class for git-based metadata repositories.

Class representing a metadata repository with utilities for fast access via git.

If no valid path to an existing metadata directory is provided, will attempt to clone the repository via SSH and git-checkout the latest stable tag (vM.m.p format).

Note

This class is designed to be subclassed, not instantiated directly. Derived classes should provide the required repository configuration (repo_url, env_var, default_dir_name) via super().__init__().

Parameters:

  • path (str | None) – path to metadata repository. If not existing, will attempt a git-clone through SSH. If None, metadata will be cloned in a temporary directory (see tempfile.gettempdir()).

  • repo_url (str) – URL of the git repository to clone (e.g., “git@github.com:org/repo”).

  • env_var (str) – name of the environment variable to check for repository path.

  • default_dir_name (str) – default directory name for cloning in temp directory.

  • **kwargs – further keyword arguments forwarded to TextDB.__init__.

Construct a TextDB object.

Parameters:

  • path (str | None) – path to the directory containing the database.

  • lazy – whether a database scan should be performed at initialization time. if auto, be non-lazy only if working in a python interactive session.

  • hidden – ignore hidden (i.e. starting with “.”) files of directories.

_except_if_not_git_repo()
_init_metadata_repo()

Clone metadata repository, if not existing, and checkout latest stable tag.

checkout(git_ref, rescan=True)

Select a metadata repository version.

property latest_stable_tag: Version | None

Latest stable metadata tag (i.e. strictly numeric vM.m.p)

show_metadata_version()

Logs version info for metadata repository and all its submodules.

legendmeta.hadesmetadata module

class legendmeta.hadesmetadata.HadesMetadata(path=None, **kwargs)

Bases: MetadataRepository

HADES metadata.

Class representing the HADES metadata repository with utilities for fast access.

If no valid path to an existing hades-metadata directory is provided, will attempt to clone https://github.com/legend-exp/hades-metadata via SSH and git-checkout the latest stable tag (vM.m.p format).

Parameters:

  • path (str | None) – path to hades-metadata repository. If not existing, will attempt a git-clone through SSH. If None, hades-metadata will be cloned in a temporary directory (see tempfile.gettempdir()).

  • **kwargs – further keyword arguments forwarded to TextDB.__init__.

Construct a TextDB object.

Parameters:

  • path (str | None) – path to the directory containing the database.

  • lazy – whether a database scan should be performed at initialization time. if auto, be non-lazy only if working in a python interactive session.

  • hidden – ignore hidden (i.e. starting with “.”) files of directories.

legendmeta.legendmetadata module

class legendmeta.legendmetadata.LegendMetadata(path=None, **kwargs)

Bases: MetadataRepository

LEGEND metadata.

Class representing the LEGEND metadata repository with utilities for fast access.

If no valid path to an existing legend-metadata directory is provided, will attempt to clone https://github.com/legend-exp/legend-metadata via SSH and git-checkout the latest stable tag (vM.m.p format).

Parameters:

  • path (str | None) – path to legend-metadata repository. If not existing, will attempt a git-clone through SSH. If None, legend-metadata will be cloned in a temporary directory (see tempfile.gettempdir()).

  • **kwargs – further keyword arguments forwarded to TextDB.__init__.

Construct a TextDB object.

Parameters:

  • path (str | None) – path to the directory containing the database.

  • lazy – whether a database scan should be performed at initialization time. if auto, be non-lazy only if working in a python interactive session.

  • hidden – ignore hidden (i.e. starting with “.”) files of directories.

channelmap(on=None, system='all', skip_version_check=False)

Get a LEGEND channel map.

Aliases legend-metadata.hardware.configuration.channelmaps.on() and tries to merge the returned channel map with the detector database legend-metadata.hardware.detectors and the analysis channel map dataprod.config.on(…).analysis.

Parameters:

  • on (str | datetime | None) – a datetime object or a string matching the pattern YYYYmmddTHHMMSSZ.

  • system ('all', 'phy', 'cal', 'lar', ...) – query only a data taking “system”.

  • skip_version_check (bool) – if True, skip the git version check and assume the latest metadata structure. This is useful when working with non-git repositories (e.g., test data).

Return type:

AttrsDict

Warning

This method assumes legend-exp/legend-metadata has a certain layout. Might stop working if changes are made to the structure of the repository.

Examples

>>> from legendmeta import LegendMetadata
>>> from datetime import datetime
>>> channel = lmeta.channelmap(on=datetime.now()).V05267B
>>> channel.geometry.mass_in_g
2362.0
>>> channel.analysis.usability
'on'

See also

textdb.TextDB.on

legendmeta.police module

class legendmeta.police._LiteralBlockDumper(stream, default_style=None, default_flow_style=False, canonical=None, indent=None, width=None, allow_unicode=None, line_break=None, encoding=None, explicit_start=None, explicit_end=None, version=None, tags=None, sort_keys=True)

Bases: Dumper

YAML Dumper that preserves multiline strings as literal block scalars.

yaml_representers = {<class 'NoneType'>: <function SafeRepresenter.represent_none>, <class 'bool'>: <function SafeRepresenter.represent_bool>, <class 'builtin_function_or_method'>: <function Representer.represent_name>, <class 'bytes'>: <function SafeRepresenter.represent_binary>, <class 'collections.OrderedDict'>: <function Representer.represent_ordered_dict>, <class 'complex'>: <function Representer.represent_complex>, <class 'datetime.date'>: <function SafeRepresenter.represent_date>, <class 'datetime.datetime'>: <function SafeRepresenter.represent_datetime>, <class 'dict'>: <function SafeRepresenter.represent_dict>, <class 'float'>: <function float_representer>, <class 'function'>: <function Representer.represent_name>, <class 'int'>: <function SafeRepresenter.represent_int>, <class 'list'>: <function SafeRepresenter.represent_list>, <class 'module'>: <function Representer.represent_module>, <class 'set'>: <function SafeRepresenter.represent_set>, <class 'str'>: <function <lambda>>, <class 'tuple'>: <function Representer.represent_tuple>, None: <function SafeRepresenter.represent_undefined>}
legendmeta.police._check_cal_override_runs(file, overridden_runs, verbose=True)

Check that no overridden run appears in a cal_groupings file.

Parameters:

  • file (str) – Path to the cal_groupings YAML file.

  • overridden_runs (set[tuple[str, str]]) – Set of (period, run) pairs that must not appear in the file.

  • verbose (bool) – If False, suppress error output.

Return type:

bool

legendmeta.police._check_chmap_key_name(key, entry, verbose=True)

Return True if entry’s ‘name’ field is absent or matches the dict key.

Return type:

bool

legendmeta.police._check_dataflow_config_paths(state, db_dir, timestamp, category, verbose=True)

Check that all concrete (non-wildcard) path strings in state exist on disk.

Parameters:

  • state (object) – Loaded database state (nested dict/list/str).

  • db_dir (Path) – The directory that $_ was expanded to (the TextDB root).

  • timestamp (str) – Timestamp string used in the query (for error messages).

  • category (str) – Category string used in the query (for error messages).

  • verbose (bool) – If False, suppress error output.

Return type:

bool

legendmeta.police._find_unreferenced_files(validity_file, exclude=None, verbose=True)

Find YAML/JSON data files not transitively referenced via the validity file.

Loads the full resolved database state for every (timestamp, category) pair and collects all file paths that appear as string values with the database directory as prefix (i.e. all $_-expanded references). Files listed directly in apply entries are also considered referenced.

Parameters:

  • validity_file (str) – Path to the validity YAML file.

  • exclude (list[str] | None) – List of glob patterns (relative to the validity file’s directory) to exclude from the unreferenced-file check. Dotfiles and files inside hidden directories are always excluded.

  • verbose (bool) – If True, print error messages.

Returns:

Relative paths of unreferenced data files.

Return type:

list[str]

legendmeta.police._fix_dsp_proc_chain_file(file)

Sort a DSP proc chain file in place. Returns True if the file was modified.

Return type:

bool

legendmeta.police._fix_evt_config_file(file)

Sort an evt config file in place. Returns True if the file was modified.

Return type:

bool

legendmeta.police._fix_groupings_file(file)

Sort a groupings file in place. Returns True if the file was modified.

Return type:

bool

legendmeta.police._fix_hit_config_file(file)

Sort a hit config file in place. Returns True if the file was modified.

Return type:

bool

legendmeta.police._fix_status_files(validity_file)

Sort channel entry keys in all status YAML files beside the validity file.

Returns True if any file was modified.

Return type:

bool

legendmeta.police._get_overridden_runs(run_override_entries, runinfo)

Return (period, run) pairs whose cal falls inside a run_override window.

The run_override file defines windows of time during which certain replacement cal files are applied (non-empty apply list). Any run whose cal.start_key falls within such a window (i.e. the timestamp is ≥ the window start and < the next reset timestamp) is considered overridden. A single window can cover multiple calibrations.

Parameters:

  • run_override_entries (list) – Parsed list of run_override entries (each a dict with ‘valid_from’ and ‘apply’).

  • runinfo (dict) – Parsed runinfo dict mapping period → run → info (with cal.start_key).

Return type:

set[tuple[str, str]]

legendmeta.police._has_invalid_underscore_runs(name)

Return True if name contains underscore runs of length other than 1 or 3.

Return type:

bool

legendmeta.police._iter_string_values(obj)

Yield all string leaf values from a nested dict/list structure.

legendmeta.police._merge_evt_configs(files)

Load and shallow-merge a list of evt config files.

Returns (merged_channels, merged_outputs, merged_operations). Later files take precedence for per-key within each operation.

Return type:

tuple[dict, list[str], dict]

legendmeta.police._needs_reorder(a, b)

Return True if key order differs (recursively) between a and b.

Return type:

bool

legendmeta.police._run_sort_key(spec)

Sort key for a run spec (individual or range): the starting run string.

Return type:

str

legendmeta.police._sort_dsp_proc_chain(data)

Return a copy of a DSP proc chain with top-level and processor keys in canonical order.

Return type:

dict

legendmeta.police._sort_evt_config(data)

Return a copy of an evt config with top-level and operation keys in canonical order.

Return type:

dict

legendmeta.police._sort_evt_operation_entry(op)

Return a copy of an evt operation dict with keys in canonical order.

Return type:

dict

legendmeta.police._sort_groupings_data(data)

Return a copy of groupings data with all keys sorted, default first.

Return type:

dict

legendmeta.police._sort_groupings_groups(groups)

Return a copy of a groups dict with groups and periods/runs sorted.

Return type:

dict

legendmeta.police._sort_hit_config(data)

Return a copy of a hit config with top-level and operation keys in canonical order.

Return type:

dict

legendmeta.police._sort_operation_entry(op)

Return a copy of an operation dict with keys in canonical order.

Return type:

dict

legendmeta.police._sort_processor_entry(proc)

Return a copy of a processor dict with keys in canonical order.

Return type:

dict

legendmeta.police._sort_status_entry(entry)

Return a copy of a channel status entry with keys in canonical order.

Return type:

dict

legendmeta.police._validate_dsp_proc_chain_file(file, verbose=True)

Validate a single tier/dsp proc chain YAML file.

Return type:

bool

legendmeta.police._validate_evt_config_file(file, verbose=True)

Validate a single tier/evt config YAML file.

Checks top-level key ordering, channel definitions, output list, and per-operation key ordering and types. Does not check cross-file field references (see _validate_evt_config_group).

Return type:

bool

legendmeta.police._validate_evt_config_group(files, ts, cat, verbose=True)

Validate a merged group of evt config files for a given timestamp/category.

Checks that all evt.X references are defined in the merged operation set, all outputs (including private ones) are defined operations, and all non-private operations appear in outputs (when any outputs are present).

Return type:

bool

legendmeta.police._validate_groupings_file(file, group_prefix, verbose=True)

Shared validation logic for cal/phy groupings files.

Parameters:

  • file (str) – Path to the groupings YAML file.

  • group_prefix (str) – Expected prefix for group names (e.g. ‘calgroup’ or ‘phygroup’).

  • verbose (bool) – If False, suppress error output.

Return type:

bool

legendmeta.police._validate_hit_config_file(file, verbose=True)

Validate a single tier/hit config YAML file.

Parameters:

  • file (str) – Path to the hit config YAML file.

  • verbose (bool) – If False, suppress error output.

Return type:

bool

legendmeta.police._validate_run_spec(runs, location, verbose=True)

Validate a run specification: string range or list of runs/ranges.

Return type:

bool

legendmeta.police.len_nested(d)

Recursively count keys in a dictionary.

Return type:

int

legendmeta.police.validate_cal_groupings()

Validate LEGEND calibration groupings files.

Invoked in CLI. Expects run_override.yaml and runinfo.yaml to be in the same directory as each groupings file. Checks:

  • default key is present and comes first;

  • non-default top-level keys are lexicographically sorted;

  • group names match calgroupNNNx (e.g. calgroup001a);

  • period names match pNN (e.g. p03), sorted lexicographically;

  • run specifications are rNNN or rNNN..rNNN, sorted within each period;

  • non-default entries do not redundantly replicate the corresponding default;

  • no run whose calibration falls inside a run_override window appears in the file.

legendmeta.police.validate_dataflow_config()

Validate LEGEND dataflow configuration files.

Invoked in CLI. Accepts validity files, tier/hit config files, DSP proc chain files, and tier/evt config files.

For evt config files the hook always resolves the associated validity.yaml (by walking up the directory tree) and performs the merged-group validation from there. This ensures that cross-file evt.X references are checked in the correct overlay context rather than against a single file in isolation.

When given a validity file the hook loads the full config database for every (timestamp, category) pair and additionally scans the sibling tier/hit, tier/dsp, and tier/evt directories. Checks:

  • all concrete path strings in the resolved state exist on disk;

  • hit config: top-level key order outputs → operations → aggregations; operations entries are dicts with optional description / parameters / lgdo_attrs and required expression string; aggregations entries map bitN keys to expression strings;

  • DSP proc chain: top-level key order outputs → processors; processors entries are dicts with known keys in canonical order;

  • evt config (per-file): top-level key order channels → outputs → operations; channels values are channel-ID strings or dicts with a system key; outputs is a list of non-private strings; operations entries are dicts (or null) with known keys, string values where required, and no invalid underscore runs (only _ or ___ allowed) in operation names;

  • evt config (merged group): all evt.X cross-references resolve to a defined operation; every output (including private ones) maps to a defined operation; every non-private operation appears in outputs.

legendmeta.police.validate_dict_schema(adict, template, greedy=True, typecheck=True, root_obj='', verbose=True)

Validate the format of a dictionary based on a template.

Prints error messages on the console. Returns false if dictionary is invalid.

Parameters:

  • adict (dict) – dictionary to analyze.

  • template (dict) – template dictionary.

  • greedy (bool) – if false, do not fail if the analyzed dictionary contains extra keys.

  • typecheck – if true, perform type checking.

  • root_obj (str) – key name (or path to) dictionary. Used for error printing.

  • verbose (bool) – if false, do not print error messages.

Return type:

bool

legendmeta.police.validate_keys_recursive(adict, template, verbose=True)

Return false if adict contains keys not in template.

Return type:

bool

legendmeta.police.validate_legend_channel_map()

Validate list of LEGEND channel map files.

Invoked in CLI. Checks:

  • required system key is present in every channel entry;

  • channel key matches the name field when present;

  • all keys and value types match the per-system YAML template.

Return type:

bool

legendmeta.police.validate_legend_detector_db()

Validate LEGEND detector database.

Invoked in CLI. Checks:

  • required type key is present and has a known value;

  • all keys and value types match the per-type YAML template.

Return type:

bool

legendmeta.police.validate_phy_groupings()

Validate LEGEND physics groupings files.

Invoked in CLI. Checks:

  • default key is present and comes first;

  • non-default top-level keys are lexicographically sorted;

  • group names match phygroupNNNx (e.g. phygroup001a);

  • period names match pNN (e.g. p03), sorted lexicographically;

  • run specifications are rNNN or rNNN..rNNN, sorted within each period;

  • non-default entries do not redundantly replicate the corresponding default.

legendmeta.police.validate_statuses()

Validate LEGEND status files.

Invoked in CLI. Accepts validity files; derives the status directory from each. Checks:

  • every channel key in each snapshot exists in the channel map (when available);

  • Ge detector entries have non-empty usability (one of on, off, ac), boolean processable, non-empty reason when usability != on, and a psd/status dict whose values are one of present, valid, missing;

  • SiPM entries have usability and boolean processable.

legendmeta.police.validate_validity()

Validate LEGEND validity files.

Invoked in CLI. Checks:

  • the file parses as a valid dbetto catalog;

  • every file listed in apply entries exists on disk;

  • no YAML/JSON data files in the same directory are unreferenced.

legendmeta.scdb_tables module

class legendmeta.scdb_tables.Base(**kwargs)

Bases: DeclarativeBase

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>
metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

class legendmeta.scdb_tables.DiodeConfMon(**kwargs)

Bases: Base

Configuration parameters of HPGe detectors.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'confid': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'pwkill': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'pwon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'rdown': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'rup': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'trip': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vmax': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

channel: Mapped[int]
confid: Mapped[int]
crate: Mapped[int]
iset: Mapped[float]
pwkill: Mapped[str]
pwon: Mapped[str]
rdown: Mapped[int]
rup: Mapped[int]
slot: Mapped[int]
trip: Mapped[float]
tstamp: Mapped[datetime]
vmax: Mapped[int]
vset: Mapped[float]
class legendmeta.scdb_tables.DiodeInfo(**kwargs)

Bases: Base

Static information about HPGe detectors.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'group': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'itol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'label': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vtol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

channel: Mapped[int]
crate: Mapped[int]
group: Mapped[str]
itol: Mapped[float]
iupd: Mapped[float]
label: Mapped[str]
slot: Mapped[int]
status: Mapped[int]
tstamp: Mapped[datetime]
vtol: Mapped[float]
vupd: Mapped[float]
class legendmeta.scdb_tables.DiodeSnap(**kwargs)

Bases: Base

Monitored parameters of HPGe detectors.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'almask': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'imon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vmon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
almask: Mapped[int]
asdict()
Return type:

AttrsDict

channel: Mapped[int]
crate: Mapped[int]
imon: Mapped[float]
slot: Mapped[int]
status: Mapped[int]
tstamp: Mapped[datetime]
vmon: Mapped[float]
class legendmeta.scdb_tables.MuonConfMon(**kwargs)

Bases: Base

Configuration parameters of PMTs from the muon veto.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'confid': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'pwkill': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'pwon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'rdown': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'rup': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'trip': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vmax': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

channel: Mapped[int]
confid: Mapped[int]
crate: Mapped[int]
iset: Mapped[float]
pwkill: Mapped[str]
pwon: Mapped[str]
rdown: Mapped[int]
rup: Mapped[int]
slot: Mapped[int]
trip: Mapped[float]
tstamp: Mapped[datetime]
vmax: Mapped[int]
vset: Mapped[float]
class legendmeta.scdb_tables.MuonInfo(**kwargs)

Bases: Base

Static information about PMTs from the muon veto.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'group': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'itol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'label': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vtol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

channel: Mapped[int]
crate: Mapped[int]
group: Mapped[str]
itol: Mapped[float]
iupd: Mapped[float]
label: Mapped[str]
slot: Mapped[int]
status: Mapped[int]
tstamp: Mapped[datetime]
vtol: Mapped[float]
vupd: Mapped[float]
class legendmeta.scdb_tables.MuonSnap(**kwargs)

Bases: Base

Monitored parameters of PMTs from the muon veto.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'almask': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'crate': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'imon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'slot': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vmon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
almask: Mapped[int]
asdict()
Return type:

AttrsDict

channel: Mapped[int]
crate: Mapped[int]
imon: Mapped[float]
slot: Mapped[int]
status: Mapped[int]
tstamp: Mapped[datetime]
vmon: Mapped[float]
class legendmeta.scdb_tables.SiPMConfMon(**kwargs)

Bases: Base

Configuration parameters of SiPMs from the LAr instrumentation.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'board': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'confid': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vset': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

board: Mapped[int]
channel: Mapped[int]
confid: Mapped[int]
iset: Mapped[float]
tstamp: Mapped[datetime]
vset: Mapped[float]
class legendmeta.scdb_tables.SiPMInfo(**kwargs)

Bases: Base

Static information about SiPMs from the LAr instrumentation.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'board': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'group': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'itol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'iupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'label': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vtol': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vupd': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
asdict()
Return type:

AttrsDict

board: Mapped[int]
channel: Mapped[int]
group: Mapped[str]
itol: Mapped[float]
iupd: Mapped[float]
label: Mapped[str]
status: Mapped[int]
tstamp: Mapped[datetime]
vtol: Mapped[float]
vupd: Mapped[float]
class legendmeta.scdb_tables.SiPMSnap(**kwargs)

Bases: Base

Monitored parameters of SiPMs from the LAr instrumentation.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

_sa_class_manager = {'almask': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'board': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'channel': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'imon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'tstamp': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'vmon': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}
almask: Mapped[int]
asdict()
Return type:

AttrsDict

board: Mapped[int]
channel: Mapped[int]
imon: Mapped[float]
status: Mapped[int]
tstamp: Mapped[datetime]
vmon: Mapped[float]

legendmeta.slowcontrol module

class legendmeta.slowcontrol.LegendSlowControlDB(connect=False)

Bases: object

A class for interacting with the LEGEND Slow Control database.

The LEGEND Slow Control system relies on a PostgreSQL database living on legend-sc.lngs.infn.it. The aim of the LegendSlowControlDB class is to simplify access to the database from Python.

connect(host='localhost', port=5432, password=None, db_name='scdb')

Establish a connection to the database.

Authentication is attempted with the read-only user scuser.

Parameters:

  • host (str) – database host. Can be a hostname (localhost, legend-sc.lngs.infn.it, etc.) or an IP address.

  • port (int) – port through which the database should be contacted.

  • password (str | None) – password for user scuser of the scdb database. May be found on LEGEND’s internal documentation (e.g. the Wiki web pages). If None, uses the value of the $LEGEND_SCDB_PW shell variable.

  • db_name (str) – name of the database, default is scdb: scdb is the current database, active since 2024-09-02; scdbL140 is the database for the 142 kg deployment, from 2022-11-09 to 2024-08-28; scdbL60 is for commissioning data with a 60 kg array, from 2021-12-23 to 2022-10-15.

Examples

If the Slow Control database connection is forwarded to a local machine (port 6942) (through e.g. an SSH tunnel), use:

>>> scdb = LegendSlowControlDB()
>>> scdb.connect("localhost", port=6942, password="···")

Alternatively to giving the password to connect(), it can be stored in the $LEGEND_SCDB_PW shell variable (in e.g. .bashrc):

~/.bashrc
export LEGEND_SCDB_PW="···"

Then:

>>> scdb.connect("localhost", port=6942)
dataframe(expr)

Query the database and return a dataframe holding the result.

Parameters:

expr (str | Select) – SQL table name, select SQL command text or SQLAlchemy selectable object.

Return type:

DataFrame

Examples

SQL select syntax text or table name:

>>> scdb.dataframe("SELECT channel, vmon FROM diode_snap LIMIT 3")
   channel    vmon
0        2  2250.0
1        1  3899.4
2        5  1120.2

>>> scdb.dataframe("diode_conf_mon")
      confid  crate  slot  channel    vset  iset  rup  rdown  trip  vmax pwkill pwon                    tstamp
0         15      0     0        0  4000.0   6.0   10      5  10.0  6000   KILL  Dis 2022-10-07 13:49:56+00:00
1         15      0     0        1  4300.0   6.0   10      5  10.0  6000   KILL  Dis 2022-10-07 13:49:56+00:00
2         15      0     0        2  4200.0   6.0   10      5  10.0  6000   KILL  Dis 2022-10-07 13:49:56+00:00
...

sqlalchemy.sql.Select object:

>>> import sqlalchemy as sql
>>> from legendmeta.slowcontrol import DiodeSnap
>>> scdb.dataframe(sql.select(DiodeSnap.channel, DiodeSnap.vmon).limit(3))
   channel    vmon
0        2  2250.0
1        1  3899.4
2        5  1120.2

See also

pandas.read_sql

disconnect()

Disconnect from the database.

get_columns(table)

Get columns available on table in the database.

Return type:

list[str]

get_tables()

Get tables available in the database.

Return type:

list[str]

make_session()

Open and return a new Session object for executing database operations.

Examples

>>> import sqlalchemy as sql
>>> from legendmeta.slowcontrol import DiodeSnap
>>> session = scdb.make_session()
>>> result = session.execute(sql.select(DiodeSnap.channel, DiodeSnap.imon).limit(3))
>>> result.all()
[(2, 0.0007), (1, 0.0001), (5, 5e-05)]

See also

SQLAlchemy documentation

Return type:

Session

status(channel, on=None, system=None)

Query status of a LEGEND DAQ channel.

Collect all the relevant information about the status of a channel at a certain time from the Slow Control database, based on the channel type (i.e. germanium detector, SiPM or PMT).

Parameters:

  • channel (dict) – this dictionary must contain information about the channel hardware configuration. Typically a LEGEND channel map entry (obtainable, for example, with core.LegendMetadata.channelmap()).

  • on (str | datetime | None) – time at which the status is requested.

  • system ("geds", "spms", "pmts", ...) – system the channel belong to. this information is used to ask the Slow Control database the right questions. If None will try to determine it from the available metadata.

Return type:

dict

Examples

>>> ts = datetime(...) # or LEGEND cycle timestamp
>>> my_bege = lmeta.channelmap(on=ts).B00089B
>>> scdb.status(my_bege, on=ts)
{'vmon': 3399.9,
 'imon': 0.00015,
 'status': 1,
 'vset': 3400.0,
 'iset': 6.0,
 ...

Warning

This class method assumes a certain structure for legend-metadata. Might stop working if that structure is altered.

legendmeta.utils module

legendmeta.utils.expand_runs(spec)

Expand a run spec (string or list) to a flat list of individual run strings.

Return type:

list[str]