lstautorta.plot_line_delay

Script plotting event_ids differences and r0dl1 rates for all observations of a night.

Functions:

Name	Description
`find_nigth_run_dirs`	Find the observation folders in a night's directory.
`log_uncaught_exceptions`	Makes all uncaught exception to be logged by the default logger.
`parse_line_subrun_dl1_events`	Read a DL1 file, to load the timestamps and event id information.
`parse_run_dl1_events`	Parse an observation DL1 files to load event ids and timestamps in a dataframe
`parse_run_log_files`	Parse the log files of the R0->DL1 process of a run to read the processing rates.
`parse_run_worker_nodes`	Parse the worker nodes of the R0->DL1 jobs during in an observation from hiperta_stream_start log file.
`process_night`	Parses DL1 and log files of an entire night to plot the event ids and processing rates wrt time
`save_line_delays_plot`	Plot and save the event id and the R0->DL1 processing rates wrt to time for an entire night

find_nigth_run_dirs

find_nigth_run_dirs(night_path)

Find the observation folders in a night's directory.

In LST, a night directory contains folders for each observation, plus a calibration and a logs folder.

Parameters:

Name	Type	Description	Default
`night_path`	`Path`	Path to the night's data folder containing all subruns.	required

Returns:

Type	Description
`List of Paths to all the obersvation folders in the night's directory.`

Source code in src/lstautorta/plot_line_delay.py

def find_nigth_run_dirs(night_path):
    """Find the observation folders in a night's directory.

    In LST, a night directory contains folders for each observation, plus a calibration and a logs folder.

    Parameters
    ----------
    night_path : pahtlib.Path
        Path to the night's data folder containing all subruns.

    Returns
    -------
        List of Paths to all the obersvation folders in the night's directory.
    """
    return [f for f in night_path.iterdir() if f.is_dir() and "calibration" not in f.stem and "logs" not in f.stem]

log_uncaught_exceptions

log_uncaught_exceptions()

Makes all uncaught exception to be logged by the default logger.

Keyboard exceptions and children classes are not logged so one can kill the program with ctr+C.

Source code in src/lstautorta/plot_line_delay.py

def log_uncaught_exceptions():
    """Makes all uncaught exception to be logged by the default logger.

    Keyboard exceptions and children classes are not logged so one can kill the program with ctr+C.
    """

    def handle_exception(exc_type, exc_value, exc_traceback):
        if not issubclass(exc_type, KeyboardInterrupt):
            logging.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))

        sys.__excepthook__(exc_type, exc_value, exc_traceback)

    sys.excepthook = handle_exception

parse_line_subrun_dl1_events

parse_line_subrun_dl1_events(line_subrun_dl1_filename, run_id, line_idx, worker_node, nb_retries, re_try_wait_time)

Read a DL1 file, to load the timestamps and event id information.

The separated timestamps_s and timestamp_qns are put back together to single datetime object (ns precision). The event_id diffs is also computed here, as the difference between the event id of an event with respect to the previous one. The first event is set to have an event_id_diff of 0. The "events from the future" are set a event_id_diff of 0, as well as the first event following an event from the future (because event ids of the event from the future are weird).

This function returns None if the file could not be read.

Parameters:

Name	Type	Description	Default
`line_subrun_dl1_filename`	`Path`	Path to the dl1 file.	required
`run_id`	`int`	Run ID of the events in the file. Written as field in the df.	required
`line_idx`	`int`	Line number of the process that wrote this file. Written as field in the df.	required
`worker_node`	`str`	Name of the worker node that processed the DL1 file to treat.	required
`nb_retries`	`int`	Number of times to re-try opening a file in case of error.	required
`re_try_wait_time`	`float`	Amount of time, in seconds, to wait before attempting to open a file again.	required

Returns:

Type	Description
`Pandas dataframe with the subrun event ids and timestamps (and line index, worker node).`

Source code in src/lstautorta/plot_line_delay.py

