Move some general functions to a util module and clean up some imports (#396)

Author: cx1111

wfdb/io/_header.py

@@ -1,12 +1,12 @@
 import datetime
-import os
 import re
 from typing import List, Tuple
 
 import numpy as np
 import pandas as pd
 
 from wfdb.io import _signal
+from wfdb.io import util
 
 
 """
@@ -603,7 +603,7 @@ def wr_header_file(self, rec_write_fields, sig_write_fields, write_dir):
             comment_lines = ["# " + comment for comment in self.comments]
             header_lines += comment_lines
 
-        lines_to_file(self.record_name + ".hea", write_dir, header_lines)
+        util.lines_to_file(self.record_name + ".hea", write_dir, header_lines)
 
 
 class MultiHeaderMixin(BaseHeaderMixin):
@@ -795,7 +795,7 @@ def wr_header_file(self, write_fields, write_dir):
             comment_lines = ["# " + comment for comment in self.comments]
             header_lines += comment_lines
 
-        lines_to_file(self.record_name + ".hea", header_lines, write_dir)
+        util.lines_to_file(self.record_name + ".hea", header_lines, write_dir)
 
     def get_sig_segments(self, sig_name=None):
         """
@@ -1122,25 +1122,3 @@ def _read_segment_lines(segment_lines):
 
 class HeaderSyntaxError(ValueError):
     """Invalid syntax found in a WFDB header file."""
-
-
-def lines_to_file(file_name, write_dir, lines):
-    """
-    Write each line in a list of strings to a text file.
-
-    Parameters
-    ----------
-    write_dir : str
-        The output directory in which the header is written.
-    lines : list
-        The lines to be written to the text file.
-
-    Returns
-    -------
-    N/A
-
-    """
-    f = open(os.path.join(write_dir, file_name), "w")
-    for l in lines:
-        f.write("%s\n" % l)
-    f.close()

wfdb/io/_signal.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 
-from wfdb.io import download, _coreio
+from wfdb.io import download, _coreio, util
 
 
 MAX_I32 = 2147483647
@@ -1563,14 +1563,14 @@ def _required_byte_num(mode, fmt, n_samp):
 
         if n_extra == 2:
             if fmt == "310":
-                n_bytes = upround(n_samp * 4 / 3, 4)
+                n_bytes = util.upround(n_samp * 4 / 3, 4)
             # 311
             else:
                 if mode == "read":
                     n_bytes = math.ceil(n_samp * 4 / 3)
                 # Have to write more bytes for WFDB c to work
                 else:
-                    n_bytes = upround(n_samp * 4 / 3, 4)
+                    n_bytes = util.upround(n_samp * 4 / 3, 4)
         # 0 or 1
         else:
             n_bytes = math.ceil(n_samp * 4 / 3)
@@ -2506,42 +2506,3 @@ def _infer_sig_len(
 
     return sig_len
 
-
-def downround(x, base):
-    """
-    Round <x> down to nearest <base>.
-
-    Parameters
-    ---------
-    x : str, int, float
-        The number that will be rounded down.
-    base : int, float
-        The base to be rounded down to.
-
-    Returns
-    -------
-    float
-        The rounded down result of <x> down to nearest <base>.
-
-    """
-    return base * math.floor(float(x) / base)
-
-
-def upround(x, base):
-    """
-    Round <x> up to nearest <base>.
-
-    Parameters
-    ---------
-    x : str, int, float
-        The number that will be rounded up.
-    base : int, float
-        The base to be rounded up to.
-
-    Returns
-    -------
-    float
-        The rounded up result of <x> up to nearest <base>.
-
-    """
-    return base * math.ceil(float(x) / base)

wfdb/io/record.py

@@ -11,6 +11,7 @@
 from wfdb.io import _signal
 from wfdb.io import _url
 from wfdb.io import download
+from wfdb.io import util
 
 
 # -------------- WFDB Signal Calibration and Classification ---------- #
@@ -468,7 +469,7 @@ def check_field(self, field, required_channels="all"):
                             "File names should only contain alphanumerics, hyphens, and an extension. eg. record-100.dat"
                         )
                     # Check that dat files are grouped together
-                    if not is_monotonic(self.file_name):
+                    if not util.is_monotonic(self.file_name):
                         raise ValueError(
                             "Signals in a record that share a given file must be consecutive."
                         )
@@ -2920,37 +2921,6 @@ def wrsamp(
     record.wrsamp(write_dir=write_dir)
 
 
-def is_monotonic(full_list):
-    """
-    Determine whether elements in a list are monotonic. ie. unique
-    elements are clustered together.
-
-    ie. [5,5,3,4] is, [5,3,5] is not.
-
-    Parameters
-    ----------
-    full_list : list
-        The input elements used for the analysis.
-
-    Returns
-    -------
-    bool
-        Whether the elements are monotonic (True) or not (False).
-
-    """
-    prev_elements = set({full_list[0]})
-    prev_item = full_list[0]
-
-    for item in full_list:
-        if item != prev_item:
-            if item in prev_elements:
-                return False
-            prev_item = item
-            prev_elements.add(item)
-
-    return True
-
-
 def dl_database(
     db_dir,
     dl_dir,

wfdb/io/util.py

@@ -0,0 +1,101 @@
+"""
+A module for general utility functions
+"""
+import math
+import os
+
+from typing import Sequence
+
+
+def lines_to_file(file_name: str, write_dir: str, lines: Sequence[str]):
+    """
+    Write each line in a list of strings to a text file.
+
+    Parameters
+    ----------
+    file_name: str
+        The base name of the file
+    write_dir : str
+        The output directory in which the file is to be written.
+    lines : list
+        The lines to be written to the text file.
+
+    Returns
+    -------
+    N/A
+
+    """
+    with open(os.path.join(write_dir, file_name), "w", encoding="utf-8") as f:
+        for l in lines:
+            f.write(f"{l}\n")
+
+
+def is_monotonic(items: Sequence) -> bool:
+    """
+    Determine whether elements in a list are monotonic. ie. unique
+    elements are clustered together.
+
+    ie. [5,5,3,4] is, [5,3,5] is not.
+
+    Parameters
+    ----------
+    items : Sequence
+        The input elements to be checked.
+
+    Returns
+    -------
+    bool
+        Whether the elements are monotonic (True) or not (False).
+
+    """
+    prev_elements = set({items[0]})
+    prev_item = items[0]
+
+    for item in items:
+        if item != prev_item:
+            if item in prev_elements:
+                return False
+            prev_item = item
+            prev_elements.add(item)
+
+    return True
+
+
+def downround(x, base):
+    """
+    Round <x> down to nearest <base>.
+
+    Parameters
+    ---------
+    x : str, int, float
+        The number that will be rounded down.
+    base : int, float
+        The base to be rounded down to.
+
+    Returns
+    -------
+    float
+        The rounded down result of <x> down to nearest <base>.
+
+    """
+    return base * math.floor(float(x) / base)
+
+
+def upround(x, base):
+    """
+    Round <x> up to nearest <base>.
+
+    Parameters
+    ---------
+    x : str, int, float
+        The number that will be rounded up.
+    base : int, float
+        The base to be rounded up to.
+
+    Returns
+    -------
+    float
+        The rounded up result of <x> up to nearest <base>.
+
+    """
+    return base * math.ceil(float(x) / base)

wfdb/plot/plot.py

@@ -1,10 +1,9 @@
-import numpy as np
 import os
-import pdb
+
+import numpy as np
 
 from wfdb.io.record import Record, rdrecord
-from wfdb.io._header import float_types
-from wfdb.io._signal import downround, upround
+from wfdb.io.util import downround, upround
 from wfdb.io.annotation import Annotation
 
 

wfdb/processing/basic.py

@@ -1,7 +1,6 @@
 import numpy as np
-from scipy import signal
 import pandas as pd
-import pdb
+from scipy import signal
 
 from wfdb.io.annotation import Annotation