API

module JuLeAna

The Julia LEGEND Analysis dataflow.

JuLeAna.main is the entry point of the juleana app: it opens the terminal user interface (JuLeAna.run_tui) or, with --no-tui, runs the enabled processors with run_processing. JuLeAna.setup_session reads the processing config (ProcessingConfig), starts the workers, and returns a DataflowSession, which the DataflowController of the interface works on. The processors are the functions in src/processors/, registered by name in JuLeAna.PROCESSORS. The module also provides the helpers processors share: distributing work over the worker pool with per-item failure logging (parallel), combining per-item log rows into a processor status (master_status), report text and formatting helpers, and the Markdown-safe logger installed on the head and all workers.

main(args::AbstractVector{<:AbstractString})

Entry point of the juleana app and of julia -m JuLeAna: parse the command line options args (see juleana --help) and return the process exit code.

With --no-tui, run all enabled steps with run_no_tui and return its exit code. Otherwise open the terminal user interface with run_tui.

Install the app with Pkg.Apps.add(path = "<path to JuLeAna.jl>"); the juleana command then starts Julia with --threads=8,1, --gcthreads=8,1 and without the user's startup.jl (--startup-file=no): packages loaded there are not in the app's environment. Julia options placed before a -- replace these defaults, e.g. juleana --threads=16,1 -- -c config.json.

setup_session(args::AbstractVector{<:AbstractString})
setup_session(parsed_args::AbstractDict; console::Bool = true, start_workers::Bool = true)

Set up a dataflow session from the command line options args (see juleana --help) or from their parsed form, as returned by parse_commandline.

Build the processing config and LegendData(:l200), run init_process on this process and register it for every worker, so that every process logs to its file in the config's log directory at the config's log_level (console = false turns off the terminal output of this process), and with start_workers start the workers of the configured run mode: local workers are always started; slurm workers only with submit_jobs, otherwise startjlworkers.sh is written to the JuLeAna.production_config_dir for manual submission. Without start_workers the worker pool starts empty. Then check the processor keyword arguments with check_processor_kwargs and build the dependency graph for the enabled steps.

Return the DataflowSession, ready for run_processing. Return nothing when args asks for --help or --version in an interactive session.

Examples

using JuLeAna
session = JuLeAna.setup_session(["--config", "config/processing_config.yaml", "--periods", "3"])
run_processing(session)

run_no_tui(parsed_args::AbstractDict; processors = PROCESSORS, loader = setup_session,
    interval = 1, record_software = true, record_metadata = true)

Run all enabled processors of the session described by the parsed command line options parsed_args without the terminal user interface and return the process exit code.

Set up the session with loader(parsed_args) (by default setup_session, which also starts the workers and the log files), start processing through a DataflowController and show the progress bars of show_progress, updated every interval seconds, until the run ends. The first Ctrl-C stops the run after the running items, a second one stops it without waiting for them (see StopProcessing). Finally log a summary with the outcome, the duration, the number of steps per state and the failed and cancelled steps. With record_software and record_metadata, the run first records the software and the metadata in the production (see DataflowController).

The exit code is 0 when the run finished, 1 when it failed or the session could not be set up, and 2 when it was stopped.

JuLeAna.run_tui(parsed_args::AbstractDict; processors = PROCESSORS, loader)

Open the terminal user interface with the pages Setup, Graph, Workers, Progress, Log, Reports and History on a DataflowController for the parsed command line options parsed_args, as returned by JuLeAna.parse_commandline. Returns nothing when the user quits.

The interface opens at once and loads the session in the background with loader(parsed_args). The default loader is setup_session without terminal output and without starting workers: it reads the config and the LEGEND data and builds the dependency graph. Log messages of this process and the output of its workers are kept in a JuLeAna.MessageBuffer instead of being printed; the newest warning or error is shown in the status panel, and the error of a failed load stays there. This process also writes its log file as in setup_session. The previous global logger is restored when the interface closes, and every worker allocation started from the interface is cancelled (see CancelAll). A startup screen with the logo is shown until the session has loaded; ? lists the keys of the visible page. See JuLeAna.JuLeAnaTUI for the keys. The log file records the thread setup (Interface threads) after the session has loaded.

parse_commandline(args::AbstractVector{<:AbstractString})

Parse the command line options of the dataflow driver and return them as a Dict{String,Any} keyed by the long option names. With --help or --version, ArgParse prints the text and exits the process in a non-interactive session; in an interactive session it returns nothing.

Options (see juleana --help for the full text):

• --config, -c: path of the processing config file. Without it, the JuLeAna.install_default_config file of the data production is used (copied from the config shipped with JuLeAna if the production has none), or nothing when LEGEND_DATA_CONFIG is not set (no data production is selected).
• --periods, -p: periods, each optionally with runs: 6 (all runs of p06), 6,1 (r001), 6,1-3 (r001 to r003), 6,1,4-6. Each occurrence gives one vector of these strings, so -p 3 4 -p 6,1 gives [["3", "4"], ["6,1"]]; see JuLeAna.parse_period_runs.
• --runs, -r: run numbers for the periods given without runs, as one vector per occurrence.
• --runmode, --rm: local or slurm, replacing the config's run mode.
• --dataset: name of the dataset in the metadata datasets/runlists that selects the runs; "default" when not given.
• --log_path, -o: log directory; "" when not given.
• --theme: name of the color theme of the interface; nothing when not given.
• Flags, false when not given: --no-tui, --new-production, --reprocess, --only_runs (--or), --only_partitions (--op), --submit_jobs, --ignore_config, --analysis_runs_only.

Examples

julia> args = JuLeAna.parse_commandline(["-c", "config.json", "-p", "3", "4", "--reprocess"]);

julia> args["periods"], args["reprocess"], args["only_runs"]
([["3", "4"]], true, false)

JuLeAna.mirror_override_folders(l200::LegendData)
JuLeAna.mirror_override_folders(overrides::AbstractString, base::AbstractString)

Create every folder below overrides that is missing below base, at the same relative path, and return the created folders. For l200, this is done for the parameter overrides <metadata>/jldataprod/overrides/par and base = data_path(l200, "par"), and likewise for jlpar; a missing override folder is skipped.

LegendDataManagement opens a parameter folder that exists only in the overrides as a folder of the overrides, so a processor would write its parameters into the metadata. With the folder in base, it opens that one and applies the overrides on top.

JuLeAna.data_config_files()

Return the files named by the environment variable LEGEND_DATA_CONFIG, which separates them by colons as LegendDataManagement reads them. The last file is the data config of the data production; the files before it are overrides, e.g. of the computing site, that LegendDataManagement applies on top of it for every production.

Throws an ArgumentError if LEGEND_DATA_CONFIG is not set.

JuLeAna.with_production_config(config::AbstractString)

Return the value of LEGEND_DATA_CONFIG that selects the data production whose data config is config: the override files of JuLeAna.data_config_files followed by config, or config alone if LEGEND_DATA_CONFIG is not set.

JuLeAna.production_setup_dir()

Return the path of the setup folder of the data production: the folder setup next to the data config of the production, the last file of JuLeAna.data_config_files, in the folder that $_ stands for in that file. It holds the subfolders config, software and metadata that JuLeAna manages. The folder need not exist and is not created.

Throws an ArgumentError if LEGEND_DATA_CONFIG is not set, or if the production folder has a top-level config, software or metadata-records folder; their contents belong in setup/config, setup/software and setup/metadata.

JuLeAna.production_config_dir()

Return the path of the config folder of the data production, the subfolder config of the JuLeAna.production_setup_dir. The folder need not exist.

Throws an ArgumentError under the conditions of JuLeAna.production_setup_dir.

JuLeAna.default_config_file()

Return the path of the default processing config file of the data production, setup/config/processing_config.yaml (see JuLeAna.production_config_dir). The file need not exist; see JuLeAna.install_default_config.

JuLeAna.install_default_config()

Return the JuLeAna.default_config_file. If the file does not exist, first create the JuLeAna.production_config_dir and copy the config/processing_config.yaml shipped with JuLeAna into it, logging the copy. An existing file is never overwritten. A processing_config.json next to the file is ignored with a warning.

JuLeAna.parse_period_runs(period_args, run_args)

Turn the --periods and --runs options of parse_commandline into (periods, runs). period_args and run_args are vectors with one vector per occurrence of the option: strings like "6", "6,1", "6,1-3" or "6,1,4-6" for periods, and run numbers for runs.

periods is a Vector{DataPeriod} in the given order, or nothing without --periods. runs is

• a Dict{DataPeriod,Vector{DataRun}} with the sorted runs of each period given with runs, when any is; the other periods keep all their runs;
• else the --runs numbers as a Vector{DataRun} for all given periods;
• else nothing.

Throws an ArgumentError for a malformed string, a period given twice, a run range that ends before it starts, and for --runs together with periods given with runs.

Examples

julia> periods, runs = JuLeAna.parse_period_runs([["3", "6,1-3"], ["7,2,5"]], []);

