Skip to content
Merged
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
43 changes: 42 additions & 1 deletion src/RESPFlow/access_files.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import os, re
import pandas as pd

#
# =============================================================================
Expand Down Expand Up @@ -130,4 +131,44 @@ def map_files(in_path: str,
elif (directory.endswith(file_ext)) and ((expression is None) or (re.match(expression, fileName)!=None)):
file_dirs[fileName] = new_path

return file_dirs
return file_dirs

#
# =============================================================================
#

from pathlib import Path
from importlib import resources as ir
import shutil

def make_sample_data(path_names: dict[str, str]) -> None:
"""
Copies sample data files from the RESPFlow package to the raw data folder
specified in `path_names`.

Parameters
----------
path_names : dict[str, str]
A dictionary of file locations with keys for stage in the processing
pipeline. Must contain a 'raw' key specifying the destination folder
for the sample data.

Raises
------
ValueError
If 'raw' key is not found in `path_names`.

Returns
-------
None
The function performs file operations only; it does not return a value.
"""
if "raw" not in path_names:
raise ValueError("Raw path not detected in path_names.")

dest = Path(path_names["raw"])
dest.mkdir(parents=True, exist_ok=True)

with ir.as_file(ir.files("RESPFlow").joinpath("data")) as src:
shutil.copytree(src, dest, dirs_exist_ok=True)

127 changes: 123 additions & 4 deletions tests/test_access_files.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,21 @@
# ============================================================================
# Imports
# ============================================================================

# --- Standard library ---
import os
from pathlib import Path
from importlib import resources as ir

# --- Third-party packages ---
import pytest
import RESPFlow

# --- Local application / package imports ---
from RESPFlow.access_files import make_paths, map_files, make_sample_data

# ============================================================================
# Tests
# ===========================================================================

# ============================================================================
# Test make_paths function
Expand All @@ -23,7 +38,7 @@ def fake_makedirs(path, exist_ok=True):


def test_make_paths_defaults(mock_filesystem):
paths = RESPFlow.make_paths()
paths = make_paths()

expected_keys = {
'raw', 'notch', 'bandpass', 'fwr',
Expand All @@ -45,7 +60,7 @@ def test_make_paths_custom_root_raw(mock_filesystem):
custom_root = "my_root"
custom_raw = "my_raw"

paths = RESPFlow.make_paths(root=custom_root, raw=custom_raw)
paths = make_paths(root=custom_root, raw=custom_raw)

# Assert custom raw should be absolute and point to my_raw
assert paths['raw'] == "/abs/my_raw"
Expand All @@ -61,10 +76,114 @@ def test_make_paths_custom_root_raw(mock_filesystem):
# Test map_files function
# ============================================================================

def _touch(p: str, text: str = "") -> None:
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "w", encoding="utf-8") as f:
f.write(text)


def test_map_files_basic_recursive(tmp_path):
# data/
# 10/file1.csv
# 10/nested/info.txt
# 23/file2.csv
root = tmp_path / "data"
f1 = root / "10" / "file1.csv"
f_txt = root / "10" / "nested" / "info.txt"
f2 = root / "23" / "file2.csv"
_touch(str(f1), "a,b\n1,2\n")
_touch(str(f_txt), "hello\n")
_touch(str(f2), "x,y\n3,4\n")

result = map_files(str(root)) # default file_ext='csv'

# Only CSVs should appear
keys = set(result.keys())
assert os.path.join("10", "file1.csv") in keys
assert os.path.join("23", "file2.csv") in keys

# Non-csv shouldn't be included
assert not any("info.txt" in k for k in keys)

# Values should be absolute paths pointing to the files
assert os.path.isabs(result[os.path.join("10", "file1.csv")])
assert os.path.exists(result[os.path.join("23", "file2.csv")])


def test_map_files_extension_filter(tmp_path):
root = tmp_path / "data"
_touch(str(root / "a.csv"), "csv\n")
_touch(str(root / "b.txt"), "txt\n")

only_txt = map_files(str(root), file_ext="txt")

assert set(only_txt.keys()) == {"b.txt"}

only_csv = map_files(str(root), file_ext="csv")

assert set(only_csv.keys()) == {"a.csv"}


def test_map_files_regex_filter_on_filename(tmp_path):
root = tmp_path / "data"
_touch(str(root / "10" / "keep_this.csv"), "ok\n")
_touch(str(root / "10" / "skip_this.csv"), "no\n")
_touch(str(root / "23" / "also_skip.csv"), "no\n")

# Match only files whose name ends with 'keep_this.csv'
expr = r".*keep_this\.csv$"

result = map_files(str(root), expression=expr)

keys = set(result.keys())
assert any(k.endswith(os.path.join("10", "keep_this.csv")) for k in keys)
assert not any(k.endswith("skip_this.csv") for k in keys)

def test_map_files_invalid_regex_raises(tmp_path):
root = tmp_path / "data"
_touch(str(root / "a.csv"), "a\n")

with pytest.raises(Exception, match="Invalid regex expression"):
map_files(str(root), expression="(") # invalid regex

# ============================================================================
# Tests for map_files
# Tests for make_sample_data function
# ============================================================================

def _fake_pkg_with_data(tmp_path: Path) -> Path:
"""Create a fake RESPFlow/data tree with a few files."""
root = tmp_path / "RESPFlow_fake"
(root / "data" / "10").mkdir(parents=True, exist_ok=True)
(root / "data" / "10" / "file1.csv").write_text("a,b\n1,2\n", encoding="utf-8")
(root / "data" / "10" / "nested").mkdir(parents=True, exist_ok=True)
(root / "data" / "10" / "nested" / "info.txt").write_text("hello\n", encoding="utf-8")
(root / "data" / "23").mkdir(parents=True, exist_ok=True)
(root / "data" / "23" / "file2.csv").write_text("x,y\n3,4\n", encoding="utf-8")
return root


def test_copies_all_and_preserves_structure(tmp_path: Path, monkeypatch):
fake_pkg = _fake_pkg_with_data(tmp_path)
# Make ir.files("RESPFlow") point to our fake package root
monkeypatch.setattr(ir, "files", lambda _pkg: fake_pkg)

dest = tmp_path / "raw"
make_sample_data({"raw": str(dest)})

assert (dest / "10" / "file1.csv").is_file()
assert (dest / "10" / "nested" / "info.txt").is_file()
assert (dest / "23" / "file2.csv").is_file()
# spot-check contents
assert (dest / "10" / "file1.csv").read_text(encoding="utf-8").startswith("a,b\n")


def test_missing_raw_raises(monkeypatch, tmp_path: Path):
fake_pkg = _fake_pkg_with_data(tmp_path)
monkeypatch.setattr(ir, "files", lambda _pkg: fake_pkg)

with pytest.raises(ValueError, match="Raw path not detected"):
make_sample_data({})

# ============================================================================
# End of tests
# ============================================================================