def parse_line_subrun_dl1_events(
    line_subrun_dl1_filename, run_id, line_idx, worker_node, nb_retries, re_try_wait_time
):
    """Read a DL1 file, to load the timestamps and event id information.

    The separated timestamps_s and timestamp_qns are put back together to single datetime object (ns precision).
    The event_id diffs is also computed here, as the difference between the event id of an event with respect to
    the previous one. The first event is set to have an event_id_diff of 0. The "events from the future" are
    set a event_id_diff of 0, as well as the first event following an event from the future (because event ids
    of the event from the future are weird).

    This function returns None if the file could not be read.

    Parameters
    ----------
    line_subrun_dl1_filename : pathlib.Path
        Path to the dl1 file.
    run_id : int
        Run ID of the events in the file. Written as field in the df.
    line_idx : int
        Line number of the process that wrote this file. Written as field in the df.
    worker_node : str
        Name of the worker node that processed the DL1 file to treat.
    nb_retries : int
        Number of times to re-try opening a file in case of error.
    re_try_wait_time : float
        Amount of time, in seconds, to wait before attempting to open a file again.

    Returns
    -------
        Pandas dataframe with the subrun event ids and timestamps (and line index, worker node).
    """
    # re-try mechanism because of fefs, sometimes we get errors even if the file is there.
    for _ in range(nb_retries):
        try:
            with tables.open_file(line_subrun_dl1_filename, "r") as dl1_f:
                event_ids = dl1_f.root.dl1.event.subarray.trigger.cols.event_id[:]
                timestamp_s = dl1_f.root.dl1.event.subarray.trigger.cols.timestamp_s[:]
                timestamp_qns = dl1_f.root.dl1.event.subarray.trigger.cols.timestamp_qns[:]
            break
        except Exception:
            time.sleep(re_try_wait_time)
    else:
        logging.warning(f"Could not open file {line_subrun_dl1_filename} !")
        return None

    # Conver to a single datetime object.
    trigger_time = (
        pd.to_datetime(timestamp_s, unit="s") + pd.to_timedelta(timestamp_qns // 4, unit="ns")
    ).to_series()
    # "Future events" have weird event ids, so set their event id diffs to 0
    # Also set the event id diff of the next event after a future event to 0 for same reason
    event_id_diffs = np.diff(event_ids, prepend=event_ids[0])
    event_id_diffs[ndimage.binary_dilation(trigger_time > pd.to_datetime("21000101"), np.array([0, 1, 1]))] = 0

    return pd.DataFrame(
        {
            "timestamp_s": timestamp_s,
            "timestamp_qns": timestamp_qns,
            "trigger_time": trigger_time,
            "event_id": event_ids,
            "event_id_diffs": event_id_diffs,
            "run_id": run_id,
            "line_idx": line_idx,
            "worker_node": worker_node,
        }
    )

parse_run_dl1_events

parse_run_dl1_events(dl1_dir_path, run_id, worker_nodes, nb_process, nb_retries, re_try_wait_time)

Parse an observation DL1 files to load event ids and timestamps in a dataframe

Paramters

dl1_dir_path : pathlib.Path Path to the folder containing the night's DL1 files. (all runs are in the same folder) run_id : int Run id of the run to analyse worker_nodes : list of str List of the worker nodes names of the R0->DL1 jobs of this run nb_process : int Number of processs to use to parse the runs files. nb_retries : int Number of times to re-try reading a DL1 file. re_try_wait_time : float Amount of time, in seconds, to wait before re-trying to open a file.

Returns:

Type	Description
`List of pandas dataframe with the events id and timestamps data per subrun.`

Source code in src/lstautorta/plot_line_delay.py

def parse_run_dl1_events(dl1_dir_path, run_id, worker_nodes, nb_process, nb_retries, re_try_wait_time):
    """Parse an observation DL1 files to load event ids and timestamps in a dataframe

    Paramters
    ---------
    dl1_dir_path : pathlib.Path
        Path to the folder containing the night's DL1 files. (all runs are in the same folder)
    run_id : int
        Run id of the run to analyse
    worker_nodes : list of str
        List of the worker nodes names of the R0->DL1 jobs of this run
    nb_process : int
        Number of processs to use to parse the runs files.
    nb_retries : int
        Number of times to re-try reading a DL1 file.
    re_try_wait_time : float
        Amount of time, in seconds, to wait before re-trying to open a file.

    Returns
    -------
        List of pandas dataframe with the events id and timestamps data per subrun.
    """
    files_dfs = []
    for line_idx in [0, 1, 2, 3]:
        with Pool(nb_process) as pool:
            line_dfs = pool.starmap(
                parse_line_subrun_dl1_events,
                [
                    (
                        line_subrun_filename,
                        run_id,
                        line_idx,
                        worker_nodes[line_idx],
                        nb_retries,
                        re_try_wait_time,
                    )
                    # For files in LST created by LSTAutoRTA:
                    # DL1 filename template: "dl1_${obs_id}_${run_id}_${tel_id}_${processIndex}_${threadIndex}_${fileIndex}.h5"
                    # obs_id = 0, tel_id = 1, threadIndex = 0 in LST when started by autoRTA.
                    for line_subrun_filename in dl1_dir_path.glob(f"dl1_0_{run_id}_1_{line_idx}*.h5")
                ],
            )

        files_dfs.extend([df for df in line_dfs if df is not None])

    return files_dfs

parse_run_log_files

parse_run_log_files(logs_dir, run_id, worker_nodes)

Parse the log files of the R0->DL1 process of a run to read the processing rates.

Parameters:

Name	Type	Description	Default
`logs_dir`	`Path`	Path to the folder containing the R0->DL1 jobs logs.	required
`run_id`	`int`	Run id of the run to parse	required
`worker_nodes`	`list of str`	List of the worker nodes names of the R0->DL1 jobs of this run	required

Returns:

Type	Description
`List of pandas dataframe with the R0->DL1 processing rates per line during the run.`

Source code in src/lstautorta/plot_line_delay.py

def parse_run_log_files(logs_dir, run_id, worker_nodes):
    """Parse the log files of the R0->DL1 process of a run to read the processing rates.

    Parameters
    ----------
    logs_dir : pathlib.Path
        Path to the folder containing the R0->DL1 jobs logs.
    run_id : int
        Run id of the run to parse
    worker_nodes : list of str
        List of the worker nodes names of the R0->DL1 jobs of this run

    Returns
    -------
        List of pandas dataframe with the R0->DL1 processing rates per line during the run.
    """
    pattern = re.compile(
        r"\[([0-9]+)\] : update_timer_message : "
        r"nbBadMessage = ([0-9]+) \(size in \[[^\]]*\]\), "
        r"nbEvent = ([0-9]+), "
        r"nbGoodEvent = ([0-9]+)"
    )
    logs_dfs = []
    for line_idx in [0, 1, 2, 3]:
        # log filename template: "r0_dl1_{run_id}_{line_id}.log"
        for log_file in logs_dir.glob(f"r0_dl1_{run_id}_{line_idx}.log"):
            timestamps = []
            nb_events = []
            nb_good_events = []
            event_per_sec = []
            good_event_per_s = []
            with open(log_file) as log_f:
                for log_line in log_f:
                    results = pattern.search(log_line)
                    if results:
                        timestamps.append(int(results.group(1)))
                        nb_events.append(int(results.group(3)))
                        nb_good_events.append(int(results.group(4)))

            logs_dfs.append(
                pd.DataFrame(
                    {
                        "timestamp": pd.to_datetime(timestamps, unit="s"),
                        "nbEvent": nb_events,
                        "run_id": run_id,
                        "line_idx": line_idx,
                        "worker_node": worker_nodes[line_idx],
                    }
                )
            )

    return logs_dfs

parse_run_worker_nodes

parse_run_worker_nodes(run_data_path)

Parse the worker nodes of the R0->DL1 jobs during in an observation from hiperta_stream_start log file.

hiperta_stream_start log file is parsed to read the slurm job ids of the R0->DL1 jobs. Then the worker nodes names are found by interogating sacct.

Paramters

run_data_path : pathlib.Path Path to the observation folder, containing the dl1, dl2 and dl3 directories and the hiperta_stream_start logs.

Returns:

Type	Description
`List of string: the names of the worker nodes, in the order of the lines.`

Source code in src/lstautorta/plot_line_delay.py

def parse_run_worker_nodes(run_data_path):
    """Parse the worker nodes of the R0->DL1 jobs during in an observation from hiperta_stream_start log file.

    hiperta_stream_start log file is parsed to read the slurm job ids of the R0->DL1 jobs. Then
    the worker nodes names are found by interogating sacct.

    Paramters
    ---------
    run_data_path : pathlib.Path
        Path to the observation folder, containing the dl1, dl2 and dl3 directories and the hiperta_stream_start logs.

    Returns
    -------
        List of string: the names of the worker nodes, in the order of the lines.
    """
    # job id file template: "jobids_hiperta_stream_{hiperta_start_timestamp}.txt"
    # timestamp format: 2023-11-23_05-45-51
    # There is another one for altaz job, not to be selected.
    job_id_files = [
        f for f in run_data_path.iterdir() if f.is_file() and "jobids" in f.stem and "altaz" not in f.stem
    ]
    if len(job_id_files) != 1:
        raise ValueError(f"Couldn't find jobid file in run dir {run_data_path!s}, got {job_id_files!s}")

    with open(job_id_files[0]) as job_file:
        job_ids = job_file.read().strip()

    sacct_process = sp.run(["sacct", "-j", job_ids, "--format", "NodeList"], capture_output=True, check=True)

    return [node_str.strip() for node_str in sacct_process.stdout.decode().split("\n")[2:]][:-1:2]

process_night

process_night(night, nb_process, force_reload, data_base_path, obs_relative_dl1_folder, obs_relative_data_folder, relative_logs_folder, output_file_prefix, rate_aggregation_str, nb_retries, re_try_wait_time)

Parses DL1 and log files of an entire night to plot the event ids and processing rates wrt time

Parameters:

Name	Type	Description	Default
`night`	`str`	Night data in format "YYYYMMDD"	required
`nb_process`	`int`	Number of process to use to parse the files	required
`force_reload`	`bool`	If true, the data and logs files are re-processed even if the dataframe is already saved at the output location.	required
`data_base_path`	`Path`	Path to the the folder containing the nights directories	required
`obs_relative_dl1_folder`	`Path`	Relative path to the folder containing dl1 files from the night directory	required
`obs_relative_data_folder`	`Path`	Relative path to the folder containing the data (dl1, dl2, dl3) subfolder from the night directory.	required
`relative_logs_folder`	`Path`	Relative path to the folder containing the night's logs from the night repository	required
`output_file_prefix`	`str`	Prefix to prepend to the DataFrame and Figure to save. The night str will be appended to make the full output file names.	required
`rate_aggregation_str`	`str`	Aggregation string in pandas time grouper syntax. For example for 10 seconds: "10S"	required
`nb_retries`	`int`	Number of times to re-try reading a DL1 file.	required
`re_try_wait_time`	`float`	Amount of time, in seconds, to wait before re-trying to open a file.	required

Source code in src/lstautorta/plot_line_delay.py

def process_night(
    night,
    nb_process,
    force_reload,
    data_base_path,
    obs_relative_dl1_folder,
    obs_relative_data_folder,
    relative_logs_folder,
    output_file_prefix,
    rate_aggregation_str,
    nb_retries,
    re_try_wait_time,
):
    """Parses DL1 and log files of an entire night to plot the event ids and processing rates wrt time

    Parameters
    ----------
    night : str
        Night data in format "YYYYMMDD"
    nb_process : int
        Number of process to use to parse the files
    force_reload : bool
        If true, the data and logs files are re-processed even if the dataframe is already saved
        at the output location.
    data_base_path : pathlib.Path
        Path to the the folder containing the nights directories
    obs_relative_dl1_folder : pathlib.Path
        Relative path to the folder containing dl1 files from the night directory
    obs_relative_data_folder : pathlib.Path
        Relative path to the folder containing the data (dl1, dl2, dl3) subfolder from the night directory.
    relative_logs_folder : pathlib.Path
        Relative path to the folder containing the night's logs from the night repository
    output_file_prefix : str
        Prefix to prepend to the DataFrame and Figure to save. The night str will be appended to
        make the full output file names.
    rate_aggregation_str : str
        Aggregation string in pandas time grouper syntax. For example for 10 seconds: "10S"
    nb_retries : int
        Number of times to re-try reading a DL1 file.
    re_try_wait_time : float
        Amount of time, in seconds, to wait before re-trying to open a file.
    """
    night_dir = data_base_path / night

    night_df_path = night_dir / Path(f"{output_file_prefix}{night}.h5")
    if night_df_path.exists() and not force_reload:
        event_df = pd.read_hdf(night_df_path, key="event_df", mode="r")
        log_df = pd.read_hdf(night_df_path, key="log_df", mode="r")
    else:
        events_dfs = []
        logs_dfs = []

        for run_dir_data_path in tqdm(find_nigth_run_dirs(night_dir)):
            try:
                run_id = int(run_dir_data_path.stem)
                worker_nodes = parse_run_worker_nodes(run_dir_data_path / obs_relative_data_folder)

                events_dfs.extend(
                    parse_run_dl1_events(
                        run_dir_data_path / obs_relative_dl1_folder,
                        run_id,
                        worker_nodes,
                        nb_process,
                        nb_retries,
                        re_try_wait_time,
                    )
                )
                logs_dfs.extend(parse_run_log_files(night_dir / relative_logs_folder, run_id, worker_nodes))
            except Exception:
                logging.error(f"Can not process run {run_dir_data_path}", exc_info=True)

        event_df = pd.concat(events_dfs, ignore_index=True)
        log_df = pd.concat(logs_dfs, ignore_index=True)

        event_df.to_hdf(night_df_path, key="event_df", mode="w")
        log_df.to_hdf(night_df_path, key="log_df", mode="a")

    save_line_delays_plot(event_df, log_df, night_dir / f"{output_file_prefix}{night}.pickle", rate_aggregation_str)

save_line_delays_plot

save_line_delays_plot(event_df, log_df, fig_pickle_path, rate_aggregation_str)

Plot and save the event id and the R0->DL1 processing rates wrt to time for an entire night

Paramters

event_df : pd.DataFrame DataFrame with events data. log_df : pd.DataFrame DataFrame with log files data (processing rate) fig_pickle_path : pathlib.Path Path to where the plot figure should be pickled. rate_aggregation_str : str Aggregation string in pandas time grouper syntax. For example for 10 seconds: "10S"

Source code in src/lstautorta/plot_line_delay.py

def save_line_delays_plot(event_df, log_df, fig_pickle_path, rate_aggregation_str):
    """Plot and save the event id and the R0->DL1 processing rates wrt to time for an entire night

    Paramters
    ---------
    event_df : pd.DataFrame
        DataFrame with events data.
    log_df : pd.DataFrame
        DataFrame with log files data (processing rate)
    fig_pickle_path : pathlib.Path
        Path to where the plot figure should be pickled.
    rate_aggregation_str : str
        Aggregation string in pandas time grouper syntax. For example for 10 seconds: "10S"
    """
    sns.set_theme()
    worker_nodes = sorted(log_df["worker_node"].unique().tolist())
    run_ids = sorted(log_df["run_id"].unique())
    run_starts = [log_df.loc[log_df["run_id"] == run_id, "timestamp"].min() for run_id in run_ids]
    fig, ax = plt.subplots(
        2,
        1,
        figsize=(15, 10),
        sharex=True,
    )
    sns.scatterplot(
        x="trigger_time",
        y="event_id_diffs",
        hue="worker_node",
        style="worker_node",
        data=event_df[
            (event_df["event_id_diffs"] != 4)  # don't plot events with normal event id diffs (too many of them)
            & (
                event_df["event_id_diffs"] != 0
            )  # those are: 1st events in subrun line file, or events after future events.
            & (event_df["trigger_time"] < pd.to_datetime("21000101"))  # don't plot events from the future
            & (event_df["event_id_diffs"] < 1e18)  # huge event ids when lines caught events of start of next run
        ],
        ax=ax[0],
        s=12,
        style_order=worker_nodes,
        hue_order=worker_nodes,
    )
    ax[0].set_yscale("log")
    sns.lineplot(
        x="timestamp",
        y="nbEvent",
        data=log_df.groupby([pd.Grouper(key="timestamp", freq=rate_aggregation_str), "worker_node"]).mean(),
        hue="worker_node",
        style="worker_node",
        units="run_id",
        markers=True,
        dashes=False,
        markersize=4,
        linewidth=1,
        sort=False,
        ax=ax[1],
        hue_order=worker_nodes,
        markeredgewidth=0.0,
    )
    for run_id, run_start in zip(run_ids, run_starts):
        ax[1].annotate(
            str(run_id),
            (run_start, 0),
            xytext=(0, 2),
            textcoords="offset fontsize",
            arrowprops=dict(arrowstyle="-|>", color="k"),
            fontsize="xx-small",
        )
    fig.tight_layout()

    with open(fig_pickle_path, "wb") as plot_fig_pickle:
        pickle.dump(fig, plot_fig_pickle)