pygama.flow package

Routines for performing structured queries on LEGEND data.

Queries selectively access LEGEND data using simple boolean selection expressions to access data from multiple datasources, including metadata and parameter databases, and many tiers of data production. The queries return the data in Tabular formats (ak, pd, np) that can be used to perform further analysis.

They also are designed to access data efficiently and at scale, by (as much as possible) accessing only what is necessary to complete the query, and by breaking the query into chunks that can fit in memory and be operated on in parallel.

Several query commands exist, to access data at different levels:

  • query_runs() accesses all runs in a data production, using information from the cycle names

  • query_meta() accesses channel and run data, grabbing information from the metadata repository, the parameters databases, and the run information from cycle names

  • query_data() accesses event data for select channels and runs, grabbing information from all data production tiers, parameter databases, and run information from cycle names

  • query_hist() accesses event data, as above, and returns a histogram

  • query_evt() accesses evt tier data for a selection of runs; note that this cannot make use of other data tiers or of metadata (yet…)

  • build_iterator() creates a LH5Iterator object to load a selection of data fields and metadata for a selection of channels and runs. This can be used to perform more advanced queries if needed.

The information about the data production required to perform these queries can be accessed from its dataflow-config.yaml file. The information needed comes from the paths used for dataflow, and a set of query parameters (which are arguments of the above set of functions). Note that the dataflow-config.yaml need not come from an existing production; one can be modified to point towards additional paths as needed!

A template for a minimal dataflow-config.yaml file:
paths:
    metadata: $_/inputs # path to metadata root dir; usually $REFPROD/inputs

    tier_raw: $_/generated/tier/raw # path to raw tier root dir
    tier_tla: $_/generated/tier/... # path to root dir for tier named TLA
    ...

    par_tla: $_/generated/par/... # path to root dir for pargen database
    ...

query:
    cycle_def: experiment-period-run-datatype-starttime # REQUIRED: hyphen-separated-list-of-fields-in-cycle-name; these will be columns of run db
    metadata: LegendMetadata # REQUIRED: name of metadata class (e.g. LegendMetadata)
    ignored_cycles: dataprod/config/ignored_cycles # path in metadata to list of cycles to skip; by default do not ignore any
    tiers: ["raw", "dsp", "hit", "tcm", "evt"] # list of tiers TLAs to use from paths for parameter and data queries; default use all

    tables: # mapping from tier names to table paths in lh5 files
        raw: ch{@chan.daq.rawid:07d}/raw
        evt: evt
        tla: # path to table for channel in tier_tla lh5 files. Use format string syntax, which can refer to values from the run_db and chan_db; e.g. "ch{@chan.daq.rawid}/raw"
        ...

    evt_tiers: ["tcm", "evt"] # list of tiers to use for event-level queries

    evt_tables: # mapping from event-level tier names to table paths in lh5 files
        evt: evt
        tla: # path to table for channel in tier_tla lh5 files. Do not use format string syntax!

    chan_db: # path in metadata to list of channels for a given run. Use format string syntax, which may refer to any values in the run DB (i.e. cycle_def fields, cycle name, and relative path). If no value was provided, call "metadata.channelmap(on = starttime)", where "starttime" is drawn from the run db
    par_db: #optional info for navigating par dbs. If missing use "par_db.on(starttime)[@chan.name]"
        cycle_entry: # sub-path to entry for cycle, using format string syntax, which may include values from run db; if missing or falsey, call .on
        chan_entry: "@chan.name" # sub-sub-path to entry for channel, using format string syntax, which may include values from run or chan dbs; if missing or falsey, assume same values for all channels in a cycle (useful if only one channel per cycle...)

    meta_dbs: # optional list of metadata DBs
        name: # @name of db for access in query_meta
            path: # metadata path to database
            cycle_entry: # sub-path to entry for cycle, using format string syntax, which may include values from run db; if missing or falsey, call .on
            chan_entry: # sub-sub-path to entry for channel, using format string syntax, which may include values from run or chan dbs; if missing or falsey, use same value for all chans
            # based on these, we will search "metadata["[path][.on(starttime)|/cycle_entry][/chan_entry]
        ...

These dataflow-config.yaml files should not often need to be created, and will be provided with most dataprods! If you set the environment variable $REFPROD then the file will automatically be accessed from the referenced directory!

Submodules

pygama.flow.build_iterator module

pygama.flow.build_iterator.build_iterator(fields, runs, channels, *, dataflow_config='$REFPROD/dataflow-config.yaml', tiers=None, tables=None, return_alias_map=False, progress=True, **query_meta_kwargs)

build a :class:LH5Iterator to access data across multiple tiers and databases.

Parameters:

  • fields (Collection[str]) – fields to include (across all tiers)

  • runs (str | Array | Mapping[str, ndarray] | DataFrame) –

    python boolean expression for selecting runs, using column names defined in cycle_def as variables. See query_runs()

    Examples:

    • "period>='p06' and period<='p08' and datatype=='cal'" selects calibration data from periods 6, 7 and 8 (assuming default cycle names)

    • "det in ['V01234A', 'V06789B'] and datatype=='th_HS2_lat_psa'" selects runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime)

  • channels (str | Array | Mapping[str, ndarray] | DataFrame) –

    expression used to select channels for each run. Expression can access values from all databases, as well as the run table.

    Examples:

    • "@chan.system=='geds' and @chan.type=='icpc' and @chan.analysis.usability=='on'" selects all ICPC detectors for each run that are marked as usable

    • "@chan.name=='S010' and @chan.analysis.processible" selects SiPM channel 10 and will only include runs where it is can be processed

    Note: if a parameter does not exist for a channel, it will evaluate to None. If this causes an error to be thrown, this expression will evaluate to False, excluding the channel. If an parameter always evaluates to False, it will raise an Exception.

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • tiers (Collection[str]) – tiers to include

  • tables (Mapping[str, str]) – mapping of tiers to format strings to access tables. Format strings may reference values from run or channel DBs. If no channel-wise information is included in the string the same table will be accessed for each channel (may be useful for evt tier). If None, read from dataflow_config. This is required.

  • return_alias_map (bool) – if True, return the pair (table, alias_map) where table is the normal output of this function and alias_map is a mapping from alias names to database paths

  • progress (Status | Console | bool) – if True draw progress spinner; can also provide a rich.Status or:class:rich.Console

  • query_meta_kwargs – additional keyword arguments for query_meta() and query_runs()

