Fix record name and add write_dir in csv2ann (#599)

Closes #595.

- The record name is now
`os.path.splitext(os.path.basename(file_name))[0]`, the same as in
`csv_to_wfdb`. Before, `file_name.split(".")[0]` turned `./x.csv` into
`""` and `data/v1.2/x.csv` into `data/v1`.
- New `write_dir` argument (default `""`, the current directory), passed
to `wrann`. It's added at the end of the signature so positional calls
keep working.
- Tests for time onsets, sample onsets without a header, and paths with
dots (`./x.csv`, `data/v1.2/x.csv`). The two tests that exercise the fix
fail on `main`.
- `csv2ann` is added to `docs/convert.rst`. The docs build passes with
`-W`.

**Behaviour change:** the annotation file used to be written next to the
CSV file (for paths without other dots). Now it's written to
`write_dir`, like `csv_to_wfdb`. That follows the note in #595 that both
functions should write to the same place. If you'd prefer to keep the
old location as the default, I can make `write_dir=None` mean "next to
the CSV file" instead.

`ruff check` is clean and `pytest -m "not network"` passes (85 tests).

Author: tompollard

docs/changes.rst

@@ -23,6 +23,9 @@ Unreleased
 **Speed up XQRS backsearch**
   ``XQRS`` no longer rechecks peaks that an earlier backsearch already rejected. Long stretches without QRS complexes (for example a disconnected lead) made detection take quadratic time, up to hours on long records. The detected QRS locations are unchanged.
 
+**Fix record name and add write directory in csv2ann**
+  ``csv2ann`` now takes the record name from the CSV file name without its directory and extension, as ``csv_to_wfdb`` does. Paths with a directory (such as ``data/x.csv``) used to fail when writing, and ``./x.csv`` gave an empty name. The annotation file is written to the new ``write_dir`` argument, which defaults to the current directory. ``csv2ann`` is now included in the API documentation.
+
 **Use Ruff for formatting and linting**
   The ``dev`` extra now includes ``ruff`` instead of ``black`` and ``pylint``, and the code is formatted with a line length of 88 (the default).
 

docs/convert.rst

@@ -8,7 +8,7 @@ CSV
 ---
 
 .. automodule:: wfdb.io.convert.csv
-   :members: csv_to_wfdb
+   :members: csv_to_wfdb, csv2ann
 
 EDF
 ---

src/wfdb/io/convert/csv.py

@@ -482,6 +482,7 @@ def csv2ann(
     header=True,
     delimiter=",",
     verbose=False,
+    write_dir="",
 ):
     """Read a CSV file and return an `Annotation` object or write a file.
 
@@ -490,11 +491,10 @@ def csv2ann(
     Parameters
     ----------
     file_name : str
-        The name of the CSV file to be read, including the '.csv' file
-        extension. If the argument contains any path delimiter characters, the
-        argument will be interpreted as PATH/BASE_RECORD. Both relative and
-        absolute paths are accepted. The BASE_RECORD file name will be used to
-        name the annotation file with the desired extension.
+        The path to the CSV file to be read, including the '.csv' file
+        extension. Both relative and absolute paths are accepted. The file
+        name without its directory and extension is used as the record name
+        of the annotation file.
     extension : str, optional
         The string annotation file extension.
     fs : float, optional
@@ -522,6 +522,9 @@ def csv2ann(
     verbose : bool, optional
         Whether to print all the information read about the file (True) or
         not (False).
+    write_dir : str, optional
+        The directory in which to write the annotation file. The default is
+        the current working directory. Ignored if `record_only` is True.
 
     Returns
     -------
@@ -627,8 +630,7 @@ def csv2ann(
         raise Exception("""The number of columns in the CSV was not
                         recognized.""")
 
-    # Remove extension from input file name
-    file_name = file_name.split(".")[0]
+    record_name = os.path.splitext(os.path.basename(file_name))[0]
     if time_onset:
         if not fs:
             raise Exception("""`fs` must be provided if `time_onset` is True
@@ -652,7 +654,7 @@ def csv2ann(
         if verbose:
             print("Finished creating Annotation object")
         return Annotation(
-            record_name=file_name,
+            record_name=record_name,
             extension=extension,
             sample=sample,
             symbol=symbol,
@@ -664,7 +666,7 @@ def csv2ann(
         )
     else:
         wrann(
-            file_name,
+            record_name,
             extension,
             sample=sample,
             symbol=symbol,
@@ -673,6 +675,7 @@ def csv2ann(
             num=num,
             aux_note=aux_note,
             fs=fs,
+            write_dir=write_dir,
         )
         if verbose:
             print("Finished writing Annotation file")

tests/io/test_convert.py

@@ -2,7 +2,8 @@
 
 import numpy as np
 
-from wfdb.io.convert.csv import csv_to_wfdb
+from wfdb.io.annotation import rdann
+from wfdb.io.convert.csv import csv2ann, csv_to_wfdb
 from wfdb.io.convert.edf import read_edf
 from wfdb.io.record import rdrecord
 
@@ -145,3 +146,52 @@ def test_write_dir(self, tmp_path):
         assert record_write.sig_name == record_100_dat.sig_name
         assert record_write.p_signal.size == 1300000
         assert record_write.p_signal.size == record_100_dat.p_signal.size
+
+
+class TestCsv2Ann:
+    """Tests for csv2ann."""
+
+    def test_time_onsets(self, tmp_path):
+        """Time onsets are converted to samples and written to write_dir."""
+        csv_file = tmp_path / "in" / "example.csv"
+        csv_file.parent.mkdir()
+        csv_file.write_text("onset,description\n0.2,p-wave\n0.8,qrs\n")
+        out_dir = tmp_path / "out"
+        out_dir.mkdir()
+
+        csv2ann(str(csv_file), fs=360, write_dir=str(out_dir))
+
+        ann = rdann(str(out_dir / "example"), "atr")
+        assert ann.sample.tolist() == [72, 288]
+        assert ann.symbol == ['"', '"']
+        assert ann.aux_note == ["p-wave", "qrs"]
+
+    def test_sample_onsets_no_header(self, tmp_path):
+        """Sample onsets are used as-is when time_onset is False."""
+        csv_file = tmp_path / "example.csv"
+        csv_file.write_text("5,p-wave\n13,qrs\n")
+
+        ann = csv2ann(
+            str(csv_file), fs=10, time_onset=False, header=False, record_only=True
+        )
+
+        assert ann.sample.tolist() == [5, 13]
+        assert ann.aux_note == ["p-wave", "qrs"]
+
+    def test_record_name_with_dots(self, tmp_path, monkeypatch):
+        """Dots in the directory or a ./ prefix don't truncate the name (#595)."""
+        data_dir = tmp_path / "data" / "v1.2"
+        data_dir.mkdir(parents=True)
+        (data_dir / "x.csv").write_text("onset,description\n0.5,qrs\n")
+        monkeypatch.chdir(data_dir)
+
+        ann = csv2ann("./x.csv", fs=100, record_only=True)
+        assert ann.record_name == "x"
+
+        ann = csv2ann(str(data_dir / "x.csv"), fs=100, record_only=True)
+        assert ann.record_name == "x"
+
+        out_dir = tmp_path / "out"
+        out_dir.mkdir()
+        csv2ann(str(data_dir / "x.csv"), fs=100, write_dir=str(out_dir))
+        assert (out_dir / "x.atr").exists()