io_helper: add helper routines pull_/push_compressed

src/finam/__init__.py

@@ -80,6 +80,8 @@ Utilities for data and metadata handling.
 
     Info
     UNITS
+    pull_compressed
+    push_compressed
 
 Interfaces
 ==========
@@ -180,6 +182,7 @@ from .sdk import (
     TimeComponent,
     TimeDelayAdapter,
 )
+from .tools import pull_compressed, push_compressed
 
 try:
     from ._version import __version__
@@ -226,6 +229,7 @@ __all__ += [
 ]
 __all__ += ["CellType", "Location"]
 __all__ += ["UNITS", "Info"]
+__all__ += ["pull_compressed", "push_compressed"]
 __all__ += [
     "FinamCircularCouplingError",
     "FinamConnectError",

src/finam/tools/__init__.py

@@ -57,12 +57,22 @@ Connect helper
     FromInput
     FromOutput
     FromValue
+
+IO helper
+=========
+
+.. autosummary::
+   :toctree: generated
+
+    pull_compressed
+    push_compressed
 """
 from .connect_helper import ConnectHelper, FromInput, FromOutput, FromValue
 from .cwd_helper import execute_in_cwd, set_directory
 from .date_helper import is_timedelta
 from .enum_helper import get_enum_value
 from .inspect_helper import inspect
+from .io_helper import pull_compressed, push_compressed
 from .log_helper import (
     ErrorLogger,
     LogCStdOutStdErr,
@@ -85,3 +95,4 @@ __all__ += [
     "LogCStdOutStdErr",
 ]
 __all__ += ["ConnectHelper", "FromInput", "FromOutput", "FromValue"]
+__all__ += ["pull_compressed", "push_compressed"]

src/finam/tools/io_helper.py

+"""Input and Output helpers."""
+import numpy as np
+
+from ..interfaces import IInput, IOutput
+
+
+def pull_compressed(input, time):
+    """
+    Pull compressed data from an Input object.
+
+    Parameters
+    ----------
+    input : IInput
+        The Input object to pull data from.
+    time : :class:`datetime <datetime.datetime>`
+        Simulation time to get the data for.
+
+    Returns
+    -------
+    :class:`pint.Quantity`
+        Flattened and unmasked data values for the given simulation time.
+
+    Raises
+    ------
+    ValueError
+        If input is not an IInput instance.
+    """
+    if not isinstance(input, IInput):
+        msg = "pull_compressed: Given input is not an Input object."
+        raise ValueError(msg)
+    return input.info.grid.to_compressed(input.pull_data(time))
+
+
+def push_compressed(output, time, data, nodata=np.nan):
+    """
+    Push compressed data to an Output object.
+
+    Parameters
+    ----------
+    output : IOutput
+        The Output object to push data to.
+    time : :class:`datetime <datetime.datetime>`
+        Simulation time of the data set.
+    data : array_like
+        Flattened and unmasked data values to push.
+    nodata : numeric, optional
+        Fill value for masked values. Should have a compatible type.
+        By default np.nan
+
+    Raises
+    ------
+    ValueError
+        If output is not an IOutput instance.
+    """
+    if not isinstance(output, IOutput):
+        msg = "push_compressed: Given output is not an Output object."
+        raise ValueError(msg)
+    output.push_data(output.info.grid.from_compressed(data, nodata), time)