pygama.flow.data_loader module

Routines for high-level data loading and skimming.

class pygama.flow.data_loader.DataLoader(config, filedb=None, file_query=None)

Bases: object

Facilitate loading of processed data across several tiers.

Where possible, uses a FileDB object so that a user can quickly select a subset of cycle files for interest, and access information at each processing tier.

Example JSON configuration file:

{
    "filedb": "path/to/filedb.h5"
    "levels": {
        "hit": {
            "tiers": ["raw", "dsp", "hit"]
        },
        "tcm": {
            "tiers": ["tcm"],
            "parent": "hit",
            "child": "evt",
            "tcm_cols": {
                "child_idx": "coin_idx",
                "parent_tb": "table_key",
                "parent_idx": "row_in_table"
            }
        },
        "evt": {
            "tiers": ["evt"]
        }
    }
}

Examples

>>> from pygama.flow import DataLoader
>>> dl = DataLoader("loader-config.json")
>>> dl.set_files("file_status == 26 and timestamp == '20220716T130443Z'")
>>> dl.set_datastreams([3, 6, 8], "ch")
>>> dl.set_cuts({"hit": "daqenergy > 1000 and AoE > 3", "evt": "muon_veto == False"})
>>> dl.set_output(fmt="pd.DataFrame", columns=["daqenergy", "channel"])
>>> data = dl.load()

Be careful, load() loads data in memory regardless of its size. If loading a lot of data (e.g. waveforms), you might want to do it in chunks. next() does exactly this:

>>> for chunk in dl.load():
...   run_my_processing(chunk)

Advanced Usage:

>>> from pygama.flow import DataLoader
>>> dl = DataLoader("loader-config.json", filedb="filedb-config.json")  # or any value accepted by the FileDB constructor
>>> dl.set_files("all")
>>> dl.set_datastreams([0], "ch")
>>> dl.set_cuts({"hit": "wf_max > 30000"})
>>> el = dl.build_entry_list(tcm_level="tcm", mode="any")
>>> el.query("hit_table == 20", inplace=True)
>>> dl.set_output(fmt="pd.DataFrame", columns=["daqenergy", "channel"])
>>> data = dl.load(el)
Parameters:

  • config (str | dict) –

    configuration dictionary or JSON file, see above for a specification. Accepts strings in the following format:

    path/to/config.json[field1/field2/...]]

    to specify the location of the DataLoader configuration in the config.json dictionary, if not at the first level.

  • filedb (str | dict | FileDB) –

    the loader needs a file database. It can be specified in multiple ways:

    If None, uses the value of the filedb key in config to instantiate a FileDB object.

  • file_query (str) – string query that should operate on columns of a FileDB.

Note

No data is loaded in memory at this point.

browse(entry_list=None, buffer_len=128, **kwargs)

Return a WaveformBrowser object for waveform inspection.

Parameters:

  • entry_list (DataFrame) – the output of build_entry_list(). If None, builds it according to the current configuration.

  • buffer_len (int) – number of waveforms to keep in memory at a time.

  • **kwargs – keyword arguments forwarded to WaveformBrowser.

Return type:

WaveformBrowser

See also

WaveformBrowser

build_entry_list(tcm_level=None, tcm_table=None, mode='only', save_output_columns=False, in_memory=True, output_file=None)

Applies cuts to the tables and files of interest.

Can only load up to two levels, those joined by tcm_level.

Parameters:

  • tcm_level (str) – the type of TCM to be used. If None, will only return information from lowest level.

  • tcm_table (int | str) – the identifier of the table inside this TCM level that you want to use. If unspecified, there must only be one table inside a TCM file in tcm_level.

  • mode (str) – if any, returns every hit in the event if any hit in the event passes the cuts. If only, only returns hits that pass the cuts.

  • save_output_columns (bool) – if True, saves any columns needed for both the cut and the output to the self.entry_list.

  • in_memory (bool) – if True, returns the generated entry list in memory.

  • output_file (str) – HDF5 file name to write the entry list to.

Returns:

entries – the entry list containing columns for {parent}_idx, {parent}_table, {child}_idx and output columns if applicable. Only returned if in_memory is True.

Return type:

dict[int, DataFrame] | DataFrame | None

Note

Does not load the column information into memory. This is done by load().

build_hit_entries(save_output_columns=False, in_memory=True, output_file=None)

Called by build_entry_list() to handle the case when tcm_level is unspecified.

Ignores any cuts set on levels above lowest level.

Parameters:

  • save_output_columns (bool) – If True, saves any columns needed for both the cut and the output to the entry list.

  • in_memory (bool) – If True, returns the generated entry list in memory.

  • output_file (str) – HDF5 file name to write the entry list to.

Returns:

entries – the entry list containing columns for {low_level}_idx, {low_level}_table, and output columns if applicable. Only returned if in_memory is True.

Return type:

dict[int, DataFrame] | DataFrame | None

get_file_list()

Returns a copy of the file database with its dataframe pared down to the current file list.

Return type:

DataFrame

get_tiers_for_col(columns, merge_files=None)

For each column given, get the tiers and tables in that tier where that column can be found.

Parameters:

columns (list | ndarray) – the columns to look for.

Returns:

