11953494901

Committed 21 Nov 2024 12:44PM UTC coverage: 92.339%. First build

Build # 11953494901

Build Type

Pull #437

github

Committed by

web-flow

Commit Message

Merge pull request #516 from OpenCOMPES/update_jupyter

Update jupyter

Pull Request Pull Request #437: Upgrade to V1

Run Details

2068 of 2189 new or added lines in 53 files covered. (94.47%)

7545 of 8171 relevant lines covered (92.34%)

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


# TODO: move to config
MULTI_INDEX = ["trainId", "pulseId", "electronId"]
PULSE_ALIAS = MULTI_INDEX[1]
FORMATS = ["per_electron", "per_pulse", "per_train"]


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", {})
    aux_alias = config_dataframe.get("aux_alias", "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,
            index,
            extend_aux,
        )
        return channels

    channels = []

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

    if formats:
        # If 'formats' is a list, check if all elements are valid.
        err_msg = (
            "Invalid format. Please choose from 'per_electron', 'per_pulse', 'per_train', 'all'."
        )
        for format_ in formats:
            if format_ not in FORMATS + ["all"]:
                raise ValueError(err_msg)

        # 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.
        if PULSE_ALIAS in available_channels:
            available_channels.remove(PULSE_ALIAS)

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

OpenCOMPES / sed / 11953494901

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