Repository: IQuOD/wodpy (language: python)

Issue:
A Generator which includes a mapping method
@BillMills, still a WIP that might help Chris. The idea here is to provide a Generator to iterate over each object in a WOD dataset. The `map()` runs on top of that, processing each object. This is a serial mode example. I'll add a concurrent version here.


Discussion / hints:

Fix patch:
diff --git a/pyproject.toml b/pyproject.toml
index 76c29e3..0553b0f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -14,12 +14,15 @@ classifiers = [
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
 ]
+requires-python = ">=3.9"
 dependencies = [
     "numpy",
     "pandas",
     "netCDF4"
 ]
-requires-python = ">=3.9"
+
+[project.optional-dependencies]
+parallel = ["loky >= 3.0"]
 
 [project.urls]
 Homepage = "https://github.com/IQuOD/wodpy"
diff --git a/wodpy/extra.py b/wodpy/extra.py
index 90927cd..cb9402b 100644
--- a/wodpy/extra.py
+++ b/wodpy/extra.py
@@ -1,6 +1,15 @@
-
+import logging, os
 from datetime import datetime
 
+module_logger = logging.getLogger("wodpy.extra")
+try:
+    from loky import get_reusable_executor
+    LOKY_AVAILABLE = True
+    module_logger.debug("Will use package loky to process in parallel.")
+except:
+    LOKY_AVAILABLE = False
+    module_logger.info("Missing package loky. Falling back to threading.")
+
 from .wod import WodProfile
 from .wodnc import Ragged, ncProfile
 
@@ -101,3 +110,42 @@ def keys(self):
 
     def __getitem__(self, item):
         return self.data[item]
+
+
+class WODFile():
+    """A WOD file object
+
+    For now, let's assume it will be a plain ASCII type. Later we expand
+    for other possibilities
+    """
+    def __init__(self, filename: str):
+        self.fid = open(filename, mode="r")
+        self.file_size = os.fstat(self.fid.fileno()).st_size
+
+
+
+
+class ConcurrentMapping():
+    def pmap(self, func, args=None, npes: int=4, timeout:int=2):
+        """Parallel mapping"""
+        executor = get_reusable_executor(max_workers=npes, timeout=timeout)
+        results = executor.map(func, self)
+        yield from results
+
+
+class WODGenerator(ConcurrentMapping, WODFile):
+    def __init__(self, filename: str):
+        super().__init__(filename)
+
+    def __iter__(self):
+          return self
+
+    def __next__(self):
+        if self.fid.tell() >= self.file_size:
+            raise StopIteration
+        return WodProfile(self.fid)
+
+    def map(self, func, args=None):
+        """(Serial) mapping"""
+        for p in self:
+            yield func(p)


Test patch:
diff --git a/tests/test_generator.py b/tests/test_generator.py
new file mode 100644
index 0000000..e4ff4ea
--- /dev/null
+++ b/tests/test_generator.py
@@ -0,0 +1,31 @@
+from datetime import datetime
+import os
+
+from wodpy import wod
+from wodpy.extra import WODFile, WODGenerator
+
+def test_file():
+    WOD = WODFile("tests/testData/classic.dat")
+
+
+def test_iter():
+    WOD = WODGenerator("tests/testData/classic.dat")
+    profiles = [p for p in WOD]
+    assert len(profiles) == 2
+
+
+def test_map():
+    WOD = WODGenerator("tests/testData/classic.dat")
+    uids = [p for p in WOD.map(lambda x: x.uid())]
+    assert uids == [67064, 15556443]
+
+
+def test_pmap():
+    try:
+        import loky
+    except:
+        # No fallback alternative for loky at this point
+        return
+    WOD = WODGenerator("tests/testData/classic.dat")
+    uids = [p for p in WOD.pmap(lambda x: x.uid())]
+    assert uids == [67064, 15556443]