col_tiers – col_tiers[file]["tables"][tier] gives a list of tables in tier that contain a column of interest. col_tiers[file]["columns"][column] gives the tier that column can be found in. If self.merge_file`s then ` col_tiers[tier]` is a list of tables in tier that contain a column of interest.

Return type:

dict

load(entry_list=None, in_memory=True, output_file=None, orientation='hit', tcm_level=None)

Loads the requested data from disk.

Loads the requested columns in self.output_columns for the entries in the given entry_list.

Parameters:

  • entry_list (DataFrame) – the output of build_entry_list(). If None, builds it according to the current configuration.

  • in_memory (bool) – if True, returns the loaded data in memory and stores in self.data.

  • output_file (str) – if not None, writes the loaded data to the specified file.

  • orientation (str) – specifies the orientation of the output table. Can be hit or evt.

  • tcm_level (str) – which TCM was used to create the entry_list.

Returns:

data – The data loaded from disk, as specified by self.output_format, self.output_columns, and self.merge_files. Only returned if in_memory is True.

Return type:

None | Table | Struct | DataFrame

load_cal_pars(query)

access the cal_pars parameter database, run a query, and return some tables.

load_detector(det_id)

special version of load designed to retrieve all file files, tables, column names, and potentially calibration/dsp parameters relevant to one single detector.

load_dsp_pars(query)

access the dsp_pars parameter database (probably JSON format) and do some kind of query to retrieve parameters of interest for our file list, and return some tables.

load_evts(entry_list=None, in_memory=False, output_file=None, tcm_level=None)

Called by load() when orientation is evt.

Return type:

None | Table | Struct | DataFrame

load_hits(entry_list, in_memory=False, output_file=None, tcm_level=None)

Called by load() when orientation is hit.

Return type:

None | Table | Struct | DataFrame

load_iterator(entry_list=None, tcm_level=None, buffer_len=3200)

Creates an :class:LH5Iterator that will load the requested columns in self.output_columns for the entries in the given entry_list in chunks. This is more memory efficient than filling a whole table and is recommended for use when loading waveforms.

Parameters:

  • entry_list (DataFrame) – the output of build_entry_list(). If None, builds it according to the current configuration.

  • tcm_level (str) – which TCM was used to create the entry_list.

  • buffer_len (int) – how many entries to load in a single chunk

Returns:

data – LH5 Iterator, which yields (lh5 table, entry, n_entries) when iterated over.

Return type:

LH5Iterator

load_settings()

get metadata stored in raw files, usually from a DAQ machine.

next(entry_list=None, chunk_size=10000, **kwargs)

Loads the requested data from disk in chunks.

This method should be used instead of load() to handle large data sets.

Note

It is a user responsibility to optimize the chunk size in order to achieve best performance.

Parameters:

  • chunk_size (int) – number of entries to load at each iteration. Adapt based on the size of each entry and the amount of memory available on the system.

  • entry_list (DataFrame) – keyword argument forwarded to load().

  • **kwargs – keyword argument forwarded to load().

Returns:

data – see load().

Return type:

Iterator[Table | Struct | DataFrame]

Examples

>>> for chunk in dl.next():
>>>    # 'chunk' has the same type of the output of dl.load()

See also

load

reset()

Resets all fields to their default values.

As if this is a newly created data loader.

set_config(config)

Load configuration dictionary.

$_ expands to the config file location, if possible, otherwise the current working directory.

set_cuts(cuts, append=False)

Apply a selection on columns in the data tables.

Parameters:

  • cut – the cuts on the columns of the data table, e.g. trapEftp_cal > 1000. If passing a dictionary, the dictionary should be structured as dict[tier] = cut_expr. If passing a list, each item in the array should be able to be applied on one level of tables. The cuts at different levels will be joined with an AND.

  • append (bool) – if True, appends cuts to the existing cuts instead of overwriting

Example

>>> dl.set_cuts({"raw": "daqenergy > 1000", "hit": "AoE > 3"})
set_datastreams(ds, word, append=False)

Apply selection on data streams (or channels).

Sets self.table_list.

Parameters:

  • ds (list | tuple | ndarray) – identifies the detectors of interest. Can be a list of detector names, serial numbers, or channels or a list of subsystems of interest e.g. ged.

  • word (str) – the type of identifier used in ds. Should be a key in the given channel map or a word defined in the configuration file.

  • append (bool) – if True, appends datastreams to the existing self.table_list instead of overwriting.

Example

>>> dl.set_datastreams(np.arange(40, 45), "ch")
set_files(query, append=False)

Apply a file selection.

Sets self.file_list, which is a list of indices corresponding to the rows in the file database.

Parameters:

  • query (str | list[str]) – if single string, defines an operation on the file database columns supported by pandas.DataFrame.query(). In addition, the all keyword is supported to select all files in the database. If list of strings, will be interpreted as key (cycle timestamp) list.

  • append (bool) – if True, appends files to the existing self.file_list instead of overwriting.

Note

Call this function before any other operation. A second call to set_files() does not replace the current file list, which gets instead integrated with the new list. Use reset() to reset the file query.

Example

>>> dl.set_files("file_status == 26 and timestamp == '20220716T130443Z'")
set_output(fmt=None, merge_files=None, columns=None, aoesa_to_vov=None)

Set the parameters for the output format of load

Parameters:

  • fmt (str) – lgdo.Table or pd.DataFrame.

  • merge_files (bool) – if True, information from multiple files will be merged into one table.

  • columns (list) – the columns that should be copied into the output.

  • aoesa_to_vov (bool) – output ArrayOfEqualSizedArrays as VectorOfVectors.

Example

>>> dl.set_output(
...   fmt="pd.DataFrame",
...   merge_files=False,
...   columns=["daqenergy", "trapEmax", "channel"]
... )
skim_waveforms(mode='hit', hit_list=None, evt_list=None)

handle this one separately because waveforms can easily fill up memory.

pygama.flow.data_loader.iskeyword()

x.__contains__(y) <==> y in x.

pygama.flow.file_db module

Utilities for LH5 file inventory.

class pygama.flow.file_db.FileDB(config, scan=True)

Bases: object

LH5 file database.

A class containing a pandas.DataFrame that has additional functions to scan the data directory, fill the dataframe’s columns with information about each file, and read or write to disk in an LGDO format.

The database contains the following columns:

  • file keys: the fields specified in the configuration file’s file_format that are required to generate a file name e.g. run, type, timestamp etc.

  • {tier}_file: generated file name for the tier.

  • {tier}_size: size of file on disk, if applicable.

  • file_status: contains a bit corresponding to whether or not a file for each tier exists for a given cycle e.g. If we have tiers raw, dsp, and hit, but only the raw file has been produced, file_status would be 0b100.

  • {tier}_tables: available data streams (channels) in the tier.

  • {tier}_col_idx: file_db.columns[{tier}_col_idx] will return the list of columns available in the tier’s file.

The database must be configured by a JSON file (or corresponding dictionary), which defines the data file names, paths and LH5 layout. For example:

{
    "data_dir": "prod-ref-l200/generated/tier",
    "tier_dirs": {
        "raw": "/raw",
        "dsp": "/dsp",
        "hit": "/hit",
        "tcm": "/tcm",
        "evt": "/evt"
    },
    "file_format": {
        "raw": "/{type}/{period}/{run}/{exp}-{period}-{run}-{type}-{timestamp}-tier_raw.lh5",
        "dsp": "/{type}/{period}/{run}/{exp}-{period}-{run}-{type}-{timestamp}-tier_dsp.lh5",
        "hit": "/{type}/{period}/{run}/{exp}-{period}-{run}-{type}-{timestamp}-tier_hit.lh5",
        "evt": "/{type}/{period}/{run}/{exp}-{period}-{run}-{type}-{timestamp}-tier_evt.lh5",
        "tcm": "/{type}/{period}/{run}/{exp}-{period}-{run}-{type}-{timestamp}-tier_tcm.lh5"
    },
    "table_format": {
        "raw": "ch{ch:03d}/raw",
        "dsp": "ch{ch:03d}/dsp",
        "hit": "{ch}/hit",
        "evt": "{grp}/evt",
        "tcm": "hardware_tcm"
    },
    "tables": {
        "raw": [0, 1, 2, 4, 5, 6, 7],
        "dsp": [0, 1, 2, 4, 5, 6, 7],
        "hit": [0, 1, 2, 4, 5, 6, 7],
        "tcm": [""],
        "evt": [""]
    },
    "columns": {
        "raw": ["baseline", "waveform", "daqenergy"],
        "dsp": ["trapEftp", "AoE", "trapEmax"],
        "hit": ["trapEftp_cal", "trapEmax_cal"],
        "tcm": ["table_key", "row_in_table"],
        "evt": ["lar_veto", "muon_veto", "ge_mult"]
    }
}

FileDB objects can be also stored on disk and read-in at later times.

Examples

>>> from pygama.flow import FileDB
>>> db = FileDB("./filedb_config.json")
>>> db.scan_tables_columns()  # read in also table columns names
>>> print(db)
<< Columns >>
[['baseline', 'card', 'ch_orca', 'channel', 'crate', 'daqenergy', 'deadtime', 'dr_maxticks', 'dr_start_pps', 'dr_start_ticks', 'dr_stop_pps', 'dr_stop_ticks', 'eventnumber', 'fcid', 'numtraces', 'packet_id', 'runtime', 'timestamp', 'to_abs_mu_usec', 'to_dt_mu_usec', 'to_master_sec', 'to_mu_sec', 'to_mu_usec', 'to_start_sec', 'to_start_usec', 'tracelist', 'ts_maxticks', 'ts_pps', 'ts_ticks', 'waveform'], ['bl_intercept', 'bl_mean', 'bl_slope', 'bl_std', 'tail_slope', 'tail_std', 'wf_blsub'], ['table_key', 'row_in_table', 'cumulative_length']]
<< DataFrame >>
   exp period   run         timestamp type  ... hit_col_idx tcm_tables tcm_col_idx evt_tables evt_col_idx
0  l60    p01  r014  20220716T105236Z  cal  ...        None         []         [2]       None        None
1  l60    p01  r014  20220716T104550Z  cal  ...        None         []         [2]       None        None
>>> db.to_disk("file_db.lh5")
Parameters:

  • config (str | dict | list[str]) – dictionary or path to JSON file specifying data directories, tiers, and file name templates. Can also be path (or list of paths or regular expression) to existing LH5 file containing FileDB object serialized by to_disk().

  • scan (bool) – whether the file database should scan the directory containing raw files to fill its rows with file keys.

from_disk(path)

Read FileDBs from disk.

Overrides the dataframe, configuration dictionary and columns with the information from a file created by to_disk().

Parameters:

path (str | list[str]) – file or file pattern (or list of the latter).

get_table_columns(table, tier, ifile=0)

Return list of columns in table table, tier tier.

Assumes that the table contents do not change across data files. If desired, ifile (default is 0) can be used to select a different file.

Return type:

list[str]

get_table_name(tier, tb)

Get the table name for a tier given its table identifier.

Parameters:

  • tier (str) – specify the tier whose table format will be used.

  • tb (str) – the table identifier that will be passed to the table format.

Returns:

table_name – the name of the table in tier with table identifier tb

Return type:

str

scan_daq_files(daq_dir, daq_template)

Does the exact same thing as scan_files() but with extra configuration arguments for a DAQ directory and template instead of using the lowest tier.

scan_files(dirs=None)

Scan the directory containing files from the lowest tier and fill the dataframe.

The lowest tier is defined as the first element of the tiers array. Only fills columns that can be populated with just these files.

Parameters:

dirs (list[str]) – restrict search to this list of directories. Specified paths can be absolute, relative to self.data_dir or relative to the root directory of the lowest-tier files. If None, the whole root lowest-tier directory is scanned. Useful to build a partial database.

scan_tables_columns(to_file=None, override=False, dir_files_conform=False)

Open files to read (and store) available tables (and columns therein) names.

Adds the available table names in each tier as a column in the dataframe by searching for group names that match the configured table_format and saving the associated keyword values.

Returns a list with each unique list of columns found in each table and adds a column {tier}_col_idx to the dataframe that maps to the column table.

Parameters:

  • to_file (str) – Optionally write the column table to an LH5 file (as a VectorOfVectors).

  • override (bool) – If the FileDB already has a columns field, the scan will not run unless this parameter is set to True.

  • dir_files_conform (bool) – if True, assume that all files in a directory contain tables with the same columns (i.e. all file contents conform to the same format) and scan only the first file. Significantly reduces processing time.

Return type:

list[str]

set_config(config, config_path=None)

Read in the configuration dictionary.

set_file_sizes()

Add columns for each tier containing the corresponding file size in bytes.

As reported by os.path.getsize().

set_file_status()

Add a column with a bit corresponding to whether each tier’s file exists.

For example, if we have tiers raw, dsp, and hit, but only the raw file has been produced, file_status would be 4 (0b100 in binary representation).

to_disk(filename, wo_mode='write_safe')

Serializes database to disk.

Parameters:

  • filename (str) – output LH5 file name.

  • wo_mode – passed to write().

pygama.flow.query_data module

pygama.flow.query_data.query_data(fields, runs, channels, entries, *, dataflow_config='$REFPROD/dataflow-config.yaml', return_query_vals=False, return_alias_map=False, processes=None, executor=None, library=None, progress=True, **kwargs)

Query data from multiple tiers and metadata. Return a table containing one entry for each hit corresponding to the selected runs, channels, and data cuts, with columns for the requested data fields. Selections may be based on data fields in any tier of data file, in select metadata tables, in the parameters databases, or in the run descriptions. Values will be returned in a tabular format denoted by library (default awkward.Array). Values from metadata and parameters databases are accessed using (see :meth:query_meta):

[alias]@db_name.par_path

In addition, parameters from data tables may be optionally aliased using:

[alias]:nested.par_name

If no alias is provided, then the on-disk name will be used; if a parameter is nested, then _ will be used to separate levels (in the above example, the default alias would be nested_par_name)

Parameters:

  • fields (Collection[str]) – list of fields to include in the table. May include fields accessible with :meth:query_runs, :meth:query_meta, and fields in any data tier accessible by this method. See above for aliasing rules.

  • runs (str | Array | DataFrame) –

    python boolean expression for selecting runs, using column names defined in cycle_def as variables. See :meth:query_runs

    Examples:

    • select calibration data from periods 6, 7 and 8 (assuming default cycle names):

      "period>='p06' and period<='p08' and datatype=='cal'"

    • select runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime):

      "det in ["V01234A", "V06789B"] and datatype=='th_HS2_lat_psa'``

  • channels (str) –

    expression used to select channels for each run. Expression can access values from all databases, as well as the run table.

    Examples:

    • select all ICPC detectors for each run that are marked as usable:

      "@chan.system=='geds' and @chan.type=='icpc' and @chan.analysis.usability=='on'"``

    • selects SiPM channel 10 and will only include runs where it is can be processed:

      "@chan.name=='S010' and @chan.analysis.processible"

    Note: if a parameter does not exist for a channel, it will evaluate to None. If this causes an error to be thrown, this expression will evaluate to False, excluding the channel. If an parameter always evaluates to False, it will raise an Exception.

  • entries (str) –

    expression used to select data entries for each run/channel. Expression can access values from any data tier, from all databases, and from run table. Parameters with aliases can be accessed using their on-disk field name or their alias.

    Examples:

    • select events with >100 keV of energy, with various event-level cuts applied:

      "(energy > 100) & (~coincident.puls) & (~coincident.spms) & (geds.multiplicity==1) & ak.all(geds.quality.is_bb_like, axis=-1)"

    • select hits with >500 keV of energy and manually applies the low A/E cut:

      ``"(cuspEmax_ctc_cal > 500) & (AoE_classifyer > @pars.pars.operations.AoE_Low_Cut.parameters.a)"``

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • return_query_vals (bool) – if True, return values found in query as columns; else only return those in fields

  • return_alias_map (bool) – if True, return the pair (table, alias_map) where table is the normal output of this function and alias_map is a mapping from alias names to database paths

  • processes (Executor | int) – number of processes. If None, use number equal to threads available to executor (if provided), or else do not parallelize

  • executor (Executor) – concurrent.futures.Executor object for managing parallelism. If None, create a concurrent.futures.`ProcessPoolExecutor with number of processes equal to processes.

  • library (str) – format of returned table. Can be ak (default), pd or np

  • progress (Status | Console | bool) – if True draw progress bar; can also provide a rich.Status or:class:rich.Console

  • kwargs – see build_iterator(), query_meta() and query_runs()

