13006965508

Committed 28 Jan 2025 09:04AM UTC coverage: 91.66% (-0.5%) from 92.174%

Build # 13006965508

Build Type

Pull #534

github

Committed by

web-flow

Commit Message

Merge 69e45951e into 8ff004010

Pull Request Pull Request #534: Hextof lab loader

Run Details

66 of 117 new or added lines in 7 files covered. (56.41%)

75 existing lines in 4 files now uncovered.

7737 of 8441 relevant lines covered (91.66%)

0.92 hits per line

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

97.87

/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", None)
    formats_list = config_dataframe.get("formats")
    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 specified channels if they are present in available_channels.
        channels_to_remove = ["pulseId", "numEvents"]
        for channel in channels_to_remove:
            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", None)	1✔
27	formats_list = config_dataframe.get("formats")	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 specified channels if they are present in available_channels.
62	channels_to_remove = ["pulseId", "numEvents"]	1✔
63	for channel in channels_to_remove:	1✔
64	if channel in available_channels:	1✔
65	available_channels.remove(channel)	1✔
66
67	for format_ in formats:	1✔
68	# Gather channels based on the specified format(s).
69	channels.extend(	1✔
70	key
71	for key in available_channels
72	if channel_dict[key]["format"] == format_ and key != aux_alias
73	)
74	# Include 'dldAuxChannels' if the format is 'per_train' and extend_aux is True.
75	# Otherwise, include 'dldAux'.
76	if format_ == "per_train" and aux_alias in available_channels:	1✔
77	if extend_aux:	1✔
78	channels.extend(	1✔
79	channel_dict[aux_alias]["sub_channels"].keys(),
80	)
81	else:
82	channels.extend([aux_alias])	1✔
83
84	return channels	1✔
85
86
87	def get_dtypes(config_dataframe: dict, df_cols: list) -> dict:	1✔
88	"""Returns a dictionary of channels and their corresponding data types.
89	Currently Auxiliary channels are not included in the dtype dictionary.
90
91	Args:
92	config_dataframe (dict): The config dictionary containing the dataframe keys.
93	df_cols (list): A list of channels in the DataFrame.
94
95	Returns:
96	dict: A dictionary of channels and their corresponding data types.
97	"""
98	channels_dict = config_dataframe.get("channels", {})	1✔
99	aux_alias = config_dataframe.get("aux_alias", "dldAux")	1✔
100	dtypes = {}	1✔
101	for channel in df_cols:	1✔
102	try:	1✔
103	dtypes[channel] = channels_dict[channel].get("dtype")	1✔
104	except KeyError:	1✔
105	try:	1✔
106	dtypes[channel] = channels_dict[aux_alias][channel].get("dtype")	1✔
107	except KeyError:	1✔
108	dtypes[channel] = None	1✔
109	return dtypes	1✔
110
111
112	class InvalidFileError(Exception):	1✔
113	"""Raised when an H5 file is invalid due to missing keys defined in the config."""
114
115	def __init__(self, invalid_channels: list[str]):	1✔
116	self.invalid_channels = invalid_channels	1✔
117	super().__init__(	1✔
118	f"Channels not in file: {', '.join(invalid_channels)}. "
119	"If you are using the loader, set 'remove_invalid_files' to True to ignore these files",
120	)

OpenCOMPES / sed / 13006965508

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