julia> string.(periods), sort([string(p) => string.(r) for (p, r) in runs])
(["p03", "p06", "p07"], ["p06" => ["r001", "r002", "r003"], "p07" => ["r002", "r005"]])

init_process(log_path::AbstractString = "", log_level::LogLevel = Logging.Info; console::Bool = true)

Prepare the current process for processing: set the LegendSpecFits fit time limit to 180 s, install the global logger built by processing_logger, and pin the Julia threads with ThreadPinning.pinthreads(AutoThreadPinning()) of ParallelProcessingTools: to the CPUs of the affinity mask of the process if it has one, as in a SLURM job, else not at all. The pinning is logged. A process inside a SLURM job (SLURM_JOB_ID is set) whose job sets JOB_SHMTMPDIR uses that node-local shared-memory folder for tempdir().

With a non-empty log_path, the process appends its records of level log_level to its file in that directory (see process_log_file). The main process shows records of level Info and above on the terminal unless console = false; a worker shows only warnings and errors there. setup_session runs it on the head and on every worker.

mutable struct DataflowSession

State of a dataflow session, as returned by JuLeAna.setup_session: the processing config, the LEGEND data and the selection and dependency graph that run_processing works on. A DataflowController holds the session of the terminal user interface.

Fields:

• args::Dict{String,Any}: parsed command line options, used to reload the config.
• l200::LegendData: the LEGEND data the session processes.
• config::ProcessingConfig: the validated processing config.
• periods::Vector{DataPeriod}: selected periods.
• runs: selected runs: nothing selects all runs of each period, a Vector{DataRun} the same runs in every period, and a Dict{DataPeriod,Vector{DataRun}} the runs per period, where a period without an entry keeps all its runs.
• run_steps::Vector{Symbol}, partition_steps::Vector{Symbol}: run and partition steps scheduled by default, sorted by rank.
• run_status, partition_status: completion flags of the dependency graph, run_status[period][step][run]::Bool for every run step and partition_status[period][step]::Bool for every partition step of the config. A step that is not scheduled starts as true, so steps that depend on it do not wait.

DataflowSession(; args, l200, config, periods, runs=config.runs, run_steps, partition_steps, run_status, partition_status) builds a session from keywords.

report_path(jlrep::AbstractString, process::Symbol, level::Symbol, period, run, category; setup = "l200")
report_path(l200::LegendData, process::Symbol, level::Symbol, period, run, category; setup = "l200")

Return the path of the report written by the processor process. jlrep is the report tier folder, l200.tier[:jlrep] for the LegendData method. level is :run for <jlrep>/rreport/<category>/<period>/<run>/<setup>-<period>-<run>-<category>-<name>.md and :partition for <jlrep>/preport/<period>/<category>/<setup>-<period>-<category>-<name>.md, where <name> is the processor name without its process_ or p_process_ prefix and without the _cal suffix of the processors in RENAMED_PROCESSORS. run is not used for the partition level. The file system is not accessed.

read_report(path::AbstractString)

Read the report at path and return a NamedTuple with the fields

• path: the given path.
• state: :missing if there is no file at path. Otherwise the Processor Status table of the report (Success is :ok, Warning is :warning, Failure is :failed) or, if the report has none, :failed if any entry failed, else :warning if any entry has a warning, else :ok.
• date: the DateTime of the Time of processing: or Date of processing: line, or nothing if there is no such line or it holds no ISO 8601 timestamp.
• n_entries, n_failed, n_warnings: the number of entries, of failed entries and of entries with a warning. An entry is a row of a table with a Status column, per detector and filter type for the fit reports and per filekey for the DSP reports. An entry fails when its status cell contains Failure and has a warning when it contains Warning.
• failures: a vector of (label, error) pairs of strings, one per failed entry. The label is the detector, the failed detectors of a DSP report (one pair per detector), or else the first cell of the row, followed by (filter or energy type, partition) when the row has such columns. The error joins the non-empty cells of the columns whose header contains Error.

An unknown value in the Processor Status table throws an ArgumentError.

report_index(jlrep::AbstractString, processors, period_runs; setup = "l200")
report_index(l200::LegendData, processors, periods, runs; setup = "l200")

Read the reports of every processor in processors (anything with the fields name, level and category, such as ProcessorSpec) for every period => runs pair in period_runs, and return one row per report. A run-level processor has one report per run of runs, a partition-level processor one per period (the runs are not used). The LegendData method takes the report folder from l200.tier[:jlrep] and the runs of each period in periods from processable_runs(l200, period, runs).

A row is the NamedTuple (; process, level, category, period, run, path, state, date, n_entries, n_failed, n_warnings, failures) with the fields of read_report and run = nothing for the partition level. A report that does not exist has state = :missing. Nothing is created in the file system.

report_index(ctrl::DataflowController)

Return the reports of the steps of the current or last processing run of ctrl: one row per step, in schedule order, with the fields of the rows of report_index(jlrep, processors, period_runs). A step whose report does not exist has the state :missing. Returns an empty vector before the first run. Throws an ArgumentError when no session is loaded.

failure_summary(rows)

Group the failed entries of rows (rows of report_index) by processor. Return a vector with one (; process, detectors) per processor, in order of first appearance, where detectors is a vector sorted by name of (; detector, labels, errors): the detector name without the (filter type, partition) suffix of its entry, the unique labels of the runs ("p03-r001") or periods in which it failed and its unique non-empty errors. The failures are the failures of the rows, so no file is read.

report_overview(jlrep::AbstractString; processors = ())
report_overview(l200::LegendData; processors = ())

Read all reports below the report folder jlrep (l200.tier[:jlrep] for the LegendData method), in the folders rreport/<category>/<period>/<run>/ and preport/<period>/<category>/, and group them by report name, level and category. Files that do not follow the naming scheme of report_path are skipped. Nothing is created in the file system.

Return a vector, sorted by level (run before partition), rank, name and category, of (; name, level, category, nreports, counts, last_date, latest_path, reports). counts is (; ok, warning, failed), the number of reports in each read_report state. last_date and latest_path are the processing date of the latest report and its path, nothing if no report has a date. reports is a vector of (; period, run, path, state, date) sorted by period and run, with the period and the run as strings and run = nothing for the partition level.

processors is anything with the fields name, level and rank, such as the ProcessorSpecs of the processing config. The rank of a report is the rank of the processor of its level whose report name (see report_path) is the name of the report; reports of processors that are not in processors come last in their level, ordered by name.

report_overview(ctrl::DataflowController)

Return the report_overview(l200) of all reports in the report folder of the session of ctrl, sorted by the ranks of the processors of its config, whatever the current selection. Throws an ArgumentError when no session is loaded.

JuLeAna.dataset_summary(l200::LegendData)

Return the runinfo of l200 per period and run: a Dict from (period, run) to the data categories of the run that have file keys, in the column order of the runinfo. Each category is a NamedTuple with

• category: the category as a Symbol, e.g. :cal or :phy,
• nkeys: the number of file keys,
• livetime: the livetime in seconds, NaN where the runinfo has none,
• start: the DateTime (UTC) of the first file key, nothing without one,
• analysis: whether the run is an analysis run of this category.

dataset_summary(ctrl::DataflowController)

Return the JuLeAna.dataset_summary of the runinfo of the session of ctrl. Throws an ArgumentError when no session is loaded.

JuLeAna.period_summary(summary, period)

Return the totals of period in summary, as returned by JuLeAna.dataset_summary, per category, in the order the categories first occur in the runs of the period: a vector of NamedTuples with category, nruns (runs with file keys of the category), nkeys, livetime (the sum of the known livetimes in seconds, NaN if no run has one), start (the earliest start, or nothing) and nanalysis (the number of analysis runs).

JuLeAna.history_path(l200::LegendData)

Return the path of the processing history of the data production l200: history.md in the jllog tier.

JuLeAna.history_section(entry)

Return the Markdown section of the processing history for the processing run entry, a NamedTuple with the fields of the entries JuLeAna.read_history returns.

JuLeAna.append_history(path::AbstractString, entry)

Append the section of the processing run entry (see JuLeAna.history_section) to the processing history file path, creating the file with its title if it does not exist. The section is written with a single write.

JuLeAna.read_history(path::AbstractString)

Read the processing history file path written by JuLeAna.append_history and return its processing runs in the order they were written, an empty vector if the file does not exist. Each run is a NamedTuple (; started, finished, user, host, outcome, log_path, software, metadata, steps) with outcome a Symbol, software the description of the software (see JuLeAna.software_version) or nothing for a section without one, metadata the record of the metadata (setup/metadata/<name>, see JuLeAna.record_metadata) or nothing for a section without one, the times as DateTime (UTC, to the second) or nothing, and steps a vector of (; process, level, category, period, run, outcome, started, finished, done, failed, notrun) with period a DataPeriod and run a DataRun, or nothing for a partition-level step.

Throws an ArgumentError naming the file and line of a line that does not follow the format.

JuLeAna.history_grid(processors, period_runs, overview, history)

Combine the reports on disk and the processing history into one cell per processor and run (run level) or period (partition level).