pygama.flow.query_evt module

pygama.flow.query_evt.query_evt(fields, runs, events, *, dataflow_config='$REFPROD/dataflow-config.yaml', tiers=None, tables=None, return_query_vals=False, processes=None, executor=None, library=None, progress=True, **kwargs)

Query evt tier data. Return a table containing one entry for each event corresponding to the selected runs and data cuts, with columns for the requested data fields. Selections may be based on data fields in the evt tier or in the run descriptions. Values will be returned in a tabular format denoted by library (default awkward.Array). Parameters may be optionally aliased using:

[alias]:nested.par_name

If no alias is provided, then the on-disk name will be used.

Parameters:

  • fields (Collection[str]) – list of fields to include in the table. May include fields accessible with :meth:query_runs, :meth:query_meta, and fields in any data tier accessible by this method. See above for aliasing rules.

  • runs (str | Array | Mapping[str, ndarray] | DataFrame | None) –

    python boolean expression for selecting runs, using column names defined in cycle_def as variables. See :meth:query_runs

    Examples:

    • select calibration data from periods 6, 7 and 8 (assuming default cycle names):

      "period>='p06' and period<='p08' and datatype=='cal'"

    • select runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime):

      "det in ["V01234A", "V06789B"] and datatype=='th_HS2_lat_psa'``

  • events (str) –

    expression used to select data events for each run/channel. Expression can access values from the event tier. Parameters with aliases can be accessed using their on-disk field name or their alias. Awkward is available using ak.

    Examples:

    • select events with >100 keV of energy, with various event-level cuts applied:

      "(energy > 100) & (~coincident.puls) & (~coincident.spms) & (geds.multiplicity==1) & ak.all(geds.quality.is_bb_like, axis=-1)"

    • select hits with >500 keV of energy and manually applies the low A/E cut:

      ``"(cuspEmax_ctc_cal > 500) & (AoE_classifyer > @pars.pars.operations.AoE_Low_Cut.parameters.a)"``

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • tiers (Collection[str]) – tiers to include

  • tables (Mapping[str, str]) – mapping of tiers to format strings to access tables. Format strings may reference values from run or channel DBs. If no channel-wise information is included in the string the same table will be accessed for each channel (may be useful for evt tier). If None, read from dataflow_config. This is required.

  • return_query_vals (bool) – if True, return values found in query as columns; else only return those in fields

  • processes (Executor | int) – number of processes. If None, use number equal to threads available to executor (if provided), or else do not parallelize

  • executor (Executor) – concurrent.futures.Executor object for managing parallelism. If None, create a concurrent.futures.`ProcessPoolExecutor with number of processes equal to processes.

  • library (str) – format of returned table. Can be ak (default), pd or np

  • progress (Status | Console | bool) – if True draw progress bar; can also provide a rich.Status or:class:rich.Console

  • kwargs – see query_runs()

pygama.flow.query_hist module

pygama.flow.query_hist.query_hist(axes, runs, channels, entries, *, dataflow_config='$REFPROD/dataflow-config.yaml', processes=None, executor=None, progress=True, **kwargs)

Query data from multiple tiers and metadata. Return a Hist filled with data from hits from selected runs, channels, and data cuts. Values from metadata and parameters databases are accessed using (see :meth:query_meta):

[alias]@db_name.par_path

In addition, parameters from data tables may be optionally aliased using:

[alias]:nested.par_name

If no alias is provided, then the on-disk name will be used; if a parameter is nested, then _ will be used to separate levels (in the above example, the default alias would be nested_par_name)

Parameters:

  • axes (Collection[<module 'hist.axis' from '/home/docs/checkouts/readthedocs.org/user_builds/pygama/envs/v2.6.2/lib/python3.12/site-packages/hist/axis/__init__.py'>] | Mapping[str, <module 'hist.axis' from '/home/docs/checkouts/readthedocs.org/user_builds/pygama/envs/v2.6.2/lib/python3.12/site-packages/hist/axis/__init__.py'>]) –

    axis, list of axes, or mapping from data field to axis to use for histogram. If axis or list of axes, use axis.name for the field. May include fields accessible with :meth:query_runs, :meth:query_meta, and fields in any data tier accessible by this method. See above for aliasing rules; if axis has no label, alias will be used

    Examples:

    • Energy histogram (300 bins ranging from 0 to 3000):

      axis.Regular(300, 0, 3000, name="cuspEmax_ctc_cal", label="Energy (keV)")

    • 2-D histogram with energy on x-axis, and detector name on y-axis:

      {
    "cuspEmax_ctc_cal": axis.Regular(300, 0, 3000, label="Energy (keV)"),
    "@chan.name": axis.StrCategory(label="Detector", growth=True)"
}

  • runs (str | Array | DataFrame) –

    python boolean expression for selecting runs, using column names defined in cycle_def as variables. See :meth:query_runs

    Examples:

    • select calibration data from periods 6, 7 and 8 (assuming default cycle names):

      "period>='p06' and period<='p08' and datatype=='cal'"

    • select runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime):

      "det in ["V01234A", "V06789B"] and datatype=='th_HS2_lat_psa'``

  • channels (str) –

    expression used to select channels for each run. Expression can access values from all databases, as well as the run table.

    Examples:

    • select all ICPC detectors for each run that are marked as usable:

      "@chan.system=='geds' and @chan.type=='icpc' and @chan.analysis.usability=='on'"``

    • selects SiPM channel 10 and will only include runs where it is can be processed:

      "@chan.name=='S010' and @chan.analysis.processible"

    Note: if a parameter does not exist for a channel, it will evaluate to None. If this causes an error to be thrown, this expression will evaluate to False, excluding the channel. If an parameter always evaluates to False, it will raise an Exception.

  • entries (str) –

    expression used to select data entries for each run/channel. Expression can access values from any data tier, from all databases, and from run table. Parameters with aliases can be accessed using their on-disk field name or their alias.

    Examples:

    • select events with >100 keV of energy, with various event-level cuts applied:

      "(energy > 100) & (~coincident.puls) & (~coincident.spms) & (geds.multiplicity==1) & ak.all(geds.quality.is_bb_like, axis=-1)"

    • select hits with >500 keV of energy and manually applies the low A/E cut:

      ``"(cuspEmax_ctc_cal > 500) & (AoE_classifyer > @pars.pars.operations.AoE_Low_Cut.parameters.a)"``

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • processes (Executor | int) – number of processes. If None, use number equal to threads available to executor (if provided), or else do not parallelize

  • executor (Executor) – concurrent.futures.Executor object for managing parallelism. If None, create a concurrent.futures.`ProcessPoolExecutor with number of processes equal to processes.

  • progress (Status | Console | bool) – if True draw progress bar; can also provide a rich.Status or:class:rich.Console

  • kwargs – see :meth:build_iterator, :meth:query_meta, :meth:query_runs, and :meth:Hist

