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

98.55

/eitprocessing/datahandling/loading/draeger.py

from __future__ import annotations

import mmap
import sys
import warnings
from functools import partial
from typing import TYPE_CHECKING, NamedTuple

import numpy as np

from eitprocessing.datahandling.continuousdata import ContinuousData
from eitprocessing.datahandling.datacollection import DataCollection
from eitprocessing.datahandling.eitdata import EITData, Vendor
from eitprocessing.datahandling.event import Event
from eitprocessing.datahandling.intervaldata import IntervalData
from eitprocessing.datahandling.loading import load_eit_data
from eitprocessing.datahandling.loading.binreader import BinReader
from eitprocessing.datahandling.sparsedata import SparseData

if TYPE_CHECKING:
    from pathlib import Path

    from numpy.typing import NDArray

load_draeger_data = partial(load_eit_data, vendor=Vendor.DRAEGER)
NAN_VALUE_INDICATOR = -1e30


def load_from_single_path(
    path: Path,
    sample_frequency: float | None = None,
    first_frame: int = 0,
    max_frames: int | None = None,
) -> dict[str, DataCollection]:
    """Load Dräger EIT data from path."""
    file_size = path.stat().st_size

    frame_size: int
    medibus_fields: list

    # iterate over the supported file formats to find the frame size that matches the file size
    for _file_format_data in _bin_file_formats.values():
        frame_size = _file_format_data["frame_size"]
        if file_size % frame_size == 0:
            # if the file size is an integer multiple of the frame size, assume this is the correct format
            medibus_fields = _file_format_data["medibus_fields"]
            break
    else:
        msg = (
            f"File size {file_size} of file {path!s} does not match the supported *.bin file formats.\n"
            "Currently this package does not support loading files containing "
            "esophageal pressure or other non-standard data. "
            "Make sure this is a valid and uncorrupted Dräger data file."
        )
        raise OSError(msg)
    total_frames = file_size // frame_size

    if (f0 := first_frame) > (fn := total_frames):
        msg = f"Invalid input: `first_frame` ({f0}) is larger than the total number of frames in the file ({fn})."
        raise ValueError(msg)

    n_frames = min(total_frames - first_frame, max_frames or sys.maxsize)

    if max_frames and max_frames != n_frames:
        msg = (
            f"The number of frames requested ({max_frames}) is larger "
            f"than the available number ({n_frames}) of frames after "
            f"the first frame selected ({first_frame}, total frames: "
            f"{total_frames}).\n {n_frames} frames will be loaded."
        )
        warnings.warn(msg)

    # We need to load 1 frame before first actual frame to check if there is an event marker. Data for the pre-first
    # (dummy) frame will be removed from self at the end of this function.
    load_dummy_frame = first_frame > 0
    first_frame_to_load = first_frame - 1 if load_dummy_frame else 0

    pixel_impedance = np.zeros((n_frames, 32, 32))
    time = np.zeros((n_frames,))
    events: list[tuple[float, Event]] = []
    phases: list[tuple[float, int]] = []
    medibus_data = np.zeros((len(medibus_fields), n_frames), dtype=np.float32)

    with path.open("br") as fo, mmap.mmap(fo.fileno(), length=0, access=mmap.ACCESS_READ) as fh:
        fh.seek(first_frame_to_load * frame_size)
        reader = BinReader(fh)
        previous_marker = None

        first_index = -1 if load_dummy_frame else 0
        for index in range(first_index, n_frames):
            previous_marker = _read_frame(
                reader,
                index,
                time,
                pixel_impedance,
                medibus_data,
                len(medibus_fields),
                events,
                phases,
                previous_marker,
            )

    # time wraps around the number of seconds in a day
    time = np.unwrap(time, period=24 * 60 * 60)

    sample_frequency = _estimate_sample_frequency(time, sample_frequency)

    eit_data = EITData(
        vendor=Vendor.DRAEGER,
        path=path,
        sample_frequency=sample_frequency,
        nframes=n_frames,
        time=time,
        label="raw",
        pixel_impedance=pixel_impedance,
    )
    eitdata_collection = DataCollection(EITData, raw=eit_data)

    (
        continuousdata_collection,
        sparsedata_collection,
    ) = _convert_medibus_data(medibus_data, medibus_fields, time, sample_frequency)
    intervaldata_collection = DataCollection(IntervalData)
    # TODO: move some medibus data to sparse / interval
    # TODO: move phases and events to sparse / interval

    continuousdata_collection.add(
        ContinuousData(
            label="global_impedance_(raw)",
            name="Global impedance (raw)",
            unit="a.u.",
            category="impedance",
            derived_from=[eit_data],
            time=eit_data.time,
            values=eit_data.calculate_global_impedance(),
            sample_frequency=sample_frequency,
        ),
    )
    sparsedata_collection.add(
        SparseData(
            label="minvalues_(draeger)",
            name="Minimum values detected by Draeger device.",
            unit=None,
            category="minvalue",
            derived_from=[eit_data],
            time=np.array([t for t, d in phases if d == -1]),
        ),
    )
    sparsedata_collection.add(
        SparseData(
            label="maxvalues_(draeger)",
            name="Maximum values detected by Draeger device.",
            unit=None,
            category="maxvalue",
            derived_from=[eit_data],
            time=np.array([t for t, d in phases if d == 1]),
        ),
    )
    if len(events):
        time_, events_ = zip(*events, strict=True)
        time = np.array(time_)
        events = list(events_)
    else:
        time, events = np.array([]), []
    sparsedata_collection.add(
        SparseData(
            label="events_(draeger)",
            name="Events loaded from Draeger data",
            unit=None,
            category="event",
            derived_from=[eit_data],
            time=time,
            values=events,
        ),
    )

    return {
        "eitdata_collection": eitdata_collection,
        "continuousdata_collection": continuousdata_collection,
        "sparsedata_collection": sparsedata_collection,
        "intervaldata_collection": intervaldata_collection,
    }


