12400367746

Committed 18 Dec 2024 07:58PM UTC coverage: 81.598% (+0.05%) from 81.545%

Build # 12400367746

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.4.8 → 1.5.0

Run Details

347 of 470 branches covered (73.83%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

7 existing lines in 3 files now uncovered.

1369 of 1633 relevant lines covered (83.83%)

0.84 hits per line

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

94.0

/eitprocessing/datahandling/loading/__init__.py

from functools import reduce
from pathlib import Path

from eitprocessing.datahandling.datacollection import DataCollection
from eitprocessing.datahandling.eitdata import EITData, Vendor
from eitprocessing.datahandling.sequence import Sequence


def load_eit_data(
    path: str | Path | list[str | Path],
    vendor: Vendor | str,
    label: str | None = None,
    name: str | None = None,
    description: str = "",
    sample_frequency: float | None = None,
    first_frame: int = 0,
    max_frames: int | None = None,
) -> Sequence:
    """Load EIT data from path(s).

    Current limitations:
    - Dräger data is assumed to have a limited set of (Medibus) data. Newer additions that add data like pleural
    pressure are not yet supported.

    Args:
        path: relative or absolute path(s) to data file.
        vendor: vendor indicating the device used.
            Note: for load functions of specific vendors (e.g. `load_draeger_data`), this argument is defaulted to the
            correct vendor.
        label: short description of sequence for computer interpretation.
            Defaults to "Sequence_<unique_id>".
        name: short description of sequence for human interpretation.
            Defaults to the same value as label.
        description: long description of sequence for human interpretation.
        sample_frequency: sample frequency at which the data was recorded.
            No default for Draeger. Will be autodetected. Warns if autodetected differs from provided.
            Default for Timpel: 50
            Default for Sentec: 50.2
        first_frame: index of first frame to load.
            Defaults to 0.
        max_frames: maximum number of frames to load.
            The actual number of frames can be lower than this if this
            would surpass the final frame.

    Raises:
        NotImplementedError: is raised when there is no loading method for
        the given vendor.

    Returns:
        Sequence: a Sequence with the given label, name and description, containing the loaded data.

    Example:
    ```
    >>> sequence = load_eit_data(
    ...     ["path/to/file1", "path/to/file2"],
    ...     vendor="sentec",
    ...     label="initial_measurement"
    ... )
    >>> pixel_impedance = sequence.eit_data["raw"].pixel_impedance
    ```
    """
    from eitprocessing.datahandling.loading import draeger, sentec, timpel  # not in top level to avoid circular import

    vendor = _ensure_vendor(vendor)
    load_from_single_path = {
        Vendor.DRAEGER: draeger.load_from_single_path,
        Vendor.TIMPEL: timpel.load_from_single_path,
        Vendor.SENTEC: sentec.load_from_single_path,
    }[vendor]

    first_frame = _check_first_frame(first_frame)

    paths = EITData.ensure_path_list(path)

    eit_datasets: list[DataCollection] = []
    continuous_datasets: list[DataCollection] = []
    sparse_datasets: list[DataCollection] = []
    interval_datasets: list[DataCollection] = []

    for single_path in paths:
        single_path.resolve(strict=True)  # raise error if any file does not exist

    for single_path in paths:
        loaded_data = load_from_single_path(
            path=single_path,
            sample_frequency=sample_frequency,
            first_frame=first_frame,
            max_frames=max_frames,
        )

        eit_datasets.append(loaded_data["eitdata_collection"])
        continuous_datasets.append(loaded_data["continuousdata_collection"])
        sparse_datasets.append(loaded_data["sparsedata_collection"])
        interval_datasets.append(loaded_data["intervaldata_collection"])

    return Sequence(
        label=label,
        name=name,
        description=description,
        eit_data=reduce(DataCollection.concatenate, eit_datasets),
        continuous_data=reduce(DataCollection.concatenate, continuous_datasets),
        sparse_data=reduce(DataCollection.concatenate, sparse_datasets),
        interval_data=reduce(DataCollection.concatenate, interval_datasets),
    )


def _check_first_frame(first_frame: int | None) -> int:
    if first_frame is None:
        first_frame = 0
    if int(first_frame) != first_frame:
        msg = f"`first_frame` must be an int, but was given as {first_frame} (type: {type(first_frame)})"
        raise TypeError(msg)
    if first_frame < 0:
        msg = f"`first_frame` can not be negative, but was given as {first_frame}"
        raise ValueError(msg)
    return int(first_frame)


def _ensure_vendor(vendor: Vendor | str) -> Vendor:
    """Check whether loading method for vendor exists, and ensure it's a Vendor object."""
    try:
        return Vendor(vendor)
    except ValueError as e:
        msg = f"No loading method for {vendor} exists."
        raise NotImplementedError(msg) from e

1	from functools import reduce	1✔
2	from pathlib import Path	1✔
3
4	from eitprocessing.datahandling.datacollection import DataCollection	1✔
5	from eitprocessing.datahandling.eitdata import EITData, Vendor	1✔
6	from eitprocessing.datahandling.sequence import Sequence	1✔
7
8
9	def load_eit_data(	1✔
10	path: str \| Path \| list[str \| Path],
11	vendor: Vendor \| str,
12	label: str \| None = None,
13	name: str \| None = None,
14	description: str = "",
15	sample_frequency: float \| None = None,
16	first_frame: int = 0,
17	max_frames: int \| None = None,
18	) -> Sequence:
19	"""Load EIT data from path(s).
20
21	Current limitations:
22	- Dräger data is assumed to have a limited set of (Medibus) data. Newer additions that add data like pleural
23	pressure are not yet supported.
24
25	Args:
26	path: relative or absolute path(s) to data file.
27	vendor: vendor indicating the device used.
28	Note: for load functions of specific vendors (e.g. `load_draeger_data`), this argument is defaulted to the
29	correct vendor.
30	label: short description of sequence for computer interpretation.
31	Defaults to "Sequence_<unique_id>".
32	name: short description of sequence for human interpretation.
33	Defaults to the same value as label.
34	description: long description of sequence for human interpretation.
35	sample_frequency: sample frequency at which the data was recorded.
36	No default for Draeger. Will be autodetected. Warns if autodetected differs from provided.
37	Default for Timpel: 50
38	Default for Sentec: 50.2
39	first_frame: index of first frame to load.
40	Defaults to 0.
41	max_frames: maximum number of frames to load.
42	The actual number of frames can be lower than this if this
43	would surpass the final frame.
44
45	Raises:
46	NotImplementedError: is raised when there is no loading method for
47	the given vendor.
48
49	Returns:
50	Sequence: a Sequence with the given label, name and description, containing the loaded data.
51
52	Example:
53	```
54	>>> sequence = load_eit_data(
55	... ["path/to/file1", "path/to/file2"],
56	... vendor="sentec",
57	... label="initial_measurement"
58	... )
59	>>> pixel_impedance = sequence.eit_data["raw"].pixel_impedance
60	```
61	"""
62	from eitprocessing.datahandling.loading import draeger, sentec, timpel # not in top level to avoid circular import	1✔
63
64	vendor = _ensure_vendor(vendor)	1✔
65	load_from_single_path = {	1✔
66	Vendor.DRAEGER: draeger.load_from_single_path,
67	Vendor.TIMPEL: timpel.load_from_single_path,
68	Vendor.SENTEC: sentec.load_from_single_path,
69	}[vendor]
70
71	first_frame = _check_first_frame(first_frame)	1✔
72
73	paths = EITData.ensure_path_list(path)	1✔
74
75	eit_datasets: list[DataCollection] = []	1✔
76	continuous_datasets: list[DataCollection] = []	1✔
77	sparse_datasets: list[DataCollection] = []	1✔
78	interval_datasets: list[DataCollection] = []	1✔
79
80	for single_path in paths:	1✔
81	single_path.resolve(strict=True) # raise error if any file does not exist	1✔
82
83	for single_path in paths:	1✔
84	loaded_data = load_from_single_path(	1✔
85	path=single_path,
86	sample_frequency=sample_frequency,
87	first_frame=first_frame,
88	max_frames=max_frames,
89	)
90
91	eit_datasets.append(loaded_data["eitdata_collection"])	1✔
92	continuous_datasets.append(loaded_data["continuousdata_collection"])	1✔
93	sparse_datasets.append(loaded_data["sparsedata_collection"])	1✔
94	interval_datasets.append(loaded_data["intervaldata_collection"])	1✔
95
96	return Sequence(	1✔
97	label=label,
98	name=name,
99	description=description,
100	eit_data=reduce(DataCollection.concatenate, eit_datasets),
101	continuous_data=reduce(DataCollection.concatenate, continuous_datasets),
102	sparse_data=reduce(DataCollection.concatenate, sparse_datasets),
103	interval_data=reduce(DataCollection.concatenate, interval_datasets),
104	)
105
106
107	def _check_first_frame(first_frame: int \| None) -> int:	1✔
108	if first_frame is None:	1✔
109	first_frame = 0	1✔
110	if int(first_frame) != first_frame:	1✔
111	msg = f"`first_frame` must be an int, but was given as {first_frame} (type: {type(first_frame)})"	1✔
112	raise TypeError(msg)	1✔
113	if first_frame < 0:	1✔
114	msg = f"`first_frame` can not be negative, but was given as {first_frame}"	1✔
115	raise ValueError(msg)	1✔
116	return int(first_frame)	1✔
117
118
119	def _ensure_vendor(vendor: Vendor \| str) -> Vendor:	1✔
120	"""Check whether loading method for vendor exists, and ensure it's a Vendor object."""
121	try:	1✔
122	return Vendor(vendor)	1✔
UNCOV 123	except ValueError as e:	×
UNCOV 124	msg = f"No loading method for {vendor} exists."	×
UNCOV 125	raise NotImplementedError(msg) from e	×

EIT-ALIVE / eitprocessing / 12400367746

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