pygama.flow.query_meta module

pygama.flow.query_meta._query_loop(run_records, col_list, channels, meta, chan_db, db_list, col_name_map, group_chans)
pygama.flow.query_meta.query_meta(fields, runs, channels, *, dataflow_config='$REFPROD/dataflow-config.yaml', group_chans=False, tiers=None, metadata=None, chan_db=None, meta_dbs=None, return_query_vals=False, return_alias_map=False, processes=None, executor=None, library='ak', progress=True, **query_run_kwargs)

Query the metadata and pars data, returning a table containing one entry for each run/channel with the requested data fields. Can also provide boolean expression to select cycles based on data from runs table, and a boolean expression to select based on information about runs and channels found in metadata and parameter databases.

Values from databases are referenced using:

[alias]@db_name.par_path

where:

  • alias: optional alias to use as column name in returned table. If not provided, column name will be db_name_par_path, replacing periods with underscores

  • @db_name: name of data source. Data sources are found on disk using information in the dataflow config file (see dataflow_config):

    • @chan: channel database from metadata.channel_map()

    • @par[_tier]: parameter database from specified tier.

    • Additional data sources defined using the meta_dbs arg or meta_dbs entry in dataflow_config

  • par_path: path in database to par, using periods to separate fields

Examples:

  • @chan.name: name of channel; will be aliased to chan_name

  • rid@chan.daq.raw_id: DAQ id of channel; aliased to rid

  • lt@run.livetime_in_s: livetime from runinfo; aliased to lt

  • aoe_lo@par_hit.pars.operations.AoE_Low_Cut.parameters.a: cut value

    for low A/E cut from hit tier