processors are anything with the fields name, level and category, such as ProcessorSpec, in the order of the rows. period_runs is a vector of period => runs pairs with DataPeriods and vectors of DataRuns, the columns. overview is the result of JuLeAna.report_overview, history that of JuLeAna.read_history.

Return (; periods, rows) with periods the vector of (; period, runs) and rows one (; process, level, category, cells) per processor. cells is a Dict from (period, run), with run = nothing for the partition level, to (; state, date, source, path):

• source = :report: the report on disk, state :ok, :warning or :failed, date its processing date (or nothing) and path the report file.
• source = :history: a step of the history that failed, was stopped or was cancelled, state its outcome, date the end of the step (or, for a step that never started, the end of its processing run) and path the log folder of its processing run. The last such step in the history replaces the report when it ended after the report's processing date or there is no dated report; a step that never started only fills a cell without a report.

A cell without a report and without such a step is missing from cells. Reports and steps of processors, periods or runs that are not in the arguments are ignored.

history_grid(ctrl::DataflowController)

Return the JuLeAna.history_grid of the session of ctrl: every processor of the processing config (run level, then partition level, each in rank order) against every period and run of the runinfo, from all reports on disk and the processing history of the data production. Throws an ArgumentError when no session is loaded.

JuLeAna.disk_counts(row, period_runs)

Count the cells of row, a row of JuLeAna.history_grid, in the runs of period_runs (pairs or a Dict of period => runs); a partition-level row has one cell per period. Return (; ok, warning, failed, missing, total): failed counts the cells in state :failed, from a report or the history, missing the cells without a report and those of a stopped or cancelled step, total all counted cells.

JuLeAna.disk_state(counts)

Return the state of a processor on disk from its JuLeAna.disk_counts: :failed if a cell failed, else :missing if a cell is missing, else :warning if a cell has a warning, else :ok; nothing if there are no cells.

run_processing(cfg::ProcessingConfig, l200, periods, run_steps, partition_steps,
    run_status, partition_status; processors::AbstractDict{Symbol}=PROCESSORS,
    check_dependencies::Bool=true, reprocess::Union{Nothing,Bool}=nothing,
    processor_log_level::Union{Nothing,LogLevel}=cfg.processor_log_level,
    processor_plot_level::Union{Nothing,Symbol}=cfg.processor_plot_level,
    max_parallel_steps::Integer=cfg.max_parallel_steps, stop::Threads.Atomic{Bool}=Threads.Atomic{Bool}(false),
    abandon::Threads.Atomic{Bool}=Threads.Atomic{Bool}(false),
    progress_store::Union{Nothing,ProgressStore}=nothing)

Run the partition-level steps partition_steps for every period in periods and the run-level steps run_steps for every period and every run listed in run_status. Processor functions are looked up by name in processors. l200 (usually a LegendData) is passed to every processor and to the run-info checks is_analysis_run and is_lrun. run_status lists the runs to process: every run of a period that has an entry in the first step's table.

Each step of a period, and at run level of a run, starts as soon as the steps it waits for have finished, so steps that do not wait for each other run at the same time. A step waits for the steps that waited_processors returns for it in the processor_graph of cfg, with run_steps and partition_steps as the selected steps:

• a run step for the same run, or, when the graph reaches it through a partition step, for all runs of the periods given by partition_periods that are in periods;
• a run step that a partition step waits for, for all runs of the periods given by partition_periods that are in periods: a partition can contain runs of later periods;
• a partition step for the same period.

A partition step whose processor returns true finishes only when the same step has finished for the previous period. Without check_dependencies a step waits only for the steps of its own level, for the same run or period.

With max_parallel_steps > 0 at most that many steps run at the same time. Steps that are ready start in the order of their rank, then name, period and run. A step that is skipped because the run is not an analysis run (with cfg.analysis_runs_only) or not in the run info counts as finished. run_status[period][step][run] and partition_status[period][step] are set to true when a step finishes.