def _estimate_sample_frequency(time: np.ndarray, sample_frequency: float | None) -> float:
    """Estimate the sample frequency from the time axis, and check with provided sample frequency."""
    estimated_sample_frequency = round((len(time) - 1) / (time[-1] - time[0]), 4)

    if sample_frequency is None:
        return estimated_sample_frequency

    if sample_frequency != estimated_sample_frequency:
        msg = (
            f"Provided sample frequency ({sample_frequency}) does not match "
            f"the estimated sample frequency ({estimated_sample_frequency})."
        )
        warnings.warn(msg, RuntimeWarning)

    return sample_frequency


def _convert_medibus_data(
    medibus_data: NDArray,
    medibus_fields: list,
    time: NDArray,
    sample_frequency: float,
) -> tuple[DataCollection, DataCollection]:
    continuousdata_collection = DataCollection(ContinuousData)
    sparsedata_collection = DataCollection(SparseData)

    for field_info, data in zip(medibus_fields, medibus_data, strict=True):
        data[data < NAN_VALUE_INDICATOR] = np.nan
        if field_info.continuous:
            continuous_data = ContinuousData(
                label=field_info.signal_name,
                name=field_info.signal_name,
                description=f"Continuous {field_info.signal_name} data loaded from file",
                unit=field_info.unit,
                time=time,
                values=data,
                category=field_info.signal_name,
                sample_frequency=sample_frequency,
            )
            continuous_data.lock()
            continuousdata_collection.add(continuous_data)

        else:
            # TODO parse sparse data
            ...

    return continuousdata_collection, sparsedata_collection


def _read_frame(
    reader: BinReader,
    index: int,
    time: NDArray,
    pixel_impedance: NDArray,
    medibus_data: NDArray,
    n_medibus_fields: int,
    events: list,
    phases: list,
    previous_marker: int | None,
) -> int:
    """Read frame by frame data from DRAEGER files.

    This method adds the loaded data to the provided arrays `time` and
    `pixel_impedance` and the provided lists `events` and `phases` when the
    index is non-negative. When the index is negative, no data is saved. In
    any case, the event marker is returned.
    """
    frame_time = round(reader.float64() * 24 * 60 * 60, 3)
    _ = reader.float32()
    frame_pixel_impedance = reader.npfloat32(length=1024)
    frame_pixel_impedance = np.reshape(frame_pixel_impedance, (32, 32), "C")
    min_max_flag = reader.int32()
    event_marker = reader.int32()
    event_text = reader.string(length=30)
    timing_error = reader.int32()

    frame_medibus_data = reader.npfloat32(length=n_medibus_fields)

    if index < 0:
        # do not keep any loaded data, just return the event marker
        return event_marker

    time[index] = frame_time
    pixel_impedance[index, :, :] = frame_pixel_impedance
    medibus_data[:, index] = frame_medibus_data

    # The event marker stays the same until the next event occurs.
    # Therefore, check whether the event marker has changed with
    # respect to the most recent event. If so, create a new event.
    if ((previous_marker is not None) and (event_marker > previous_marker)) or (index == 0 and event_text):
        events.append((frame_time, Event(event_marker, event_text)))
    if timing_error:
        warnings.warn("A timing error was encountered during loading.")
        # TODO: expand on what timing errors are in some documentation.
    if min_max_flag in (1, -1):
        phases.append((frame_time, min_max_flag))

    return event_marker