Parameters:

  • fields (Collection[str]) –

    list of fields to include in the table. See above for description of syntax for naming data sources from metadata and parameter databases.

    Example:

    ["@chan.daq.rawid", "@run.livetime", "aoe_low_cut@par_hit.pars.operations.AoE_Low_Cut.parameters.a"]

  • runs (str | Array | DataFrame) –

    boolean python expression for selecting runs, using column names defined in cycle_def as variables. See query_runs()

    Examples:

    • select calibration data from periods 6, 7 and 8 (assuming l200-style cycle names):

      "period>='p06' and period<='p08' and datatype=='cal'"

    • select runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime):

      "det in ["V01234A", "V06789B"] and datatype=='th_HS2_lat_psa'"

  • channels (str) –

    expression used to select channels for each run. Expression can access values from channel, metadata, and parameter databases (with the channel database @chan likely being the most useful)

    Examples:

    • select all ICPC channels for each run that are marked as usable:

      "@chan.type=='icpc' and @chan.analysis.usability=='on'"

    • select SiPM channel 10 and will only include runs where it is can be processed:

      "@chan.name=='S010' and @chan.analysis.processible"

    Note: if a parameter does not exist for a channel, it will evaluate to None. If this causes an error to be thrown, this expression will evaluate to False, excluding the channel. If an parameter always evaluates to None, it will raise an Exception.

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • tiers (Collection[str] | None) –

    search only provided tiers for pars. If None search all found tiers. By default, get from dataflow-config.

    Examples: ["raw", "dsp", "hit"] or ["raw", "psp", "pht", "evt"]

  • metadata (str | type | MetadataRepository) – class or name of class to use to construct metadata

  • chan_db (str | None) – format string for path in metadata to list of channels for a given cycle. Format string may reference values from the run DB. By default, get from dataflow-config or call LegendMetadata.channelmap(starttime)().

  • meta_dbs (Mapping | None) –

    mapping from database name (i.e. the thing after @) to mapping of configuration parameters. Parameters are as follows:

    • path: path to root of database relative to root of metadata

    • cycle_entry [optional]: sub-path to entry for a given cycle. May be a format string with references to run DB fields. If not provided use dbetto.TextDB.on(starttime)() to find the cycle entry

    • channel_entry [optional]: sub-sub-path to entry for a given channel. May be a format string with references to run DB and channel DB fields. If not provided and group_cycle is False, use @chan.name

    Examples:

    meta_dbs = {
    "runinfo": {
        "path": "path/to/runinfo",
        "cycle_entry": "{period}/{run}/{datatype}",
    },
    "chaninfo": {
        "path": "path/to/chaninfo",
        "cycle_entry": None, # use chaninfo.on(starttime)
        "channel_entry": "@chan.name"
    }
    ...
}

  • group_chans (bool) – if True, return one entry for each run or group of runs, with channel data nested as ragged arrays. Else, return one entry for each channel/run.

  • return_query_vals (bool) – if True, return values found in query as columns; else only return those in fields

  • return_alias_map (bool) – if True, return the pair (table, alias_map) where table is the normal output of this function and alias_map is a mapping from alias names to database paths

  • processes (int | None) – number of processes. If None, use number equal to threads available to executor (if provided), or else do not parallelize

  • executor (Executor | None) – concurrent.futures.Executor object for managing parallelism. If None, create a concurrent.futures.`ProcessPoolExecutor with number of processes equal to processes.

  • library (str) – format of returned table. Can be ak (default), pd or np

  • progress (Status | Console | bool) – if True draw progress spinner; can also provide a rich.Status or:class:rich.Console

  • query_run_kwargs – see query_runs()

pygama.flow.query_runs module

pygama.flow.query_runs._get_run_records_loop(files, relpath, col_names, tiers, removed, runs)
pygama.flow.query_runs.list_run_fields(dataflow_config='$REFPROD/dataflow-config.yaml', cycle_def=None, tiers=None)

List the fields that are available to query_runs().

Parameters:

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • cycle_def (str | None) –

    hyphen-separated names of fields in cycle names; names will be used for columns. By default get from dataflow-config.

    Examples: - experiment-period-run-datatype-cycle for a L200 cycle, e.g. l200-p03-r001-cal-19720101T000000Z - experiment-chan-datatype-run-starttime for a Hades cycle, e.g. char_data-V05268A-th_HS2_lat_psa-r001-20201008T122118Z

  • tiers (str | Collection[str] | Mapping[str, str] | None) – tiers used to find files. First tier in list is used to walk through directories to populate run DB. Remaining tiers are checked for presence of cycles; a cycle is only added if it exists for each tier. File relative path for each tier’s file is added as a column called tier_[t]. Can provide: - Mapping from tier name to path to root of tier - List of tier names/single tier name. Paths will be found in dataflow_config["paths"] - None: read from dataflow_config; if tiers entry not found, use "raw"

Return type:

list[str]

pygama.flow.query_runs.query_runs(runs=None, *, dataflow_config='$REFPROD/dataflow-config.yaml', group_by=None, sort_by='cycle', cycle_def=None, tiers=None, ignored_cycles=None, processes=None, executor=None, library='ak', progress=True)

Query runs and return a table containing one entry for each cycle and data extracted from cycle names. Optionally apply a boolean selection of runs to include using an expression runs.

Run DB is built by recursively cycling through directories in one of the data tiers (using a list of excluded files from metadata). The fields are parsed from the hyphen-separated elements of cycle names (as defined by the cycle-def arg below).

Parameters:

  • runs (str | None) –

    boolean python expression for selecting runs, using column names defined in cycle_def as variables.

    Examples:

    • select calibration data from periods 6, 7 and 8 (assuming l200-style cycle names):

      "period>='p06' and period<='p08' and datatype=='cal'"

    • select runs for detectors V01234A and V06789B from Th calibration data (using Hades data cycle name experiment-det-datatype-run-starttime):

      "det in ['V01234A', 'V06789B'] and datatype=='th_HS2_lat_psa'"

  • dataflow_config (Path | str | Mapping) – config file of reference production. If not provided, use the environment variable $REFPROD as a directory, and find file dataflow-config.yaml

  • group_by (str | Collection[str] | None) – if None (default) return a flat array with all cycles. If one or more fields are provided, group entries by these fields (using ak.run_lengths(), so group consecutive equal values; this is done after sorting, so be careful if sorting changes order!) Fields that vary within groups will be un-flattened into 2-D ragged arrays. Note that runs query cannot act collectively on grouped cycles.

  • sort_by (str | Collection[str]) – field by which to sort table, or list of fields in order by priority

  • cycle_def (str | None) –

    hyphen-separated names of fields in cycle names; names will be used for columns. By default get from dataflow-config.

    Examples: - experiment-period-run-datatype-cycle for a L200 cycle, e.g. l200-p03-r001-cal-19720101T000000Z - experiment-chan-datatype-run-starttime for a Hades cycle, e.g. char_data-V05268A-th_HS2_lat_psa-r001-20201008T122118Z

  • tiers (str | Collection[str] | Mapping[str, str] | None) – tiers used to find files. First tier in list is used to walk through directories to populate run DB. Remaining tiers are checked for presence of cycles; a cycle is only added if it exists for each tier. File relative path for each tier’s file is added as a column called tier_[t]. Can provide: - Mapping from tier name to path to root of tier - List of tier names/single tier name. Paths will be found in dataflow_config["paths"] - None: read from dataflow_config; if tiers entry not found, use "raw"

  • ignored_cycles (str | Collection[str] | None) – path(s) in metadata to list(s) of ignored cycles. By default get from dataflow-config, or else do not skip any cycles.

  • processes (int | None) – number of processes. If None, use number equal to threads available to executor (if provided), or else do not parallelize

  • executor (Executor | None) – concurrent.futures.Executor object for managing parallelism. If None, create a concurrent.futures.`ProcessPoolExecutor with number of processes equal to processes.

  • library (str) – format of returned table. Can be ak (default), pd or np

  • progress (Status | Console | bool) – if True draw progress spinner; can also provide a rich.Status or:class:rich.Console