When a step throws, no further step starts, and once the running steps have ended a CompositeException of the failures is thrown. Steps that wait for each other in a circle throw an ErrorException that names them. Setting stop[] to true stops the run: no further step starts, and parallel hands no further items to workers and throws ProcessingStopped once its running items have finished. With abandon[] also true, parallel throws without waiting for its running items, which go on in the background. reprocess, when not nothing, overrides the reprocess keyword of every processor, processor_log_level, when not nothing, the log_level of every processor (a processor's log records are written at that level, see ProcessorLevelLogger), and processor_plot_level, when not nothing, the plot_level of every processor (see @debug_plot). Plots are saved in the plot_formats of cfg (see save_plot). A step without an entry in processors throws an ArgumentError before any work starts. Return nothing.

With a ProgressStore as progress_store, the store is reset to the scheduled steps, every step's state is recorded there (:skipped for a run that is not processed, :failed or :stopped for a step that threw) and parallel records the items of the step that calls it. Steps that never finished are :stopped when the run was stopped and :cancelled otherwise.

run_processing(session::DataflowSession; kwargs...)

Run the session's scheduled steps over its periods and update its status flags, as run_processing(session.config, session.l200, session.periods, session.run_steps, session.partition_steps, session.run_status, session.partition_status; kwargs...). The keywords are those of that method. Return nothing.

run_step(step::Symbol, period, run = nothing; detectors = nothing, outputs = :temp,
    output_dir = nothing, config = JuLeAna.default_config_file(), l200 = nothing,
    show_report::Bool = isinteractive(), debug = false, pause = nothing, error_plots = nothing, log_level = nothing,
    plot_level = nothing,
    processors::AbstractDict{Symbol} = PROCESSORS, kwargs...)

Run the processor step for period and, for a run-level processor, run, in the calling Julia process, for example in the REPL. A partition-level processor takes no run. step is the name of a registered processor, a key of JuLeAna.PROCESSORS, and JuLeAna.processors lists them with their level, category and rank; ?JuLeAna.<step>, e.g. ?JuLeAna.process_energy_calibration_cal, shows what the processor does and its keywords.

Only the defaults are needed: by default, nothing is written to the data production, and the plots, reports and data files go to a temporary directory.

Keywords

• detectors: the detectors to process, e.g. [:V01234A]; they are processed again, also when their parameters or output exist. nothing (the default) processes all detectors, with reprocess = true unless given. A detector that the processor does not process in this run or partition stops the call with an error listing the detectors it processes.
• outputs: :temp (the default) writes the plots, reports and the data tiers the processor writes (its output_tiers, see register_processor!) to output_dir, and no parameters or validity; parameters are read from the production, and the detectors not in detectors are not run. :production writes everything to the production paths of l200, which cannot be undone; the detectors not in detectors are skipped as already processed, so the parameters, validity and reports stay complete.
• output_dir: the directory of the temporary outputs; by default one temporary directory for all calls in the Julia session, see step.output_dir. The report of each call is appended to <output_dir>/reports/<step>.md: the report of the processor with only the rows of detectors, and below it the plots of each detector the processor saved during this call, in the order they were saved.
• config: a ProcessingConfig or the path of a config file; by default the config of the data production. The processor must be in it; for a processor that is not, use the config of the package, joinpath(pkgdir(JuLeAna), "config", "processing_config.yaml").
• l200: the LegendData to use; by default LegendData(:l200), built after the env_variables of config are set.
• show_report: display the report at the end with display_report; the default in an interactive session.
• debug: true runs the items one after the other in the calling process, without workers and timeout, and captures each failed @process_step as a Failure with the variables of the processor at that step; see JuLeAna.failure. A summary of the failures is printed at the end. :infiltrate does the same and also stops at each failed step in an Infiltrator prompt with these variables; leaving the prompt goes on with the error. Needs using Infiltrator.
• pause: :all, or the names of steps, e.g. ["e_cusp simple calibration"], with their interpolations done: stop in an Infiltrator prompt after each of these steps that succeeds, with the variables of the processor and the value of the step as step_value. Implies debug = true if debug is false. Needs using Infiltrator.
• error_plots: whether a failed @process_step runs its on_error hook, which saves a debug plot; nothing (the default) takes the error_plots setting of the processor in config. In the report, these plots, whose names start with error_, are linked, not embedded.
• log_level: :debug, :info, :warn or :error: the processor logs at this level, also to the terminal; at :debug this includes the debug records of the DEBUG_LOG_PACKAGES. nothing (the default) takes the log_level of the processor in config, which applies to the log files only. Items on workers log to the workers' log files.
• plot_level: :info or :debug; at :debug the processor also saves its @debug_plot plots, which the report shows like the others. nothing (the default) takes the plot_level of the processor in config.
• processors: the processors by name, JuLeAna.PROCESSORS by default.
• kwargs: keyword arguments of the processor, e.g. timeout = 0. The processor gets the keyword arguments of config, then only_first_period = false for a partition-level processor and reprocess = true when detectors is nothing, then kwargs.

Workers

The items of the processor, e.g. its detectors, go to the workers of the ParallelProcessingTools worker pool, or run in the calling process when no workers are started. Workers started from the session need JuLeAna and ParallelProcessingTools in the active environment:

using JuLeAna
ParallelProcessingTools.@always_everywhere using JuLeAna
ParallelProcessingTools.runworkers(ParallelProcessingTools.OnLocalhost(n = 4))

runworkers returns at once; the workers join the pool when they are up, see ParallelProcessingTools.worker_resources().

Returns

Return a StepOutput with the report, the reports of the processor, the results of its items and the captured failures.

Throws an ArgumentError if step is not registered or not in config, if run is given for a partition-level processor or missing for a run-level one, if detectors is given for a processor that cannot select detectors, or if outputs is not :temp or :production.

Examples

using JuLeAna
# run level: a detector of p03 r000
step = run_step(:process_energy_calibration_cal, 3, 0; detectors = [:V01234A])
# partition level, no run
step = run_step(:p_process_energy_cal, 3; detectors = [:V01234A])
JuLeAna.display_report(step.report)
# capture the failed steps of a detector
step = run_step(:process_energy_calibration_cal, 3, 0; detectors = [:V01234A], debug = true)
f = JuLeAna.failure()

ReportPlot(file)

The PNG plot file in the report of run_step. In the Markdown and HTML report, the plot is embedded as an image; as text, it is shown by its file name. display_report displays it as an image.

display_report(report)
display_report(path::AbstractString)

Display the report report of run_step, e.g. step.report, or the Markdown report file at path, e.g. a report of the production found with JuLeAna.report_path or step.report_file.

In VS Code with the plot pane on, the report is shown there as one page, with its tables and plots. Otherwise report is shown with display: the text and tables as usual, and each ReportPlot as an image by the first display in the display stack that shows image/png. In a terminal, such displays come with KittyTerminalImages (Kitty, Ghostty) and ImageInTerminal (terminals with Sixel support, needs ImageIO to read the PNG). Without such a display, a plot is shown by its file name. A report file is shown as formatted Markdown text.

Examples

using KittyTerminalImages
JuLeAna.display_report(step.report)
JuLeAna.display_report(JuLeAna.report_path(l200, :process_energy_calibration_cal, :run, DataPeriod(3), DataRun(0), :cal))

JuLeAna.processors(config = JuLeAna.default_config_file(); processors = PROCESSORS)

Table of the registered processors processors: the run-level processors of config in the order of their rank, then the partition-level ones, then the processors that are not in config. config is a ProcessingConfig or the path of a config file.

Columns: Processor; from config Level (:run or :partition), Category, Rank and Enabled, missing for a processor that is not in config; and from PROCESSOR_TRAITS Detectors, whether run_step can run the processor for chosen detectors, and OutputTiers, the data tiers it writes.

Examples

JuLeAna.processors()
JuLeAna.processors(joinpath(pkgdir(JuLeAna), "config", "processing_config.yaml"))

StepOutput

The output of run_step, with the fields

• step, args: the processor and its period and run,
• report: the report of the call as a LegendReport, with the plots as ReportPlots; display_report shows it with the plots as images,
• report_file: the path of the report file the report was appended to,
• reports: the reports the processor wrote, complete with all detectors,
• output_dir: the output directory,
• results: the item => result pairs of all parallel calls of the processor,
• failures: the Failures captured with debug or pause.

JuLeAna.@using_packages

Load the exported names of the packages JuLeAna works with into the calling module, e.g. the REPL, as using does: the LEGEND packages with LDMUtils, ParallelProcessingTools, CairoMakie with the names of Makie, LegendMakie, and the packages for units, tables, measurements and timing, with mvalue and muncert for the value and uncertainty of a Measurement.

The packages are loaded from JuLeAna, so they need not be in the active environment. Names that two packages both export, such as fit, must be qualified, e.g. StatsBase.fit.

Examples

using JuLeAna
JuLeAna.@using_packages
channelinfo(l200, filekey)

ProcessingStopped()

Exception thrown by parallel when processing was stopped while the batch ran: items that had not started are not run, and the processor that called parallel ends without writing results.

partition_periods(l200::LegendData, period::DataPeriod)

Return the sorted periods that share a default cal or phy partition with period, period included. The default partitions in datasets.cal_groupings and datasets.phy_groupings are the widest: the partitions of single detectors lie within them.

rank_order_dependencies(specs::AbstractDict{Symbol,ProcessorSpec})

Return a copy of the processor table specs (one level) in which every processor also depends on the processor before it in the order of rank, then name. The dependencies then express that the processors run one after another in rank order.

mutable struct DataflowController

Headless state of an interactive dataflow session. All changes go through dispatch! with a ControllerCommand; snapshot returns a read-only view of the state and progress the progress of the current or last processing run. Commands may be sent from any task; long-running ones (loading the session, reloading the config, processing) run in the background.

The controller is in one of the states

• :idle: no session loaded yet,
• :loading: the session or config is being loaded,
• :ready: a session is loaded and no processing runs,
• :running: processing runs,
• :stopping: processing was asked to stop and runs until the running items finish.

Constructors:

• DataflowController(args::AbstractDict; processors = PROCESSORS, console = true) starts :idle with the parsed command line options args; LoadSession sets up the session, with terminal log output of this process only if console.
• DataflowController(session::DataflowSession; processors = PROCESSORS) starts :ready with an existing session.

processors maps processor names to their functions. Both constructors take the keyword workers, the JuLeAna.WorkerManager of the worker allocations, and record_software: whether each processing run first records the software of this process in the setup/software folder of the production (see JuLeAna.record_software), and record_metadata: whether it records the metadata and the config files (the data config files and the processing config args["config"]) in its setup/metadata folder (see JuLeAna.record_metadata). A run fails if a record fails.

dispatch!(ctrl::DataflowController, cmd::ControllerCommand)

Apply cmd to ctrl and return nothing. Throw an ArgumentError that explains why when the command is not possible in the controller's state or its arguments are invalid; the controller is then unchanged. Commands that run in the background return as soon as they are started; see snapshot for their state and wait(ctrl) to block until they are done.

snapshot(ctrl::DataflowController)

Return a JuLeAna.ControllerSnapshot, a NamedTuple with fixed field types, with a copy of the controller state:

• state, outcome (:finished, :failed, :stopped or nothing of the last processing run), last_error (message or nothing), started, finished (UTC times of the last processing run or nothing);
• config_file, log_path, dataset (the name of the dataset, "default" for every run);
• available_periods, available_runs (Dict period => runs of the dataset), periods (selected periods), runs (selected runs: nothing for all runs, a vector for the same runs in every period, or a dict of runs per period where a period without an entry keeps all its runs), run_steps, partition_steps (selected steps in processing order);
• processors: one NamedTuple per configured processor, run level first and then by rank, with name, level, category, rank, enabled, selected, dependencies, kwargs (with the overrides applied), config_kwargs (as in the config), accepted (the keyword arguments the processor function accepts besides the ones the dataflow passes itself, nothing if it accepts any), overridden (names of the overridden kwargs, and :log_level and :plot_level if overridden), log_level, plot_level (with the overrides of SetStepLevel applied) and config_log_level, config_plot_level (as in the config);
• reprocess, processor_log_level, processor_plot_level, analysis_runs_only and runmode (the settings in effect), runmode_settings (local_n, local_julia_settings, slurm_flags and slurm_julia_settings in effect, or nothing without a session), check_dependencies, nworkers (workers in the pool), allocations (the worker allocations, see JuLeAna.allocations, without their output), idle_deadline (UTC time at which idle allocations are cancelled, or nothing, see JuLeAna.idle_deadline).

wait(ctrl::DataflowController)

Block until the background task of the last loading or processing command has ended.

progress(store::ProgressStore)
progress(ctrl::DataflowController)

Return a copy of the progress of the current or last processing run as a JuLeAna.ProgressSummary, a NamedTuple with fixed field types:

• started, finished: UTC times of the run, nothing if not (yet) reached.
• fraction: overall progress in [0, 1], the mean of the step fractions. A finished, failed, skipped, stopped or cancelled step counts as 1; a waiting step as 0; a running step as (done + failed + notrun) / (total + 1) of its items, so that the overall progress reaches 1 only after the last step has finished.
• eta: estimated seconds until the run finishes, from the elapsed time and fraction, or nothing before any progress or after the run.
• counts: number of steps per state, a NamedTuple with one field per state (waiting, running, done, failed, skipped, stopped, cancelled).
• steps: one NamedTuple per step in schedule order with period, run (nothing for partition-level steps), step, state, total, done, failed, notrun, workers (sorted workers of the running items), started, finished, error (the error message of a failed step, else nothing), nmessages (number of messages kept for the step, see step_messages), fraction, rate (items done or failed per second since the step queued its first item, up to the step's end; nothing before the first item ended) and eta (seconds for the remaining items of a running step at that rate, or nothing).
• groups: one NamedTuple per period (partition level) and per run, in schedule order, with period, run, nsteps, counts and fraction over its steps.

busy_workers(store::ProgressStore)
busy_workers(ctrl::DataflowController)

Return the items that workers are running, oldest first, as NamedTuples with worker, item (the item's label), period, run (nothing for partition-level steps), step, since (UTC time the worker got the item) and abandoned. An abandoned item belongs to a processing run that was stopped without waiting for it; its worker stays busy until the item's call returns, which for an unresponsive worker never happens. For a controller, the items of workers that have ended are left out: they wait for a retry on another worker.

step_messages(store::ProgressStore, key)
step_messages(ctrl::DataflowController, key)

Return a copy of the messages logged on this process while step key of the current or last processing run ran, oldest first, as NamedTuples with level (a Logging.LogLevel), time (UTC) and text (possibly several lines). key is (period, run, step) with run = nothing for a partition-level step. Only the last 500 messages of a step are kept. Returns an empty vector for a step that is not part of the run.

ProgressStore()

Thread-safe record of the progress of a processing run, filled by run_processing and parallel when run_processing is given the store as progress_store. It holds one entry per scheduled step of a period (partition level) or of a run (run level) with the step's state, the number of items handed to parallel and how many of them finished, failed, were not run or are running on which worker. Messages logged on this process while a step runs are kept with the step (see step_messages) when the logger in use passes them to the store, as JuLeAna.MessageBuffer, the logger of the terminal user interface, does.

The store also records the busy workers: every item that a worker runs, from the moment the item is handed to the worker until its call returns. This record is kept across processing runs, so that items whose run was stopped without waiting for them stay listed until their workers are free again.

progress returns a copy of the current state with the aggregated progress and busy_workers the busy workers.

show_progress(f, store::ProgressStore; interval::Real = 1)

Call f() and, while it runs, log the progress in store every interval seconds as progress records of ProgressLogging: one bar for the overall progress and one below it for each running step. A TerminalLogger draws these bars below the log messages and keeps a line with the final state of each step whose bar it closes; other loggers receive them as Info records whose message is a ProgressLogging.Progress. All bars are closed when f returns or throws. Returns the value of f().

effective_config(ctrl::DataflowController)

Return the session's ProcessingConfig with the kwarg overrides of the controller merged into the processor kwargs, its level overrides (see SetStepLevel) applied to the processors, and its analysis_runs_only, runmode and run mode settings (see SetRunmodeSettings) applied.

abstract type ControllerCommand

Supertype of the commands that dispatch! applies to a DataflowController.

LoadSession()

Set up the session from the controller's command line options with setup_session: read the config and the LEGEND data and start the workers. Runs in the background; the controller is :loading until it is :ready. Allowed when the controller is :idle.

ReloadConfig()

Read the processing config file again with the original command line options, keeping the log directory, and reset the selection to the config's periods, runs and enabled processors. Drops all kwarg overrides. Runs in the background like LoadSession.

SelectPeriods(periods)

Select the periods to process, given as DataPeriods or anything DataPeriod accepts. Every period must be in the runinfo of the dataset. Periods that stay selected keep their runs.

SelectRuns(runs)

Select the runs to process.

• nothing selects all runs of the selected periods.
• A vector of DataRuns (or anything DataRun accepts) selects the same runs in every selected period; every run must be available in at least one selected period.
• A dict of period => runs (keys and values as DataPeriod and DataRun accept) selects exactly these periods, each with its runs, where nothing stands for all runs of the period. Every period and every run of it must be in the dataset.

SelectSteps(run_steps, partition_steps)

Select the run-level and partition-level processors to run, by name. Every name must be a processor of that level in the config, enabled or not. The steps run in the order of their dependencies, see run_processing.

SetReprocess(reprocess::Union{Nothing,Bool})

Set the reprocess keyword passed to every processor: true or false override the config, nothing keeps the configured values.

SetProcessorLogLevel(level::Union{Nothing,Logging.LogLevel})

Set the level at which the records of every processor are written to the log files, for example Logging.Debug, replacing the log_level of each processor entry. nothing keeps the levels of the processor entries. The level starts as processor_log_level of the config.

SetProcessorPlotLevel(level::Union{Nothing,Symbol})

Set the plot level of every processor, :info or :debug, replacing the plot_level of each processor entry. nothing keeps the levels of the processor entries. The level starts as processor_plot_level of the config.

SetAnalysisRunsOnly(only::Bool)

Choose whether processing skips the runs that are not analysis runs (true) or processes every run in the run info (false), instead of the setting of the config.

SetRunmode(runmode::Symbol)

Use the run mode runmode, :local or :slurm, instead of the one of the config, for the next worker allocations; running allocations are not changed.

SetRunmodeSettings(; local_n = nothing, local_julia_settings = nothing,
    slurm_flags = nothing, slurm_julia_settings = nothing)

Use the given settings of the run modes instead of those of the config, for the next worker allocations and the start script; running allocations are not changed. Also possible during a processing run, which keeps its config. Each argument that is nothing keeps the setting in effect. local_n is the number of local workers and must be positive; local_julia_settings, slurm_flags and slurm_julia_settings are vectors of strings, the extra options of the local Julia processes, the flags of the SLURM job and the options of the Julia processes it starts.

SetCheckDependencies(check::Bool)

Choose whether a processing run rebuilds the dependency graph for the selected steps and waits for their dependencies (true), or runs the selected steps on the current graph without waiting (false).

SetKwarg(step, key, value)

Override the keyword argument key of the processor step for this session. The processor function must accept key.

SetStepLevel(step, key, value)

Override the log level (key = :log_level, value a LogLevel or one of "debug", "info", "warn" and "error") or the plot level (key = :plot_level, value :info or :debug) of the processor step for this session. A level set for all processors with SetProcessorLogLevel or SetProcessorPlotLevel takes precedence.

ResetKwargs(step = nothing, key = nothing)

Drop the kwarg and level overrides (see SetKwarg and SetStepLevel) of all processors, of the processor step, or only the override of key (a kwarg, :log_level or :plot_level) of the processor step.

ResetDependencyGraph()

Mark the selected steps as unfinished for the selected periods and runs.

StartProcessing()

Run the selected steps over the selected periods and runs in the background with run_processing, using the config with the session's kwarg overrides. The controller is :running until the run ends; the outcome is :finished, :failed or :stopped. progress shows the progress of the run. When the run ends, its steps are added to the processing history of the data production (see JuLeAna.append_history and JuLeAna.history_path). Throws an ArgumentError when no step is selected or a selected processor does not accept its keyword arguments.

StopProcessing(; force::Bool = false)

Stop the processing run: no further step or item starts, items running on workers finish, and the run ends as :stopped. With force = true, the run ends without waiting for the running items. They count as not run, go on in the background and keep their workers busy until they end; busy_workers lists them as abandoned. No worker is stopped.

WriteConfig(path::AbstractString)
WriteConfig()

Write the processing config file of the session to path with the current settings of the session: the kwarg overrides are applied to the processor entries, every processor is enabled exactly when it is selected, and processing.analysis_runs_only is the setting in effect. Without path, the file is processing_config.yaml in the JuLeAna.production_config_dir, which is created if needed. An existing file at the path is first copied to path * ".bak". The file is written as YAML. The written file becomes the config file of the session, which ReloadConfig reads.

AddWorkers()

Start a worker allocation: run the worker start script of the session in the background (see JuLeAna.add_workers!). The workers join the worker pool when they have started.

CancelAllocation(id::Integer)

Cancel the worker allocation id (see JuLeAna.cancel_allocation!): its workers end and items running on them fail.

CancelAll()

Cancel every running worker allocation of the session (see JuLeAna.cancel_all!).

PostponeIdleCancel()

Restart the idle time after which the worker allocations are cancelled (see JuLeAna.check_idle!) from now.

EditStartScript(text::AbstractString)

Replace the worker start script of the session by text for the next allocations (see JuLeAna.edit_start_script!).

CreateProduction(spec::ProductionSpec)

Create the data production spec describes with JuLeAna.create_production in the background and switch the session to it: cancel all worker allocations (their workers use the data of the previous production), set LEGEND_DATA_CONFIG of this process to the new config.yaml, and load a session with the JuLeAna.install_default_config file of the new production and a new log folder. The lines of the creation are kept for production_log. Possible while no session is loaded or no processing runs; the controller is :loading meanwhile and returns to its previous state if the creation or the load fails.

SelectProduction(path::AbstractString)

Switch the session to the existing data production in the folder path, which must hold a config.json or config.yaml (config.json if both exist): cancel all worker allocations, set LEGEND_DATA_CONFIG of this process to that file, reset the dataset to default, and load a session with the JuLeAna.install_default_config file of the production and a new log folder. Possible while no session is loaded or no processing runs; the controller is :loading meanwhile and is :idle without a session if the load fails.

SelectDataset(dataset::Symbol)

Reload the session for the dataset dataset, a name in the metadata datasets/runlists of the data production or :default (every run), and cancel all worker allocations. The dataset is the --dataset option. Possible while no session is loaded or no processing runs; a dataset that is not in the metadata makes the load fail, leaving the controller :idle without a session.

JuLeAna.check_idle!(ctrl::DataflowController)

Cancel the worker allocations of ctrl if they have been idle for the idle_timeout_minutes of the config (see JuLeAna.check_idle!); allocations are never idle while processing runs or is stopping. Return whether they were cancelled. The terminal user interface calls it with every read of the controller state.

WorkerManager(; squeue::Cmd = `squeue`, scancel::Cmd = `scancel`, sinfo::Cmd = `sinfo`)

The worker allocations of this process and the worker start script they run.

The start script is generated from the processing config of the session by JuLeAna.start_script! and saved as startjlworkers.sh in the JuLeAna.production_config_dir. It holds the cluster cookie and the address of this process, so it is generated anew for every config (JuLeAna.set_worker_config!) and can be edited for the rest of the session (JuLeAna.edit_start_script!). JuLeAna.add_workers! runs it as a new allocation, JuLeAna.cancel_allocation! and JuLeAna.cancel_all! end allocations, JuLeAna.allocations lists them.

Running allocations are cancelled when nothing has used them for the idle_timeout_minutes of the config: see JuLeAna.idle_deadline and JuLeAna.check_idle!.

squeue, scancel and sinfo are the SLURM commands, without arguments; SLURM counts as available when the program of squeue is found. JuLeAna.workers_overview collects everything the Workers page shows. All functions may be called from any task.

WorkerAllocation

One run of the worker start script, started by JuLeAna.add_workers!. Fields:

• id::Int: number of the allocation, counted from 1 per JuLeAna.WorkerManager.
• backend::Symbol: :local or :slurm.
• job_name: the SLURM job name of the allocation's workers, nothing for :local.
• started, finished: UTC times; finished is nothing while the script runs.
• process: the Base.Process of the shell that runs the script, the leader of its own process group.
• state::Symbol: :running, :ended (the script exited with code 0), :failed (any other exit) or :cancelled.
• exit_code: the exit code of the script, nothing while it runs.
• job_ids::Vector{String}: the SLURM job IDs of the allocation at the last query.
• output::Vector{String}: the last 1000 lines of the standard and error output of the script.

JuLeAna.set_worker_config!(mgr::WorkerManager, config::ProcessingConfig)

Use the run mode of config for the next allocations of mgr. The start script is generated anew on its next use; running allocations are not changed.

JuLeAna.start_script!(mgr::WorkerManager)

Return (; path, text) of the start script of mgr. The first call after JuLeAna.set_worker_config! generates the script for the run mode of the config with ParallelProcessingTools.write_worker_start_script and writes it to startjlworkers.sh in the JuLeAna.production_config_dir, keeping an existing file as startjlworkers.sh.bak. Throws an ArgumentError if no config is set.

JuLeAna.edit_start_script!(mgr::WorkerManager, text::AbstractString)

Replace the start script of mgr by text and save it to the script file. For the slurm run mode the script must name its SLURM job --job-name=julia-<n>-<m>: each allocation replaces the name by its own, by which it finds and cancels its jobs. Throws an ArgumentError for a SLURM script without such a name.

JuLeAna.add_workers!(mgr::WorkerManager)

Run the start script of mgr (see JuLeAna.start_script!) with bash in the background as a new allocation and return its JuLeAna.WorkerAllocation. The shell leads its own process group, so it keeps running when this process ends unless it is cancelled. Its standard and error output are kept in the allocation. For the slurm run mode, the job name of the script is replaced by julia-<pid of this process>-<n> with a new n. The started workers connect to this process and join its worker pool.

JuLeAna.update_job_ids!(mgr::WorkerManager)

Query squeue for the jobs of the running SLURM allocations of mgr and store their IDs in the allocations. Does nothing without such allocations.

JuLeAna.cancel_allocation!(mgr::WorkerManager, id::Integer)

End the allocation id of mgr if it is running: for a SLURM allocation, scancel the jobs that squeue lists for this user under the allocation's job name (and no others), then send SIGTERM to the process group of its script, which ends the local workers or srun. The workers of the allocation leave the worker pool; items running on them fail. Throws an ArgumentError for an unknown id.

JuLeAna.cancel_all!(mgr::WorkerManager)

Cancel every running allocation of mgr with JuLeAna.cancel_allocation!.

JuLeAna.allocations(mgr::WorkerManager; output::Bool = true)

Return a copy of the allocations of mgr as NamedTuples with the fields of JuLeAna.WorkerAllocation except process, and without output unless output: a Vector{JuLeAna.AllocationOutputInfo}, or a Vector{JuLeAna.AllocationInfo} without output.

JuLeAna.mark_activity!(mgr::WorkerManager, time::DateTime = now(UTC))

Record that the workers of mgr were used at time (UTC), such as by the end of a processing run or a postponement: the idle time of JuLeAna.idle_deadline counts from the latest such time.

JuLeAna.idle_deadline(mgr::WorkerManager; processing::Bool)

Return the UTC time at which JuLeAna.check_idle! cancels the running allocations of mgr: the last activity (see JuLeAna.mark_activity!; a new allocation counts too) plus the idle_timeout_minutes of the config. Return nothing while processing, if no allocation runs, without a config, or if the timeout is 0.

JuLeAna.check_idle!(mgr::WorkerManager; processing::Bool, time::DateTime = now(UTC))

Cancel all running allocations of mgr with JuLeAna.cancel_all! if time (UTC) has reached the JuLeAna.idle_deadline. Return whether they were cancelled.

JuLeAna.workers_overview(mgr::WorkerManager)

Collect what the Workers page shows and return a JuLeAna.WorkersOverview (; allocations, script, workers, slurm):

• allocations: JuLeAna.allocations with their output, after JuLeAna.update_job_ids!;
• script: the (; path, text) of JuLeAna.start_script!, or the message of the error that kept it from being generated;
• workers: one (; pid, host) per worker process, host nothing until it is known; the host of a new worker is asked for in the background;
• slurm: nothing without SLURM, else (; queue, idle), the output of squeue -u <user> and of sinfo -t idle.

Runs SLURM commands, so it can take a while; call it in the background.

ProductionSpec(; name, parent, metadata_url = "git@github.com:legend-exp/legend-metadata", metadata_ref = nothing,
    jldataprod_branch = nothing, dataprod_branch = nothing, tier_paths = Pair{String,String}[], pars = String[],
    set_default = false, copy_software = true, shell = basename(get(ENV, "SHELL", "")),
    home = homedir())

What JuLeAna.create_production creates: the data production name in the folder parent (an absolute path).

• metadata_url: the legend-metadata repository to clone.
• metadata_ref: branch or tag to check out in the metadata repository, with every submodule at the commit that ref records; nothing to keep the cloned default branch. The two branches below are checked out on top of it.
• jldataprod_branch: branch to check out in the jldataprod/config* and jldataprod/overrides* submodules of the metadata, nothing to keep the cloned ones.
• dataprod_branch: branch to check out in the metadata repository and all other submodules except simprod/config*, nothing to keep the cloned ones.
• tier_paths: label => path pairs of tiers kept in other folders, written to config.yaml as tier/<label>; each path must be an existing folder.
• pars: folders of parameters to copy into generated/jlpar; each path must contain a /rpars or /ppars component, and the part from there on is kept below jlpar.
• set_default: whether to set LEGEND_DATA_CONFIG to the new production in the startup file of the shell shell (bash: .bashrc, zsh: .zshrc) in home.
• copy_software: whether to copy the software of this process into the production (see JuLeAna.record_software).

JuLeAna.check_production(spec::ProductionSpec)

Return the folder of the production spec describes, after checking spec without changing anything: the name contains only letters, digits, ., _ and -, the parent is an absolute path, the production folder does not exist, the branch names contain only letters, digits, ., _, / and - and do not start with - (also the metadata_ref), every tier label is unique and contains only letters, digits, _ and -, every tier path and parameter folder exists, every parameter folder has a /rpars or /ppars component, and with set_default the shell is bash or zsh. Throws an ArgumentError that names the first problem.

JuLeAna.create_production(spec::ProductionSpec; log = line -> @info(line))

Create the data production spec describes and return the path of its data config file config.yaml, which LEGEND_DATA_CONFIG names to use it. Checks spec first with JuLeAna.check_production, then

1. creates the production folder and clones the metadata with its submodules into legend-metadata; with metadata_ref it checks out that branch or tag and updates all submodules to the commits it records, then it checks out the other branches of spec,
2. creates the folders generated/{tier,jllog,jlplt,jlreport,jlpar/rpars,jlpar/ppars},
3. writes config.yaml (see JuLeAna.data_config_text),
4. copies the parameter folders into generated/jlpar,
5. copies the software of this process into a subfolder of setup/software (see JuLeAna.record_software),
6. sets LEGEND_DATA_CONFIG in the shell startup file: replaces the lines that export it, or appends one.

log gets a line for every step and the output of git. A failing step throws and leaves what was created so far.

JuLeAna.data_config_text(tier_paths, parent)

Return the text of the config.yaml of a new production: the paths of the l200 setup relative to the folder of the file ($_), and after the tier path one tier/<label> entry per pair of tier_paths. A path below parent, the folder that holds the production, is written relative to it as $_/... The paths are single-quoted YAML strings.

JuLeAna.running_manifest()

Return the path of the manifest of the environment this process loaded JuLeAna from: the first environment of Base.load_path() whose manifest has a JuLeAna entry. Throws an ErrorException if there is none.

JuLeAna.snapshot_software(dest::AbstractString; manifest = JuLeAna.running_manifest(),
    source = pkgdir(JuLeAna), log = line -> @info(line))

Copy the software of this process to the folder dest, so that it can be installed again with Pkg.Apps.develop(path = "<dest>/JuLeAna.jl"):

• the package source of JuLeAna to dest/JuLeAna.jl: for a git checkout the files git tracks, with their uncommitted changes, and the untracked files it does not ignore, else every file except .git and docs/build; its own manifests are not copied;
• every other package that manifest takes from a path (developed packages) to dest/<name>, likewise;
• manifest to dest/JuLeAna.jl/Manifest.toml, with the JuLeAna entry taken from . and the copied packages from ../<name>, so that every package keeps its version;
• each copied package folder is a git repository with one commit of all its files, including Manifest.toml, so that Pkg.add(path = ...) accepts it. The commit has a fixed author and does not depend on the git configuration of the user;
• dest/SNAPSHOT.toml: the date, user, host and Julia version, for each package its source, its commit and whether it was modified (for a git checkout) and the tree of the commit of the copy, and a fingerprint of the Julia version and the trees. Equal software has equal fingerprints;
• dest/PROVENANCE.md: the same for reading, and the steps to run the copy.

Returns dest. Throws if dest exists.

JuLeAna.record_software(root::AbstractString; manifest = JuLeAna.running_manifest(),
    source = pkgdir(JuLeAna), log = line -> @info(line))

Make sure the folder root/setup/software holds a snapshot of the software of this process and return the name of its subfolder. Takes a new snapshot with JuLeAna.snapshot_software and compares its fingerprint with that of every subfolder that has a SNAPSHOT.toml; other subfolders are ignored. For equal software it removes the new snapshot and returns the existing subfolder. Otherwise the new snapshot becomes the subfolder <yyyy-mm-ddTHHMMSSZ>-<short JuLeAna commit>, with -modified appended if JuLeAna had uncommitted changes, and nogit instead of the commit if it is not a git checkout. Throws if a step fails.

JuLeAna.software_version(snapshot = nothing)

Return a one-line description of the software of this process for the processing history: the JuLeAna version, and for a git checkout its short commit and modified if it has uncommitted changes, the Julia version, and with snapshot, the name of a subfolder of setup/software (see JuLeAna.record_software), snapshot setup/software/<snapshot>.

JuLeAna.record_metadata(root::AbstractString, metadata::AbstractString;
    configs = String[], log = line -> @info(line))

Make sure the folder root/setup/metadata holds a record of the metadata checkout metadata and its submodules as they are now, together with the config files configs, and return the name of its subfolder. Reads the metadata with git only and does not change it. Throws an ArgumentError if metadata is not a git checkout.

A record holds:

• METADATA.toml: the date, user and host; for every repository ("." for metadata, else the path of the submodule) its commit, branch (unless detached), remote, whether it is modified, for a submodule whether it is moved from the commit its superproject records, its patch and its untracked files with their path, hash, size and whether they were copied; for every config file its source, file and hash; and a fingerprint of all of it. Equal metadata and configs have equal fingerprints;
• patches/: for every repository with uncommitted changes the git diff --binary of its files against its commit;
• untracked/: the untracked files that git does not ignore, up to 1 MB each;
• config/: copies of configs;
• PROVENANCE.md: the same for reading, and the steps to restore the metadata.

For a fingerprint that a subfolder already has, nothing is written and that subfolder is returned. Otherwise the record becomes the subfolder <yyyy-mm-ddTHHMMSSZ>-<short commit of metadata>, with -modified appended if a repository has uncommitted changes or untracked files, or a submodule is not at the commit its superproject records.

JuLeAna.production_log(ctrl::DataflowController)

Return a copy of the lines of the last CreateProduction of ctrl.

struct ProcessingConfig

Validated processing settings, built from a processing config file and the command line options of the dataflow driver.

Fields:

• runmode::Symbol: :local or :slurm.
• submit_jobs::Bool: whether the driver starts the workers itself.
• local_n::Int: number of local workers (config.runmode_settings.local.n).
• local_julia_settings::Vector{String}: Julia flags for local workers.
• slurm_flags::Vector{String}: SLURM flags for SLURM workers.
• slurm_julia_settings::Vector{String}: Julia flags for SLURM workers.
• env_variables::Dict{String,String}: environment variables for the driver and workers.
• periods::Union{Nothing,Vector{DataPeriod}}: selected periods; nothing selects all.
• runs::Union{Nothing,Vector{DataRun}}: selected runs; nothing selects all.
• analysis_runs_only::Bool: process only runs marked as analysis runs.
• only_partitions::Bool: process only partition-level processors.
• only_runs::Bool: process only run-level processors.
• max_parallel_steps::Int: maximum number of steps that run at the same time; 0 sets no limit.
• log_path::String: directory for log files; "" when not resolved.
• log_level::LogLevel: level of the records written to the log files.
• processor_log_level::Union{Nothing,LogLevel}: level at which the processors' own records are written to the log files; nothing takes log_level.
• processor_plot_level::Union{Nothing,Symbol}: :info or :debug, the plot level of all processors; nothing takes the plot_level of each processor.
• plot_formats::Vector{Symbol}: formats in which plots are saved, :png and :jls; :jls also saves the Makie figure with Serialization next to the PNG, under the same name with the extension .jls. The file can be read again by a session with the same package versions.
• idle_timeout_minutes::Float64: minutes without processing after which the worker allocations started by the interface are cancelled; 0 never cancels them.
• processors::Dict{Symbol,ProcessorSpec}: run-level processors.
• p_processors::Dict{Symbol,ProcessorSpec}: partition-level processors.

Besides the positional constructor and the config-file constructors documented below, ProcessingConfig(; processors, p_processors, runmode=:local, ...) builds a config from keywords named like the fields. Every field except the two processor tables has a default: empty flag and settings lists, no environment variables, periods and runs of nothing, all Bool fields false, max_parallel_steps = 0, local_n = 1, log_path = "", log_level = Logging.Info, processor_log_level = nothing, processor_plot_level = nothing, plot_formats = [:png] and idle_timeout_minutes = 20. Every constructor throws an ArgumentError when runmode is not :local or :slurm, processor_plot_level is not nothing, :info or :debug, plot_formats is empty, repeats an entry or holds anything but :png and :jls, local_n < 1, max_parallel_steps < 0, idle_timeout_minutes is negative or not finite, both only_runs and only_partitions are set, a table key differs from its spec's name or level, a name appears in both tables, a dependency names an unknown processor, or the dependencies form a cycle (see processor_graph).

See run_steps, partition_steps and check_processor_kwargs.

struct ProcessorSpec

Settings of one processor from the processors (run level) or p_processors (partition level) table of a processing config.

Fields:

• name::Symbol: name of the processor function.
• level::Symbol: :run for an entry of processors, :partition for an entry of p_processors.
• enabled::Bool: whether the processor is selected for execution by default.
• category::Symbol: data category the processor acts on, :cal or :phy.
• rank::Int: position in listings and tie rule: among processors that are ready to run, lower ranks start first.
• dependencies::Vector{Symbol}: processors of either level that must finish first.
• kwargs::Dict{Symbol,Any}: keyword arguments passed to the processor function.
• error_plots::Bool: whether a failed @process_step of the processor runs its on_error hook, which saves a debug plot; key error_plots of the processor entry, optional, false by default.
• log_level::LogLevel: level of the records of the processor in the log files; key log_level of the processor entry ("debug", "info", "warn" or "error"), required.
• plot_level::Symbol: :info saves the normal plots of the processor, :debug also its debug plots; key plot_level of the processor entry ("info" or "debug"), required.

The positional constructor takes the fields in the order listed above, ProcessorSpec(name, level, enabled, category, rank, dependencies, kwargs, error_plots, log_level, plot_level); ProcessorSpec(; name, level, category, rank, enabled=true, dependencies=Symbol[], kwargs=Dict{Symbol,Any}(), error_plots=false, log_level=Logging.Info, plot_level=:info) builds a spec from keywords. Every constructor throws an ArgumentError when level is not :run or :partition, category is not :cal or :phy, or the processor depends on itself.

run_steps(cfg::ProcessingConfig; enabled_only::Bool=true)

Return the names of the run-level processors of cfg as a Vector{Symbol}, sorted by rank (ties by name). With enabled_only = false the disabled processors are included.

ProcessorGraph

Dependency graph of the run-level and partition-level processors of a configuration, as returned by processor_graph.

Node i is the processor names[i] of level levels[i] (:run or :partition) and rank ranks[i]. An edge u → v of graph (a Graphs.SimpleDiGraph) means that v names u in its dependencies: u must finish before v, in a period, starts. Processors that no path connects can run at the same time.

processor_graph(cfg::ProcessingConfig)
processor_graph(specs)

Return the ProcessorGraph of all processors of cfg, enabled or not, or of specs, an iterable of objects with the properties name, level (:run or :partition), rank and dependencies like ProcessorSpec.

Throws an ArgumentError if a processor name occurs twice, a dependency names an unknown processor, or the dependencies form a cycle: processing would then wait forever. The error names the processors of the cycle.

topological_order(g::ProcessorGraph)

Return the processor names of g in an order in which every processor comes after all processors it waits for. Among processors that are free at the same point, run-level processors come first, then lower rank, then name.

required_processors(g::ProcessorGraph, names)

Return the set of the processors in names together with all processors they depend on, directly or through other processors.

dependent_processors(g::ProcessorGraph, names)

Return the set of the processors in names together with all processors that depend on one of them, directly or through other processors. This is the reverse of required_processors: b is in required_processors(g, [a]) exactly if a is in dependent_processors(g, [b]).

waited_processors(g::ProcessorGraph, selected)

Return a Dict that maps each processor in selected to the selected processors it waits for: the nearest selected processors before it in g, reached through processors that are not selected. A processor that is not selected passes on what it waits for, so selecting a processor without the ones it depends on keeps their requirements.

Each entry is a vector of pairs name => same_run. same_run is true for a run-level processor that a run-level processor waits for through run-level processors only: the wait is for the same run. Otherwise a run-level processor is waited for in all runs, and a partition-level processor in the same period.

layered_layout(g::ProcessorGraph)

Place the processors of g in layers for drawing: return a vector with one (layer, position) pair per node of g, both starting at 1. The layer of a processor is one more than the highest layer of the processors it depends on, so every edge points to a higher layer. Within a layer, processors are ordered to reduce edge crossings: by the mean position of their neighbors in the layers above, then below, a few times over.

partition_steps(cfg::ProcessingConfig; enabled_only::Bool=true)

Return the names of the partition-level processors of cfg as a Vector{Symbol}, sorted by rank (ties by name). With enabled_only = false the disabled processors are included.

check_processor_kwargs(cfg::ProcessingConfig; processors::AbstractDict{Symbol} = PROCESSORS)

Check that every enabled processor of cfg is registered in processors and that its function accepts all keyword arguments configured in its kwargs, plus the keywords run_processing may pass on its own: reprocess for every processor and only_first_period for partition processors. Throw an ArgumentError listing every mismatch; return nothing otherwise.

PROCESSORS::Dict{Symbol,Function}

Registry of the processor functions of the dataflow, keyed by processor name. The names are the keys of the processors and p_processors tables of a processing config. Every file in src/processors/ adds its processor with register_processor! when the module is defined.

register_processor!(f; output_tiers = Symbol[], detector_selection::Bool = true)

Add the processor function f to PROCESSORS under nameof(f) and return f. output_tiers are the data tiers the processor writes, for example [:jldsp]; detector_selection is false for a processor that does not process the detectors redo_detector names again. Both go to PROCESSOR_TRAITS for run_step. Throw an ArgumentError when a processor of that name is already registered.

PROCESSOR_TRAITS::Dict{Symbol,NamedTuple}

Properties of the registered processors that run_step needs, set by register_processor!: output_tiers, the data tiers the processor writes, and detector_selection, whether the processor processes the detectors that redo_detector names again and skips the others.

display_name(name::Symbol) -> String

The name of the processor name as the terminal interface shows it: without the process_ or p_process_ prefix and without the category suffix _cal or _phy, for example "decay_time" for :process_decay_time_cal. The category is shown separately. A name without the prefix, such as a report name, loses only the suffix.

RENAMED_PROCESSORS::Dict{Symbol,Symbol}

Former processor names mapped to their current names. A processing config that uses a former name as a key of processors or p_processors, or in a dependencies list, is read as if it used the current name.

parallel(f, iterator::AbstractArray, log_nt; timeout::Real=0, retry::Bool=false, process_name::AbstractString="")
parallel(iterator::AbstractArray, log_nt; kwargs...) do itr
    ...
end

Apply f to every element of iterator on the ParallelProcessingTools worker pool and return an array of itr => f(itr) pairs with the same shape and axes as iterator. The function comes first, as in map, so a do-block can supply it. Inside run_processing with a ProgressStore, each item's start, worker and outcome are recorded for the current step; otherwise a progress bar labeled process_name shows the finished items.

log_nt is the NamedTuple type of the processor's report rows. On success f returns a NamedTuple with at least the fields processed::Bool and log, a row of type log_nt (or a Dict of such rows), so that successful and failed items have the same shape for master_status and the report. An item whose result has skipped = true, which partition steps return for a detector left to the step of an earlier period, counts as not run in the progress.

An item whose f throws, or exceeds timeout seconds when timeout > 0, does not abort the batch: its pair is itr => (processed = false, log = failure_log(log_nt, itr, msg)). A timed-out item's worker process is terminated and not replaced; an item on the main process keeps running in the background. With retry = true, each item gets up to three attempts after a worker loss or a timeout; an error thrown by f is not retried.

Element types without a failure_log method for log_nt are rejected with an ArgumentError before any work starts.

Inside run_step, f sees the detector selection of run_step also on the workers (see redo_detector); with outputs = :temp, items whose detector is not selected are not run and are missing from the result. For items with a detector field, parallel throws an ArgumentError before any work starts if a chosen detector is not among them.

Inside run_step with debug = true, the items run one after the other in the calling task, without workers and without timeout.

Inside run_processing, a stop of the processing run takes effect here: items that have not been handed to a worker are not run, items already running finish, and parallel then throws ProcessingStopped instead of returning. When the run is stopped without waiting for running items, parallel throws right away and the running items go on in the background; their workers return to the pool when the items end.

Examples

using LegendDataManagement: FileKey, ProcessStatus
Row = NamedTuple{(:Filekey, :Status, :Events, :Error)}
filekeys = [FileKey("l200-p03-r000-cal-20230311T235840Z")]
results = parallel(filekeys, Row; process_name = "count events") do fk
    (processed = true, log = Row((fk, ProcessStatus(1), 1000, "")))
end
master_status(results)  # process_succeeded

failure_log(log_nt, itr, message::AbstractString)

Build the report row for an item of parallel that threw. log_nt is the processor's NamedTuple{names} row type and message the truncated error text. The status column is ProcessStatus(0) and every column without data is filled with "-".

Methods exist for FileKey items, giving (itr, status, "-"..., message), and for NamedTuple items with detector and usability fields, giving (detector, usability, [partition,] status, "-"..., message); the partition column is present when log_nt has a Partition column. A new item type for parallel needs its own failure_log method.

master_status(results...)

Combine the report rows of a processor into one status: process_succeeded, process_warning or process_failed.

Each element of results is a collection of itr => res pairs as returned by parallel, or nothing, which is ignored. res.log is a report row with a Status field, or a Dict of such rows. When a row has a usability column, the detector usability is read from itr.usability; otherwise the row counts as :on.

The result is process_failed if any row fails, otherwise process_warning if any row warns, otherwise process_succeeded. Without any rows the result is process_failed.

process_skipped

Report status of an item that a partition step leaves to the same step of the period in which the partition starts (only_first_period): Skipped as an HTML <span> string for Markdown reports, in gray. master_status counts it as a success.

detector_status(usability::Symbol)

Return usability as an HTML <span> string for Markdown reports, colored green for :on, yellow for :ac, red for :off and gray for any other value.

truncate_string(s::AbstractString, max_length::Integer=1000)

Return s if it has at most max_length characters, otherwise its first max_length characters followed by "...".

Examples

julia> truncate_string("process_dsp_cal", 7)
"process..."

truncate_error(e, max_length::Integer=1000)

Return the text of the root_exception of e, shortened with truncate_string to max_length characters.

load_qc_evaluator(l200::LegendData, filekey::FileKey)

Return the ML quality-cut evaluator for filekey. The ML quality cut is disabled: return missing and log an info message. The DSP processors skip the Haar wavelet classification when the evaluator is missing and set qc_label to -1.

redo_detector(det)::Bool

true if run_step runs a processor for chosen detectors and det is one of them. A processor processes such a detector again even if its parameters or output already exist, as with reprocess = true for this detector only.

detector_selected(det)::Bool

true unless run_step runs a processor for chosen detectors and det is not one of them. Processors that loop over detectors inside one item, such as the DSP processors, skip the detectors that are not selected.

write_pars(args...)

Write parameters with writelprops(args...), except inside run_step with outputs = :temp, where nothing is written.

write_validity(args...; kwargs...)

Write a validity entry with LegendDataManagement.LDMUtils.writevalidity, except inside run_step with outputs = :temp, where nothing is written.

make_par_dir(path) -> path

Create the parameter directory path like mkpath, except inside run_step with outputs = :temp, where nothing is created.

write_report(filename, report)

Write the processor report report to filename with writelreport. Inside run_step, report also goes into the report of the step.