Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions eitprocessing/datahandling/eitdata.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import warnings
from dataclasses import InitVar, dataclass, field
from enum import Enum
from pathlib import Path
from pathlib import PurePath
from typing import TYPE_CHECKING, Any, TypeVar

import numpy as np
Expand Down Expand Up @@ -40,7 +40,7 @@ class is meant to hold data from (part of) a singular continuous measurement.
pixel_impedance: Impedance values for each pixel at each frame.
""" # TODO: fix docstring

path: str | Path | list[Path | str] = field(compare=False, repr=False)
path: str | PurePath | list[PurePath | str] = field(compare=False, repr=False)
nframes: int = field(repr=False)
time: np.ndarray = field(repr=False)
sample_frequency: float = field(metadata={"check_equivalence": True}, repr=False)
Expand Down Expand Up @@ -94,16 +94,16 @@ def framerate(self) -> float:

@staticmethod
def ensure_path_list(
path: str | Path | list[str | Path],
) -> list[Path]:
path: str | PurePath | list[str | PurePath],
) -> list[PurePath]:
"""Return the path or paths as a list.

The path of any EITData object can be a single str/Path or a list of str/Path objects. This method returns a
list of Path objects given either a str/Path or list of str/Paths.
The path of any EITData object can be a single str/PurePath or a list of str/PurePath objects. This method
returns a list of PurePath objects given either a str/PurePath or list of str/PurePaths.
"""
if isinstance(path, list):
return [Path(p) for p in path]
return [Path(path)]
return [PurePath(p) for p in path]
return [PurePath(path)]

def __add__(self: Self, other: Self) -> Self:
return self.concatenate(other)
Expand Down
6 changes: 3 additions & 3 deletions eitprocessing/datahandling/loading/__init__.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,13 @@
from functools import reduce
from pathlib import Path
from pathlib import Path, PurePath

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],
path: str | PurePath | list[str | PurePath],
vendor: Vendor | str,
label: str | None = None,
name: str | None = None,
Expand Down Expand Up @@ -78,7 +78,7 @@ def load_eit_data(
interval_datasets: list[DataCollection] = []

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

for single_path in paths:
loaded_data = load_from_single_path(
Expand Down
9 changes: 4 additions & 5 deletions eitprocessing/datahandling/loading/draeger.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import sys
import warnings
from functools import partial
from pathlib import Path, PurePath
from typing import TYPE_CHECKING, NamedTuple
from warnings import catch_warnings

Expand All @@ -21,8 +22,6 @@
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)
Expand All @@ -31,13 +30,13 @@


def load_from_single_path( # noqa: PLR0915
path: Path,
path: PurePath,
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
file_size = Path(path).stat().st_size

frame_size: int
medibus_fields: list
Expand Down Expand Up @@ -88,7 +87,7 @@ def load_from_single_path( # noqa: PLR0915
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:
with Path(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
Expand Down
7 changes: 4 additions & 3 deletions eitprocessing/datahandling/loading/sentec.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import warnings
from enum import IntEnum
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING

import numpy as np
Expand All @@ -18,7 +19,7 @@
from eitprocessing.datahandling.sparsedata import SparseData

if TYPE_CHECKING:
from pathlib import Path
from pathlib import PurePath

from numpy.typing import NDArray

Expand All @@ -28,15 +29,15 @@


def load_from_single_path(
path: Path,
path: PurePath,
sample_frequency: float | None = None,
first_frame: int = 0,
max_frames: int | None = None,
) -> dict[str, DataCollection]:
"""Load Sentec EIT data from path."""
time: list[float] = []
index = 0
with path.open("br") as fo, mmap.mmap(fo.fileno(), length=0, access=mmap.ACCESS_READ) as fh:
with Path(path).open("br") as fo, mmap.mmap(fo.fileno(), length=0, access=mmap.ACCESS_READ) as fh:
file_length = os.fstat(fo.fileno()).st_size
reader = BinReader(fh, endian="little")
version = reader.uint8()
Expand Down
7 changes: 4 additions & 3 deletions eitprocessing/datahandling/loading/timpel.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import warnings
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING

import numpy as np
Expand All @@ -15,7 +16,7 @@
from eitprocessing.datahandling.sparsedata import SparseData

if TYPE_CHECKING:
from pathlib import Path
from pathlib import PurePath

from numpy.typing import NDArray

Expand All @@ -30,7 +31,7 @@


def load_from_single_path(
path: Path,
path: PurePath,
sample_frequency: float | None = TIMPEL_SAMPLE_FREQUENCY,
first_frame: int = 0,
max_frames: int | None = None,
Expand All @@ -41,7 +42,7 @@ def load_from_single_path(

try:
data: NDArray = np.loadtxt(
str(path),
str(Path(path)),
dtype=float,
delimiter=",",
skiprows=first_frame,
Expand Down
8 changes: 6 additions & 2 deletions tests/eitdata/test_loading_draeger.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import sys
import tempfile
from pathlib import Path
from pathlib import Path, PurePath

import numpy as np
import pytest
Expand Down Expand Up @@ -80,6 +80,10 @@ def test_load_draeger(
"Loading without sample frequency should yield the same data"
)

path = sequence.eit_data["raw"].path
assert isinstance(path, PurePath), "EITData path should be a PurePath object"
assert not isinstance(path, Path), "EITData path should be a PurePath object"


def test_draeger_20hz_healthy_volunteer_2_differ(
draeger_20hz_healthy_volunteer: Sequence, draeger_20hz_healthy_volunteer_fixed_rr: Sequence
Expand Down Expand Up @@ -187,7 +191,7 @@ def test_event_on_first_frame(draeger_20hz_healthy_volunteer: Sequence):
with tempfile.NamedTemporaryFile(**kwargs) as temporary_file:
# Create a temporary file, that is removed after the context manager is closed
tempfile_path = Path(temporary_file.name)
with draeger_20hz_healthy_volunteer.eit_data["raw"].path.open("rb") as original_file:
with Path(draeger_20hz_healthy_volunteer.eit_data["raw"].path).open("rb") as original_file:
original_file.seek(ignore_bytes) # skip frames before the event
temporary_file.write(original_file.read()) # write remaining data to temp file

Expand Down
12 changes: 11 additions & 1 deletion tests/eitdata/test_loading_sentec.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from pathlib import Path
from pathlib import Path, PurePath

import numpy as np
import pytest
Expand Down Expand Up @@ -55,6 +55,9 @@ def test_load_sentec_single_file(
label="something else",
), "Loading with a different label should yield same data"

assert isinstance(sequence.eit_data["raw"].path, PurePath), "EITData path should be a PurePath object"
assert not isinstance(sequence.eit_data["raw"].path, Path), "EITData path should be a PurePath object"


@pytest.mark.parametrize(
("sequence_a_fixture_name", "sequence_b_fixture_name", "sequence_merge_fixture_name"),
Expand Down Expand Up @@ -95,6 +98,13 @@ def test_load_sentec_multiple_files(
"Second part of merged sequence should match second individual sequence"
)

assert all(isinstance(path_, PurePath) for path_ in sequence_merged.eit_data["raw"].path), (
"EITData path should be a PurePath object"
)
assert not any(isinstance(path_, Path) for path_ in sequence_merged.eit_data["raw"].path), (
"EITData path should not be a Path object"
)


def test_load_sentec_skip_frames(sentec_healthy_volunteer_1a: Sequence, sentec_healthy_volunteer_1a_path: Path):
n_frames = len(sentec_healthy_volunteer_1a)
Expand Down
5 changes: 5 additions & 0 deletions tests/eitdata/test_loading_timpel.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from pathlib import Path, PurePath

import pytest

from eitprocessing.datahandling.eitdata import EITData, Vendor
Expand Down Expand Up @@ -25,6 +27,9 @@ def test_loading_timpel(
loaded_using_enum_vendor = load_eit_data(sequence.eit_data["raw"].path, vendor=Vendor.TIMPEL, label="timpel")
assert sequence == loaded_using_enum_vendor

assert isinstance(sequence.eit_data["raw"].path, PurePath), "EITData path should be a PurePath object"
assert not isinstance(sequence.eit_data["raw"].path, Path), "EITData path should be a PurePath object"


def test_loading_timpel_multiple_files():
# TODO: find out whether it is possible to have a single measurements split into multiple files
Expand Down
Loading