pygama.flow.utils module

pygama.flow.utils.dict_to_table(col_dict, attr_dict)
pygama.flow.utils.fill_col_dict(tier_table, col_dict, attr_dict, tcm_idx, table_length, aoesa_to_vov)
pygama.flow.utils.format_vars(fstring)

Helper to get list of variables referenced in format string

pygama.flow.utils.get_recursive(db, path)

Helper to recursively access values from nested dict-likes

pygama.flow.utils.inplace_sort(df, by)
pygama.flow.utils.parse_query_paths(expr, fullmatch=False)

Parse input string for variable names of the form:

[alias][@ or :][par.path]

and return a list of each matching 3-tuple of the form:

(full_match, alias, path)

Aliases and names in paths must be legal python names (i.e. alphanumeric, doesn’t start with a digit). If @ is used to separate the alias and path, it is left in the path (to denote a metadata location); if : is used, it is omitted. Note that function names (i.e. a valid name followed by () are excluded. Values inside of [...], {...}, "...", and '...' are also excluded.

If fullmatch is True, expect full string to match pattern and return single tuple. Otherwise return a list of tuples, for each match found.

Return type:

list[tuple[str, str | None, str]] | tuple[str, str | None, str]

pygama.flow.utils.to_datetime(key)

Convert LEGEND cycle key to datetime.

Assumes key is formatted as YYYYMMDDTHHMMSSZ (UTC).

Return type:

datetime

pygama.flow.utils.to_unixtime(key)

Convert LEGEND cycle key to POSIX timestamp.

Return type:

int