class _MedibusField(NamedTuple):
    signal_name: str
    unit: str
    continuous: bool


_bin_file_formats = {
    "original": {
        "frame_size": 4358,
        "medibus_fields": [
            _MedibusField("airway pressure", "mbar", True),
            _MedibusField("flow", "L/min", True),
            _MedibusField("volume", "mL", True),
            _MedibusField("CO2 (%)", "%", True),
            _MedibusField("CO2 (kPa)", "kPa", True),
            _MedibusField("CO2 (mmHg)", "mmHg", True),
            _MedibusField("dynamic compliance", "mL/mbar", False),
            _MedibusField("resistance", "mbar/L/s", False),
            _MedibusField("r^2", "", False),
            _MedibusField("spontaneous inspiratory time", "s", False),
            _MedibusField("minimal pressure", "mbar", False),
            _MedibusField("P0.1", "mbar", False),
            _MedibusField("mean pressure", "mbar", False),
            _MedibusField("plateau pressure", "mbar", False),
            _MedibusField("PEEP", "mbar", False),
            _MedibusField("intrinsic PEEP", "mbar", False),
            _MedibusField("mandatory respiratory rate", "/min", False),
            _MedibusField("mandatory minute volume", "L/min", False),
            _MedibusField("peak inspiratory pressure", "mbar", False),
            _MedibusField("mandatory tidal volume", "L", False),
            _MedibusField("spontaneous tidal volume", "L", False),
            _MedibusField("trapped volume", "mL", False),
            _MedibusField("mandatory expiratory tidal volume", "mL", False),
            _MedibusField("spontaneous expiratory tidal volume", "mL", False),
            _MedibusField("mandatory inspiratory tidal volume", "mL", False),
            _MedibusField("tidal volume", "mL", False),
            _MedibusField("spontaneous inspiratory tidal volume", "mL", False),
            _MedibusField("negative inspiratory force", "mbar", False),
            _MedibusField("leak minute volume", "L/min", False),
            _MedibusField("leak percentage", "%", False),
            _MedibusField("spontaneous respiratory rate", "/min", False),
            _MedibusField("percentage of spontaneous minute volume", "%", False),
            _MedibusField("spontaneous minute volume", "L/min", False),
            _MedibusField("minute volume", "L/min", False),
            _MedibusField("airway temperature", "degrees C", False),
            _MedibusField("rapid shallow breating index", "1/min/L", False),
            _MedibusField("respiratory rate", "/min", False),
            _MedibusField("inspiratory:expiratory ratio", "", False),
            _MedibusField("CO2 flow", "mL/min", False),
            _MedibusField("dead space volume", "mL", False),
            _MedibusField("percentage dead space of expiratory tidal volume", "%", False),
            _MedibusField("end-tidal CO2", "%", False),
            _MedibusField("end-tidal CO2", "kPa", False),
            _MedibusField("end-tidal CO2", "mmHg", False),
            _MedibusField("fraction inspired O2", "%", False),
            _MedibusField("spontaneous inspiratory:expiratory ratio", "", False),
            _MedibusField("elastance", "mbar/L", False),
            _MedibusField("time constant", "s", False),
            _MedibusField(
                "ratio between upper 20% pressure range and total dynamic compliance",
                "",
                False,
            ),
            _MedibusField("end-inspiratory pressure", "mbar", False),
            _MedibusField("expiratory tidal volume", "mL", False),
            _MedibusField("time at low pressure", "s", False),
        ],
    },
    "pressure_pod": {
        "frame_size": 4382,
        "medibus_fields": [
            _MedibusField("airway pressure", "mbar", True),
            _MedibusField("flow", "L/min", True),
            _MedibusField("volume", "mL", True),
            _MedibusField("CO2 (%)", "%", True),
            _MedibusField("CO2 (kPa)", "kPa", True),
            _MedibusField("CO2 (mmHg)", "mmHg", True),
            _MedibusField("dynamic compliance", "mL/mbar", False),
            _MedibusField("resistance", "mbar/L/s", False),
            _MedibusField("r^2", "", False),
            _MedibusField("spontaneous inspiratory time", "s", False),
            _MedibusField("minimal pressure", "mbar", False),
            _MedibusField("P0.1", "mbar", False),
            _MedibusField("mean pressure", "mbar", False),
            _MedibusField("plateau pressure", "mbar", False),
            _MedibusField("PEEP", "mbar", False),
            _MedibusField("intrinsic PEEP", "mbar", False),
            _MedibusField("mandatory respiratory rate", "/min", False),
            _MedibusField("mandatory minute volume", "L/min", False),
            _MedibusField("peak inspiratory pressure", "mbar", False),
            _MedibusField("mandatory tidal volume", "L", False),
            _MedibusField("spontaneous tidal volume", "L", False),
            _MedibusField("trapped volume", "mL", False),
            _MedibusField("mandatory expiratory tidal volume", "mL", False),
            _MedibusField("spontaneous expiratory tidal volume", "mL", False),
            _MedibusField("mandatory inspiratory tidal volume", "mL", False),
            _MedibusField("tidal volume", "mL", False),
            _MedibusField("spontaneous inspiratory tidal volume", "mL", False),
            _MedibusField("negative inspiratory force", "mbar", False),
            _MedibusField("leak minute volume", "L/min", False),
            _MedibusField("leak percentage", "%", False),
            _MedibusField("spontaneous respiratory rate", "/min", False),
            _MedibusField("percentage of spontaneous minute volume", "%", False),
            _MedibusField("spontaneous minute volume", "L/min", False),
            _MedibusField("minute volume", "L/min", False),
            _MedibusField("airway temperature", "degrees C", False),
            _MedibusField("rapid shallow breating index", "1/min/L", False),
            _MedibusField("respiratory rate", "/min", False),
            _MedibusField("inspiratory:expiratory ratio", "", False),
            _MedibusField("CO2 flow", "mL/min", False),
            _MedibusField("dead space volume", "mL", False),
            _MedibusField("percentage dead space of expiratory tidal volume", "%", False),
            _MedibusField("end-tidal CO2", "%", False),
            _MedibusField("end-tidal CO2", "kPa", False),
            _MedibusField("end-tidal CO2", "mmHg", False),
            _MedibusField("fraction inspired O2", "%", False),
            _MedibusField("spontaneous inspiratory:expiratory ratio", "", False),
            _MedibusField("elastance", "mbar/L", False),
            _MedibusField("time constant", "s", False),
            _MedibusField(
                "ratio between upper 20% pressure range and total dynamic compliance",
                "",
                False,
            ),
            _MedibusField("end-inspiratory pressure", "mbar", False),
            _MedibusField("expiratory tidal volume", "mL", False),
            _MedibusField("high pressure", "mbar", False),
            _MedibusField("low pressure", "mbar", False),
            _MedibusField("time at low pressure", "s", False),
            _MedibusField("airway pressure (pod)", "mbar", True),
            _MedibusField("esophageal pressure (pod)", "mbar", True),
            _MedibusField("transpulmonary pressure (pod)", "mbar", True),
            _MedibusField("gastric pressure/auxiliary pressure (pod)", "mbar", True),
        ],
    },
}

1	from __future__ import annotations	1✔
2
3	import mmap	1✔
4	import sys	1✔
5	import warnings	1✔
6	from functools import partial	1✔
7	from typing import TYPE_CHECKING, NamedTuple	1✔
8
9	import numpy as np	1✔
10
11	from eitprocessing.datahandling.continuousdata import ContinuousData	1✔
12	from eitprocessing.datahandling.datacollection import DataCollection	1✔
13	from eitprocessing.datahandling.eitdata import EITData, Vendor	1✔
14	from eitprocessing.datahandling.event import Event	1✔
15	from eitprocessing.datahandling.intervaldata import IntervalData	1✔
16	from eitprocessing.datahandling.loading import load_eit_data	1✔
17	from eitprocessing.datahandling.loading.binreader import BinReader	1✔
18	from eitprocessing.datahandling.sparsedata import SparseData	1✔
19
20	if TYPE_CHECKING:
21	from pathlib import Path
22
23	from numpy.typing import NDArray
24
25	load_draeger_data = partial(load_eit_data, vendor=Vendor.DRAEGER)	1✔
26	NAN_VALUE_INDICATOR = -1e30	1✔
27
28
29	def load_from_single_path(	1✔
30	path: Path,
31	sample_frequency: float \| None = None,
32	first_frame: int = 0,
33	max_frames: int \| None = None,
34	) -> dict[str, DataCollection]:
35	"""Load Dräger EIT data from path."""
36	file_size = path.stat().st_size	1✔
37
38	frame_size: int
39	medibus_fields: list
40
41	# iterate over the supported file formats to find the frame size that matches the file size
42	for _file_format_data in _bin_file_formats.values():	1✔
43	frame_size = _file_format_data["frame_size"]	1✔
44	if file_size % frame_size == 0:	1✔
45	# if the file size is an integer multiple of the frame size, assume this is the correct format
46	medibus_fields = _file_format_data["medibus_fields"]	1✔
47	break	1✔
48	else:
49	msg = (	1✔
50	f"File size {file_size} of file {path!s} does not match the supported *.bin file formats.\n"
51	"Currently this package does not support loading files containing "
52	"esophageal pressure or other non-standard data. "
53	"Make sure this is a valid and uncorrupted Dräger data file."
54	)
55	raise OSError(msg)	1✔
56	total_frames = file_size // frame_size	1✔
57
58	if (f0 := first_frame) > (fn := total_frames):	1✔
59	msg = f"Invalid input: `first_frame` ({f0}) is larger than the total number of frames in the file ({fn})."	1✔
60	raise ValueError(msg)	1✔
61
62	n_frames = min(total_frames - first_frame, max_frames or sys.maxsize)	1✔
63
64	if max_frames and max_frames != n_frames:	1✔
65	msg = (	1✔
66	f"The number of frames requested ({max_frames}) is larger "
67	f"than the available number ({n_frames}) of frames after "
68	f"the first frame selected ({first_frame}, total frames: "
69	f"{total_frames}).\n {n_frames} frames will be loaded."
70	)
71	warnings.warn(msg)	1✔
72
73	# We need to load 1 frame before first actual frame to check if there is an event marker. Data for the pre-first
74	# (dummy) frame will be removed from self at the end of this function.
75	load_dummy_frame = first_frame > 0	1✔
76	first_frame_to_load = first_frame - 1 if load_dummy_frame else 0	1✔
77
78	pixel_impedance = np.zeros((n_frames, 32, 32))	1✔
79	time = np.zeros((n_frames,))	1✔
80	events: list[tuple[float, Event]] = []	1✔
81	phases: list[tuple[float, int]] = []	1✔
82	medibus_data = np.zeros((len(medibus_fields), n_frames), dtype=np.float32)	1✔
83
84	with path.open("br") as fo, mmap.mmap(fo.fileno(), length=0, access=mmap.ACCESS_READ) as fh:	1✔
85	fh.seek(first_frame_to_load * frame_size)	1✔
86	reader = BinReader(fh)	1✔
87	previous_marker = None	1✔
88
89	first_index = -1 if load_dummy_frame else 0	1✔
90	for index in range(first_index, n_frames):	1✔
91	previous_marker = _read_frame(	1✔
92	reader,
93	index,
94	time,
95	pixel_impedance,
96	medibus_data,
97	len(medibus_fields),
98	events,
99	phases,
100	previous_marker,
101	)
102
103	# time wraps around the number of seconds in a day
104	time = np.unwrap(time, period=24 * 60 * 60)	1✔
105
106	sample_frequency = _estimate_sample_frequency(time, sample_frequency)	1✔
107
108	eit_data = EITData(	1✔
109	vendor=Vendor.DRAEGER,
110	path=path,
111	sample_frequency=sample_frequency,
112	nframes=n_frames,
113	time=time,
114	label="raw",
115	pixel_impedance=pixel_impedance,
116	)
117	eitdata_collection = DataCollection(EITData, raw=eit_data)	1✔
118
119	(	1✔
120	continuousdata_collection,
121	sparsedata_collection,
122	) = _convert_medibus_data(medibus_data, medibus_fields, time, sample_frequency)
123	intervaldata_collection = DataCollection(IntervalData)	1✔
124	# TODO: move some medibus data to sparse / interval
125	# TODO: move phases and events to sparse / interval
126
127	continuousdata_collection.add(	1✔
128	ContinuousData(
129	label="global_impedance_(raw)",
130	name="Global impedance (raw)",
131	unit="a.u.",
132	category="impedance",
133	derived_from=[eit_data],
134	time=eit_data.time,
135	values=eit_data.calculate_global_impedance(),
136	sample_frequency=sample_frequency,
137	),
138	)
139	sparsedata_collection.add(	1✔
140	SparseData(
141	label="minvalues_(draeger)",
142	name="Minimum values detected by Draeger device.",
143	unit=None,
144	category="minvalue",
145	derived_from=[eit_data],
146	time=np.array([t for t, d in phases if d == -1]),
147	),
148	)
149	sparsedata_collection.add(	1✔
150	SparseData(
151	label="maxvalues_(draeger)",
152	name="Maximum values detected by Draeger device.",
153	unit=None,
154	category="maxvalue",
155	derived_from=[eit_data],
156	time=np.array([t for t, d in phases if d == 1]),
157	),
158	)
159	if len(events):	1✔
160	time_, events_ = zip(*events, strict=True)	1✔
161	time = np.array(time_)	1✔
162	events = list(events_)	1✔
163	else:
164	time, events = np.array([]), []	1✔
165	sparsedata_collection.add(	1✔
166	SparseData(
167	label="events_(draeger)",
168	name="Events loaded from Draeger data",
169	unit=None,
170	category="event",
171	derived_from=[eit_data],
172	time=time,
173	values=events,
174	),
175	)
176
177	return {	1✔
178	"eitdata_collection": eitdata_collection,
179	"continuousdata_collection": continuousdata_collection,
180	"sparsedata_collection": sparsedata_collection,
181	"intervaldata_collection": intervaldata_collection,
182	}
183
184
185	def _estimate_sample_frequency(time: np.ndarray, sample_frequency: float \| None) -> float:	1✔
186	"""Estimate the sample frequency from the time axis, and check with provided sample frequency."""
187	estimated_sample_frequency = round((len(time) - 1) / (time[-1] - time[0]), 4)	1✔
188
189	if sample_frequency is None:	1✔
190	return estimated_sample_frequency	1✔
191
192	if sample_frequency != estimated_sample_frequency:	1✔
193	msg = (	1✔
194	f"Provided sample frequency ({sample_frequency}) does not match "
195	f"the estimated sample frequency ({estimated_sample_frequency})."
196	)
197	warnings.warn(msg, RuntimeWarning)	1✔
198
199	return sample_frequency	1✔
200
201
202	def _convert_medibus_data(	1✔
203	medibus_data: NDArray,
204	medibus_fields: list,
205	time: NDArray,
206	sample_frequency: float,
207	) -> tuple[DataCollection, DataCollection]:
208	continuousdata_collection = DataCollection(ContinuousData)	1✔
209	sparsedata_collection = DataCollection(SparseData)	1✔
210
211	for field_info, data in zip(medibus_fields, medibus_data, strict=True):	1✔
212	data[data < NAN_VALUE_INDICATOR] = np.nan	1✔
213	if field_info.continuous:	1✔
214	continuous_data = ContinuousData(	1✔
215	label=field_info.signal_name,
216	name=field_info.signal_name,
217	description=f"Continuous {field_info.signal_name} data loaded from file",
218	unit=field_info.unit,
219	time=time,
220	values=data,
221	category=field_info.signal_name,
222	sample_frequency=sample_frequency,
223	)
224	continuous_data.lock()	1✔
225	continuousdata_collection.add(continuous_data)	1✔
226
227	else:
228	# TODO parse sparse data
229	...	1✔
230
231	return continuousdata_collection, sparsedata_collection	1✔
232
233
234	def _read_frame(	1✔
235	reader: BinReader,
236	index: int,
237	time: NDArray,
238	pixel_impedance: NDArray,
239	medibus_data: NDArray,
240	n_medibus_fields: int,
241	events: list,
242	phases: list,
243	previous_marker: int \| None,
244	) -> int:
245	"""Read frame by frame data from DRAEGER files.
246
247	This method adds the loaded data to the provided arrays `time` and
248	`pixel_impedance` and the provided lists `events` and `phases` when the
249	index is non-negative. When the index is negative, no data is saved. In
250	any case, the event marker is returned.
251	"""
252	frame_time = round(reader.float64() * 24 * 60 * 60, 3)	1✔
253	_ = reader.float32()	1✔
254	frame_pixel_impedance = reader.npfloat32(length=1024)	1✔
255	frame_pixel_impedance = np.reshape(frame_pixel_impedance, (32, 32), "C")	1✔
256	min_max_flag = reader.int32()	1✔
257	event_marker = reader.int32()	1✔
258	event_text = reader.string(length=30)	1✔
259	timing_error = reader.int32()	1✔
260
261	frame_medibus_data = reader.npfloat32(length=n_medibus_fields)	1✔
262
263	if index < 0:	1✔
264	# do not keep any loaded data, just return the event marker
265	return event_marker	1✔
266
267	time[index] = frame_time	1✔
268	pixel_impedance[index, :, :] = frame_pixel_impedance	1✔
269	medibus_data[:, index] = frame_medibus_data	1✔
270
271	# The event marker stays the same until the next event occurs.
272	# Therefore, check whether the event marker has changed with
273	# respect to the most recent event. If so, create a new event.
274	if ((previous_marker is not None) and (event_marker > previous_marker)) or (index == 0 and event_text):	1✔
275	events.append((frame_time, Event(event_marker, event_text)))	1✔
276	if timing_error:	1!
UNCOV 277	warnings.warn("A timing error was encountered during loading.")	×
278	# TODO: expand on what timing errors are in some documentation.
279	if min_max_flag in (1, -1):	1✔
280	phases.append((frame_time, min_max_flag))	1✔
281
282	return event_marker	1✔
283
284
285	class _MedibusField(NamedTuple):	1✔
286	signal_name: str	1✔
287	unit: str	1✔
288	continuous: bool	1✔
289
290
291	_bin_file_formats = {	1✔
292	"original": {
293	"frame_size": 4358,
294	"medibus_fields": [
295	_MedibusField("airway pressure", "mbar", True),
296	_MedibusField("flow", "L/min", True),
297	_MedibusField("volume", "mL", True),
298	_MedibusField("CO2 (%)", "%", True),
299	_MedibusField("CO2 (kPa)", "kPa", True),
300	_MedibusField("CO2 (mmHg)", "mmHg", True),
301	_MedibusField("dynamic compliance", "mL/mbar", False),
302	_MedibusField("resistance", "mbar/L/s", False),
303	_MedibusField("r^2", "", False),
304	_MedibusField("spontaneous inspiratory time", "s", False),
305	_MedibusField("minimal pressure", "mbar", False),
306	_MedibusField("P0.1", "mbar", False),
307	_MedibusField("mean pressure", "mbar", False),
308	_MedibusField("plateau pressure", "mbar", False),
309	_MedibusField("PEEP", "mbar", False),
310	_MedibusField("intrinsic PEEP", "mbar", False),
311	_MedibusField("mandatory respiratory rate", "/min", False),
312	_MedibusField("mandatory minute volume", "L/min", False),
313	_MedibusField("peak inspiratory pressure", "mbar", False),
314	_MedibusField("mandatory tidal volume", "L", False),
315	_MedibusField("spontaneous tidal volume", "L", False),
316	_MedibusField("trapped volume", "mL", False),
317	_MedibusField("mandatory expiratory tidal volume", "mL", False),
318	_MedibusField("spontaneous expiratory tidal volume", "mL", False),
319	_MedibusField("mandatory inspiratory tidal volume", "mL", False),
320	_MedibusField("tidal volume", "mL", False),
321	_MedibusField("spontaneous inspiratory tidal volume", "mL", False),
322	_MedibusField("negative inspiratory force", "mbar", False),
323	_MedibusField("leak minute volume", "L/min", False),
324	_MedibusField("leak percentage", "%", False),
325	_MedibusField("spontaneous respiratory rate", "/min", False),
326	_MedibusField("percentage of spontaneous minute volume", "%", False),
327	_MedibusField("spontaneous minute volume", "L/min", False),
328	_MedibusField("minute volume", "L/min", False),
329	_MedibusField("airway temperature", "degrees C", False),
330	_MedibusField("rapid shallow breating index", "1/min/L", False),
331	_MedibusField("respiratory rate", "/min", False),
332	_MedibusField("inspiratory:expiratory ratio", "", False),
333	_MedibusField("CO2 flow", "mL/min", False),
334	_MedibusField("dead space volume", "mL", False),
335	_MedibusField("percentage dead space of expiratory tidal volume", "%", False),
336	_MedibusField("end-tidal CO2", "%", False),
337	_MedibusField("end-tidal CO2", "kPa", False),
338	_MedibusField("end-tidal CO2", "mmHg", False),
339	_MedibusField("fraction inspired O2", "%", False),
340	_MedibusField("spontaneous inspiratory:expiratory ratio", "", False),
341	_MedibusField("elastance", "mbar/L", False),
342	_MedibusField("time constant", "s", False),
343	_MedibusField(
344	"ratio between upper 20% pressure range and total dynamic compliance",
345	"",
346	False,
347	),
348	_MedibusField("end-inspiratory pressure", "mbar", False),
349	_MedibusField("expiratory tidal volume", "mL", False),
350	_MedibusField("time at low pressure", "s", False),
351	],
352	},
353	"pressure_pod": {
354	"frame_size": 4382,
355	"medibus_fields": [
356	_MedibusField("airway pressure", "mbar", True),
357	_MedibusField("flow", "L/min", True),
358	_MedibusField("volume", "mL", True),
359	_MedibusField("CO2 (%)", "%", True),
360	_MedibusField("CO2 (kPa)", "kPa", True),
361	_MedibusField("CO2 (mmHg)", "mmHg", True),
362	_MedibusField("dynamic compliance", "mL/mbar", False),
363	_MedibusField("resistance", "mbar/L/s", False),
364	_MedibusField("r^2", "", False),
365	_MedibusField("spontaneous inspiratory time", "s", False),
366	_MedibusField("minimal pressure", "mbar", False),
367	_MedibusField("P0.1", "mbar", False),
368	_MedibusField("mean pressure", "mbar", False),
369	_MedibusField("plateau pressure", "mbar", False),
370	_MedibusField("PEEP", "mbar", False),
371	_MedibusField("intrinsic PEEP", "mbar", False),
372	_MedibusField("mandatory respiratory rate", "/min", False),
373	_MedibusField("mandatory minute volume", "L/min", False),
374	_MedibusField("peak inspiratory pressure", "mbar", False),
375	_MedibusField("mandatory tidal volume", "L", False),
376	_MedibusField("spontaneous tidal volume", "L", False),
377	_MedibusField("trapped volume", "mL", False),
378	_MedibusField("mandatory expiratory tidal volume", "mL", False),
379	_MedibusField("spontaneous expiratory tidal volume", "mL", False),
380	_MedibusField("mandatory inspiratory tidal volume", "mL", False),
381	_MedibusField("tidal volume", "mL", False),
382	_MedibusField("spontaneous inspiratory tidal volume", "mL", False),
383	_MedibusField("negative inspiratory force", "mbar", False),
384	_MedibusField("leak minute volume", "L/min", False),
385	_MedibusField("leak percentage", "%", False),
386	_MedibusField("spontaneous respiratory rate", "/min", False),
387	_MedibusField("percentage of spontaneous minute volume", "%", False),
388	_MedibusField("spontaneous minute volume", "L/min", False),
389	_MedibusField("minute volume", "L/min", False),
390	_MedibusField("airway temperature", "degrees C", False),
391	_MedibusField("rapid shallow breating index", "1/min/L", False),
392	_MedibusField("respiratory rate", "/min", False),
393	_MedibusField("inspiratory:expiratory ratio", "", False),
394	_MedibusField("CO2 flow", "mL/min", False),
395	_MedibusField("dead space volume", "mL", False),
396	_MedibusField("percentage dead space of expiratory tidal volume", "%", False),
397	_MedibusField("end-tidal CO2", "%", False),
398	_MedibusField("end-tidal CO2", "kPa", False),
399	_MedibusField("end-tidal CO2", "mmHg", False),
400	_MedibusField("fraction inspired O2", "%", False),
401	_MedibusField("spontaneous inspiratory:expiratory ratio", "", False),
402	_MedibusField("elastance", "mbar/L", False),
403	_MedibusField("time constant", "s", False),
404	_MedibusField(
405	"ratio between upper 20% pressure range and total dynamic compliance",
406	"",
407	False,
408	),
409	_MedibusField("end-inspiratory pressure", "mbar", False),
410	_MedibusField("expiratory tidal volume", "mL", False),
411	_MedibusField("high pressure", "mbar", False),
412	_MedibusField("low pressure", "mbar", False),
413	_MedibusField("time at low pressure", "s", False),
414	_MedibusField("airway pressure (pod)", "mbar", True),
415	_MedibusField("esophageal pressure (pod)", "mbar", True),
416	_MedibusField("transpulmonary pressure (pod)", "mbar", True),
417	_MedibusField("gastric pressure/auxiliary pressure (pod)", "mbar", True),
418	],
419	},
420	}

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