12869112395

Committed 20 Jan 2025 01:29PM UTC coverage: 83.777% (+0.08%) from 83.698%

Build # 12869112395

Build Type

push

github

Committed by

psomhorst

Commit Message

Bump version: 1.5.2 → 1.6.0

Run Details

349 of 456 branches covered (76.54%)

Branch coverage included in aggregate %.

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

4 existing lines in 2 files now uncovered.

1381 of 1609 relevant lines covered (85.83%)

0.86 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).

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

EIT-ALIVE / eitprocessing / 12869112395

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