13419398366

Committed 19 Feb 2025 06:09PM UTC coverage: 91.6% (-0.6%) from 92.174%

Build # 13419398366

Build Type

Pull #534

github

Committed by

web-flow

Commit Message

Merge df78f6964 into 6b927a2db

Pull Request Pull Request #534: Hextof lab loader

Run Details

71 of 124 new or added lines in 7 files covered. (57.26%)

3 existing lines in 1 file now uncovered.

7731 of 8440 relevant lines covered (91.6%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.83

/src/sed/loader/flash/utils.py

from __future__ import annotations


def get_channels(
    config_dataframe: dict = {},
    formats: str | list[str] = None,
    index: bool = False,
    extend_aux: bool = False,
) -> list[str]:
    """
    Returns a list of channels associated with the specified format(s).
    'all' returns all channels but 'pulseId' and 'dldAux' (if not extended).

    Args:
        config_dataframe (dict): The config dictionary containing the dataframe keys.
        formats (str | list[str]): The desired format(s)
        ('per_pulse', 'per_electron', 'per_train', 'all').
        index (bool): If True, includes channels from the multiindex.
        extend_aux (bool): If True, includes channels from the subchannels of the auxiliary channel.
                else just includes the auxiliary channel alias.

    Returns:
        List[str]: A list of channels with the specified format(s).
    """
    channel_dict = config_dataframe.get("channels", {})
    index_list = config_dataframe.get("index", ["trainId", "pulseId", "electronId"])
    formats_list = config_dataframe.get("formats", ["per_train", "per_pulse", "per_electron"])
    aux_alias = channel_dict.get("auxiliary", "dldAux")

    # If 'formats' is a single string, convert it to a list for uniform processing.
    if isinstance(formats, str):
        formats = [formats]

    # If 'formats' is a string "all", gather all possible formats.
    if formats == ["all"]:
        channels = get_channels(
            config_dataframe,
            formats_list,
            index,
            extend_aux,
        )
        return channels

    channels = []

    # Include channels from index_list if 'index' is True.
    if index:
        channels.extend(index_list)

    if formats:
        # If 'formats' is a list, check if all elements are valid.
        for format_ in formats:
            if format_ not in formats_list + ["all"]:
                raise ValueError(
                    f"Invalid format: {format_}. " f"Valid formats are: {formats_list + ['all']}",
                )

        # Get the available channels excluding 'pulseId'.
        available_channels = list(channel_dict.keys())
        # pulse alias is an index and should not be included in the list of channels.
        # Remove index channels if they are present in available_channels.
        for channel in index_list:
            if channel in available_channels:
                available_channels.remove(channel)

        for format_ in formats:
            # Gather channels based on the specified format(s).
            channels.extend(
                key
                for key in available_channels
                if channel_dict[key]["format"] == format_ and key != aux_alias
            )
            # Include 'dldAuxChannels' if the format is 'per_train' and extend_aux is True.
            # Otherwise, include 'dldAux'.
            if format_ == "per_train" and aux_alias in available_channels:
                if extend_aux:
                    channels.extend(
                        channel_dict[aux_alias]["sub_channels"].keys(),
                    )
                else:
                    channels.extend([aux_alias])

    return channels


def get_dtypes(config_dataframe: dict, df_cols: list) -> dict:
    """Returns a dictionary of channels and their corresponding data types.
    Currently Auxiliary channels are not included in the dtype dictionary.

    Args:
        config_dataframe (dict): The config dictionary containing the dataframe keys.
        df_cols (list): A list of channels in the DataFrame.

    Returns:
        dict: A dictionary of channels and their corresponding data types.
    """
    channels_dict = config_dataframe.get("channels", {})
    aux_alias = config_dataframe.get("aux_alias", "dldAux")
    dtypes = {}
    for channel in df_cols:
        try:
            dtypes[channel] = channels_dict[channel].get("dtype")
        except KeyError:
            try:
                dtypes[channel] = channels_dict[aux_alias][channel].get("dtype")
            except KeyError:
                dtypes[channel] = None
    return dtypes


class InvalidFileError(Exception):
    """Raised when an H5 file is invalid due to missing keys defined in the config."""

    def __init__(self, invalid_channels: list[str]):
        self.invalid_channels = invalid_channels
        super().__init__(
            f"Channels not in file: {', '.join(invalid_channels)}. "
            "If you are using the loader, set 'remove_invalid_files' to True to ignore these files",
        )

1	from __future__ import annotations	1✔
2
3
4	def get_channels(	1✔
5	config_dataframe: dict = {},
6	formats: str \| list[str] = None,
7	index: bool = False,
8	extend_aux: bool = False,
9	) -> list[str]:
10	"""
11	Returns a list of channels associated with the specified format(s).
12	'all' returns all channels but 'pulseId' and 'dldAux' (if not extended).
13
14	Args:
15	config_dataframe (dict): The config dictionary containing the dataframe keys.
16	formats (str \| list[str]): The desired format(s)
17	('per_pulse', 'per_electron', 'per_train', 'all').
18	index (bool): If True, includes channels from the multiindex.
19	extend_aux (bool): If True, includes channels from the subchannels of the auxiliary channel.
20	else just includes the auxiliary channel alias.
21
22	Returns:
23	List[str]: A list of channels with the specified format(s).
24	"""
25	channel_dict = config_dataframe.get("channels", {})	1✔
26	index_list = config_dataframe.get("index", ["trainId", "pulseId", "electronId"])	1✔
27	formats_list = config_dataframe.get("formats", ["per_train", "per_pulse", "per_electron"])	1✔
28	aux_alias = channel_dict.get("auxiliary", "dldAux")	1✔
29
30	# If 'formats' is a single string, convert it to a list for uniform processing.
31	if isinstance(formats, str):	1✔
32	formats = [formats]	1✔
33
34	# If 'formats' is a string "all", gather all possible formats.
35	if formats == ["all"]:	1✔
36	channels = get_channels(	1✔
37	config_dataframe,
38	formats_list,
39	index,
40	extend_aux,
41	)
42	return channels	1✔
43
44	channels = []	1✔
45
46	# Include channels from index_list if 'index' is True.
47	if index:	1✔
48	channels.extend(index_list)	1✔
49
50	if formats:	1✔
51	# If 'formats' is a list, check if all elements are valid.
52	for format_ in formats:	1✔
53	if format_ not in formats_list + ["all"]:	1✔
NEW 54	raise ValueError(	×
55	f"Invalid format: {format_}. " f"Valid formats are: {formats_list + ['all']}",
56	)
57
58	# Get the available channels excluding 'pulseId'.
59	available_channels = list(channel_dict.keys())	1✔
60	# pulse alias is an index and should not be included in the list of channels.
61	# Remove index channels if they are present in available_channels.
62	for channel in index_list:	1✔
63	if channel in available_channels:	1✔
64	available_channels.remove(channel)	1✔
65
66	for format_ in formats:	1✔
67	# Gather channels based on the specified format(s).
68	channels.extend(	1✔
69	key
70	for key in available_channels
71	if channel_dict[key]["format"] == format_ and key != aux_alias
72	)
73	# Include 'dldAuxChannels' if the format is 'per_train' and extend_aux is True.
74	# Otherwise, include 'dldAux'.
75	if format_ == "per_train" and aux_alias in available_channels:	1✔
76	if extend_aux:	1✔
77	channels.extend(	1✔
78	channel_dict[aux_alias]["sub_channels"].keys(),
79	)
80	else:
81	channels.extend([aux_alias])	1✔
82
83	return channels	1✔
84
85
86	def get_dtypes(config_dataframe: dict, df_cols: list) -> dict:	1✔
87	"""Returns a dictionary of channels and their corresponding data types.
88	Currently Auxiliary channels are not included in the dtype dictionary.
89
90	Args:
91	config_dataframe (dict): The config dictionary containing the dataframe keys.
92	df_cols (list): A list of channels in the DataFrame.
93
94	Returns:
95	dict: A dictionary of channels and their corresponding data types.
96	"""
97	channels_dict = config_dataframe.get("channels", {})	1✔
98	aux_alias = config_dataframe.get("aux_alias", "dldAux")	1✔
99	dtypes = {}	1✔
100	for channel in df_cols:	1✔
101	try:	1✔
102	dtypes[channel] = channels_dict[channel].get("dtype")	1✔
103	except KeyError:	1✔
104	try:	1✔
105	dtypes[channel] = channels_dict[aux_alias][channel].get("dtype")	1✔
106	except KeyError:	1✔
107	dtypes[channel] = None	1✔
108	return dtypes	1✔
109
110
111	class InvalidFileError(Exception):	1✔
112	"""Raised when an H5 file is invalid due to missing keys defined in the config."""
113
114	def __init__(self, invalid_channels: list[str]):	1✔
115	self.invalid_channels = invalid_channels	1✔
116	super().__init__(	1✔
117	f"Channels not in file: {', '.join(invalid_channels)}. "
118	"If you are using the loader, set 'remove_invalid_files' to True to ignore these files",
119	)

OpenCOMPES / sed / 13419398366

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous