Enable parsing bytes

.github/workflows/pipeline.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.12", "3.13", "3.14", "3.15"]
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python ${{ matrix.python-version }}

detl/__init__.py

@@ -1,7 +1,17 @@
 import importlib.metadata
+import pathlib
+import sys
+import warnings
+from typing import overload
 
 from . import parsing
-from .core import DASwareParser, DASwareVersion, DWData
+from .core import DASwareParser, DASwareVersion, DWData, Source, read_source
+
+if sys.version_info >= (3, 13):
+    from warnings import deprecated
+else:
+    from typing_extensions import deprecated
+
 
 __version__ = importlib.metadata.version(__package__ or __name__)
 
@@ -11,26 +21,23 @@
 }
 
 
-def get_parser(filepath) -> DASwareParser:
+def get_parser(source: Source) -> DASwareParser:
     """Analyzes a raw DASware CSV file and selects an appropiate parser.
 
     Args:
-        filepath (str or pathlib.Path): path pointing to the file of interest
+        source (pathlib.Path, str, bytes or binary stream): Path to the file of interest, its content as bytes, or a binary stream
 
     Returns:
         DWDParser: a parser that can be used for the provided file type
 
     Raises:
         NotImlementedError: when the file contents do not match with a known DASware CSV style
     """
+    content = read_source(source)
     try:
-        with open(filepath, "r", encoding="utf-8") as f:
-            lines = f.readlines()
+        lines = content.decode("utf-8").splitlines()
     except UnicodeDecodeError:
-        with open(filepath, "r", encoding="latin-1") as f:
-            lines = f.readlines()
-
-    version = None
+        lines = content.decode("latin-1").splitlines()
 
     if len(lines) > 2 and lines[2].startswith('"FngArchiv";"4.0.1"'):
         version = DASwareVersion.V4
@@ -44,23 +51,56 @@ def get_parser(filepath) -> DASwareParser:
     return parser_cls()
 
 
-def parse(filepath, *, inoculation_times: dict = None) -> DWData:
+@overload
+def parse(source: Source, *, inoculation_times: dict | None = None) -> DWData: ...
+
+
+@overload
+@deprecated("'filepath' is deprecated. Use 'source' instead.")
+def parse(*, filepath, inoculation_times: dict | None = None) -> DWData: ...
+
+
+def parse(
+    source: Source | None = None,
+    filepath: str | pathlib.Path | None = None,
+    *,
+    inoculation_times: dict | None = None,
+) -> DWData:
     """Parses a raw DASware CSV file into a DWData object.
 
-    Args:
-        filepath (str or pathlib.Path): path pointing to the file of interest
-        inoculation_times (dict or None): optional overrides for inoculation timepoints
-            key (int): reactor number
-            value (datetime.datetime): timezone-aware datetime object of the real inoculation time (computer clock!)
+    Parameters:
+    -----------
+        source: pathlib.Path | str | bytes
+            Path pointing to the file of interest, its content as bytes, or a binary stream. A ``str`` is always interpreted as a path. Streams must be opened in binary mode;\n
+            they are read from their current position and are not closed.
+        inoculation_times: dict | None
+            optional overrides for inoculation timepoints
+                key (int): reactor number
+                value (datetime.datetime): timezone-aware datetime object of the real inoculation time (computer clock!)
 
     Returns:
+    --------
         DWData: parsed data object
 
     Raises:
         NotImlementedError: when the file contents do not match with a known DASware CSV style
     """
-    parser = get_parser(filepath)
-    data = parser.parse(filepath)
+    if filepath is not None:
+        if source is not None:
+            raise TypeError("parse() got both 'source' and 'filepath'")
+        # @deprecated on an overload only informs type checkers, so warn at runtime too
+        warnings.warn(
+            "'filepath' is deprecated, use 'source' instead.", DeprecationWarning, stacklevel=2
+        )
+        source = filepath
+    elif source is None:
+        raise TypeError("parse() missing required argument: 'source'")
+
+    # Read once, so streams are consumed only once by detection and parsing
+    content = read_source(source)
+
+    parser = get_parser(content)
+    data = parser.parse(content)
 
     if inoculation_times:
         for r, dt_inoculate in inoculation_times.items():

detl/core.py

@@ -3,7 +3,7 @@
 import abc
 import enum
 import pathlib
-from typing import Dict
+from typing import Dict, Protocol
 
 import numpy
 import pandas
@@ -233,18 +233,51 @@ def get_narrow_data(self, kdim: str = "process_time"):
         return narrow_data
 
 
+class SupportsRead[T](Protocol):
+    """Object with a ``read()`` method returning ``T``, such as a file opened in binary mode."""
+
+    def read(self) -> T: ...
+
+
+type Source = pathlib.Path | str | bytes | SupportsRead[bytes]
+"""Raw DASWare data: a file path, the file content, or a binary stream.
+
+A ``str`` is always interpreted as a path. Streams must be opened in binary mode;
+they are read from their current position and are not closed.
+"""
+
+
+def read_source(source: Source) -> bytes:
+    """Return the raw content of ``source``, reading it from disk or the stream if needed."""
+    if isinstance(source, bytes):
+        return source
+    if isinstance(source, (str, pathlib.Path)):
+        return pathlib.Path(source).read_bytes()
+    content = source.read()
+    if not isinstance(content, bytes):
+        raise TypeError(
+            f"Expected a binary stream, got {type(content).__name__} content. "
+            "Open the file in binary mode ('rb')."
+        )
+    return content
+
+
 class DASwareParser(object):
     """Abstract type for parsers that read DASware CSV files."""
 
     __metaclass__ = abc.ABCMeta
 
-    @abc.abstractmethod
-    def parse(self, filepath: pathlib.Path) -> DWData:
+    def parse(self, source: Source) -> DWData:
         """Parses the provided DASware CSV file into a data object.
 
         Args:
-            filepath (str or pathlib.Path): path pointing to the file of interest
+            source (pathlib.Path | str | bytes | SupportsRead[bytes]): Path to the file, its content as bytes, or a binary stream
         """
+        return self._parse(read_source(source))
+
+    @abc.abstractmethod
+    def _parse(self, raw: bytes) -> DWData:
+        """Parse the raw content of a DASWare CSV file"""
         raise NotImplementedError(
             "Whoever implemented {} screwed up.".format(self.__class__.__name__)
         )

detl/parsing/common.py

@@ -1,10 +1,9 @@
 import collections
 import datetime
+import io
 import logging
-import pathlib
 import re
 import warnings
-from io import StringIO
 
 import numpy
 import pandas
@@ -15,32 +14,36 @@
 logger = logging.getLogger("detl.parsing.common")
 
 
-def split_blocks(filepath: pathlib.Path) -> dict:
-    """Reads a CSV file and splits its contents into scoped blocks.
+def split_blocks(raw: bytes) -> dict[int | None, dict[str, str]]:
+    """Splits the raw content of a DASware CSV file into scoped blocks.
 
     Args:
-        filepath (pathlib.Path): path to the raw CSV
+        raw (bytes): content of the raw CSV
 
     Returns:
         scoped_blocks (dict): dicationary mapping scope to dictionary of blocks
     """
-    if not isinstance(filepath, (str, pathlib.Path)):
-        raise ValueError("Please provide filepath either as str or pathlib.Path object")
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError:
+        text = raw.decode("latin-1")
 
     # split the entire file into table-blocks
-    blocks = [[]]
-    with pathlib.Path(filepath).open(mode="r", errors="replace") as file:
-        for line in file:
-            if len(line) == 1:
-                blocks.append([])
-            else:
-                blocks[-1].append(line)
+    blocks: list[list[str]] = [[]]
+    # universal newlines like open(); str.splitlines() would also split on Latin-1 "\x85"
+    for line in io.StringIO(text, newline=None):
+        if len(line) == 1:
+            blocks.append([])
+        else:
+            blocks[-1].append(line)
     # drop empty blocks
     blocks = [block for block in blocks if len(block) > 1]
 
     # group blocks by scope (None or reactor-number)
-    scoped_blocks = collections.defaultdict(dict)
-    scope = None
+    scoped_blocks: collections.defaultdict[int | None, dict[str, str]] = collections.defaultdict(
+        dict
+    )
+    scope: int | None = None
     for blocklines in blocks:
         blockheader = blocklines[0].strip()
         setup_matches = re.findall(r'"\[Setup(\d+)\]"', blockheader)
@@ -87,7 +90,7 @@ def transform_to_dwdata(
 
 
 def parse_generic(header, block, scope):
-    df = pandas.read_csv(StringIO(block), sep=";")
+    df = pandas.read_csv(io.StringIO(block), sep=";")
     attr = "_" + header.lower().replace(" ", "_").replace("-", "_")
     return (attr, df)
 

detl/parsing/dw4.py

@@ -104,13 +104,9 @@
 
 
 class DASware4Parser(core.DASwareParser):
-    def parse(self, filepath) -> core.DWData:
-        """Parses the provided DASware CSV file into a data object.
-
-        Args:
-            filepath (str or pathlib.Path): path pointing to the file of interest
-        """
-        scoped_blocks = common.split_blocks(filepath)
+    def _parse(self, raw: bytes) -> core.DWData:
+        """Parses the raw content of a DASware CSV file into a data object."""
+        scoped_blocks = common.split_blocks(raw)
         dd = common.transform_to_dwdata(scoped_blocks, BLOCKPARSERS, version=core.DASwareVersion.V5)
 
         for _, reactor in dd.items():

detl/parsing/dw5.py

@@ -1,5 +1,4 @@
 import logging
-import pathlib
 
 import numpy
 import pandas
@@ -124,13 +123,9 @@
 
 
 class DASware5Parser(core.DASwareParser):
-    def parse(self, filepath: pathlib.Path) -> core.DWData:
-        """Parses the provided DASware CSV file into a data object.
-
-        Args:
-            filepath (str or pathlib.Path): path pointing to the file of interest
-        """
-        scoped_blocks = common.split_blocks(filepath)
+    def _parse(self, raw: bytes) -> core.DWData:
+        """Parses the raw content of a DASware CSV file into a data object."""
+        scoped_blocks = common.split_blocks(raw)
         scoped_blocks = {
             key: value for (key, value) in scoped_blocks.items() if "TrackData" in list(value)
         }

pyproject.toml

@@ -4,10 +4,10 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "detl"
-version = "1.2.0"
+version = "1.3.0"
 description = "Package for parsing and transforming DASware raw data exports."
 readme = "README.md"
-requires-python = ">=3.11"
+requires-python = ">=3.12"
 license = "AGPL-3.0-or-later"
 authors = [
     {name = "Michael Osthege", email = "m.osthege@fz-juelich.de"},
@@ -17,17 +17,18 @@ authors = [
 classifiers = [
     "Programming Language :: Python",
     "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
     "Programming Language :: Python :: 3.14",
+    "Programming Language :: Python :: 3.15",
     "Intended Audience :: Science/Research",
     "Topic :: Scientific/Engineering",
 ]
 dependencies = [
     "numpy",
     "pandas",
     "pytz",
+    "typing_extensions>=4.5; python_version < '3.13'",
 ]
 
 [project.urls]