Merge pull request #11 from au-imclab/dev

We have a working "Mean" segment

pyproject.toml

@@ -54,6 +54,8 @@ dependencies = [
   "pandas",
   "StrEnum; python_version < '3.11'",
 ]
+[tool.pytest.ini_options]
+pythonpath = "src"
 [tool.hatch.envs.default.scripts]
 test = "pytest {args:tests}"
 test-cov = "coverage run -m pytest {args:tests}"

src/mopipe/__about__.py

@@ -1,4 +1,4 @@
 # SPDX-FileCopyrightText: 2023-present zeyus <[email protected]>
 #
 # SPDX-License-Identifier: MIT
-__version__ = "0.0.5"
+__version__ = "0.1.0"

src/mopipe/analysis/pipeline.py → src/mopipe/core/analysis/pipeline.py

@@ -6,7 +6,7 @@
 
 import typing as t
 
-from mopipe.segments import Segment
+from mopipe.core.segments import Segment
 
 
 class Pipeline(t.MutableSequence[Segment]):
@@ -41,15 +41,15 @@ def add_segment(self, segment: Segment) -> int:
         self._segments.append(segment)
         return len(self._segments) - 1
 
-    def run(self, *args, **kwargs) -> t.Any:
+    def run(self, **kwargs) -> t.Any:
         """Run the pipeline."""
         output = None
         self._check_kwargs(**kwargs)
         for segment in self._segments:
             # most basic version here
             # we could also keep track of the output from each step
             # if that is useful, for now it's just I -> Segment -> O -> Segment -> O -> ...
-            kwargs["input"] = segment(*args, **kwargs)
+            kwargs["x"] = segment(**kwargs)
         return output
 
     def __repr__(self) -> str:

src/mopipe/common/util.py → src/mopipe/core/common/util.py

@@ -6,6 +6,8 @@
 import typing as t
 from uuid import uuid4
 
+import pandas as pd
+
 
 def maybe_generate_id(
     _id: t.Optional[str] = None, prefix: t.Optional[str] = None, suffix: t.Optional[str] = None
@@ -36,3 +38,24 @@ def maybe_generate_id(
     prefix = "" if prefix is None else prefix + "_"
     suffix = "" if suffix is None else "_" + suffix
     return prefix + str(uuid4()) + suffix
+
+
+def int_or_str_slice(s: slice) -> t.Union[type[int], type[str]]:
+    start = s.start
+    stop = s.stop
+    if isinstance(start, int) and isinstance(stop, int):
+        return int
+    if isinstance(start, str) and isinstance(stop, str):
+        return str
+    msg = "Invalid slice."
+    raise ValueError(msg)
+
+
+def df_slice(df: pd.DataFrame, s: slice) -> t.Union[pd.DataFrame, pd.Series]:
+    slice_type = int_or_str_slice(s)
+    if slice_type == int:
+        return df.iloc[s]
+    if slice_type == str:
+        return df.loc[s]
+    msg = "Invalid slice."
+    raise ValueError(msg)

src/mopipe/data/collator.py → src/mopipe/core/data/collator.py

@@ -8,7 +8,7 @@
 
 from pandas import DataFrame
 
-from mopipe.data import AbstractReader
+from mopipe.core.data import AbstractReader
 
 
 class MocapDataCollator:

src/mopipe/data/empirical.py → src/mopipe/core/data/empirical.py

@@ -4,10 +4,10 @@
 
 from pandas import DataFrame, Series
 
-from mopipe.common import MocapMetadataEntries, maybe_generate_id
+from mopipe.core.common import MocapMetadataEntries, maybe_generate_id
 
 if t.TYPE_CHECKING:
-    from mopipe.data import ExperimentLevel
+    from mopipe.core.data import ExperimentLevel
 
 
 class MetaData(dict):
@@ -16,8 +16,8 @@ class MetaData(dict):
     Base class for all metadata associated with data.
     """
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
 
 class MocapMetaData(MetaData):
@@ -27,11 +27,11 @@ class MocapMetaData(MetaData):
     known names in MocapMetadataEntries.
     """
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, **kwargs):
         for key, value in kwargs.items():
             if key in MocapMetadataEntries:
                 kwargs[MocapMetadataEntries[key]] = value
-        super().__init__(*args, **kwargs)
+        super().__init__(**kwargs)
 
     def __setitem__(self, key: str, value: t.Any):
         if key in MocapMetadataEntries:

src/mopipe/data/experiment.py → src/mopipe/core/data/experiment.py

@@ -11,12 +11,12 @@
 
     from strenum import StrEnum
 
-from mopipe.common import maybe_generate_id
+from mopipe.core.common import maybe_generate_id
 
 if t.TYPE_CHECKING:
-    from mopipe.data import EmpiricalData
+    from mopipe.core.data import EmpiricalData
 
-from mopipe.data import MetaData
+from mopipe.core.data import MetaData
 
 
 class LDType(StrEnum):

src/mopipe/data/reader.py → src/mopipe/core/data/reader.py

@@ -11,9 +11,9 @@
 
 import pandas as pd
 
-from mopipe.common import MocapMetadataEntries, maybe_generate_id
-from mopipe.common.qtm import parse_metadata_row
-from mopipe.data import EmpiricalData, MetaData, MocapMetaData, MocapTimeSeries
+from mopipe.core.common import MocapMetadataEntries, maybe_generate_id
+from mopipe.core.common.qtm import parse_metadata_row
+from mopipe.core.data import EmpiricalData, MetaData, MocapMetaData, MocapTimeSeries
 
 
 class AbstractReader(ABC):

src/mopipe/core/segments/inputs.py

@@ -0,0 +1,160 @@
+from abc import abstractmethod
+from typing import Any
+
+from mopipe.core.segments.io import IOType, IOTypeBaseMixin
+
+
+class InputTypeBaseMixin(IOTypeBaseMixin):
+    """Mixin class for all segments input types."""
+
+    _input_type: IOType
+
+    @property
+    def input_type(self) -> IOType:
+        """The type of the input."""
+        return self._input_type
+
+    def _ensure_input_exists(self, **kwargs) -> bool:
+        """Ensure that the input exists."""
+        if "x" in kwargs:
+            return True
+        return False
+
+    @abstractmethod
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        raise NotImplementedError
+
+
+class UnivariateSeriesInput(InputTypeBaseMixin):
+    """Mixin class for univariate series input segments."""
+
+    _input_type = IOType.UNIVARIATE_SERIES
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_series_or_dataframe(kwargs["x"]):
+            return False
+        if not self._validate_shape(kwargs["x"], row_min=2, col_max=1):
+            return False
+        return True
+
+
+class MultivariateSeriesInput(InputTypeBaseMixin):
+    """Mixin class for multivariate series input segments."""
+
+    _input_type = IOType.MULTIVARIATE_SERIES
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_series_or_dataframe(kwargs["x"]):
+            return False
+        if not self._validate_shape(kwargs["x"], row_min=2, col_min=2):
+            return False
+        return True
+
+
+class SingleValueInput(InputTypeBaseMixin):
+    """Mixin class for single value input segments."""
+
+    _input_type = IOType.SINGLE_VALUE
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_single_value(kwargs["x"]):
+            return False
+        return True
+
+
+class MultiValueInput(InputTypeBaseMixin):
+    """Mixin class for multiple values input segments."""
+
+    _input_type = IOType.MULTIPLE_VALUES
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_multiple_values(kwargs["x"]):
+            return False
+        return True
+
+
+class SingleNumericValueInput(InputTypeBaseMixin):
+    """Mixin class for single numeric value input segments."""
+
+    _input_type = IOType.SINGLE_NUMERIC_VALUE
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_single_numeric_value(kwargs["x"]):
+            return False
+        return True
+
+
+class AnySeriesInput(InputTypeBaseMixin):
+    """Mixin class for any series input segments."""
+
+    _input_type = IOType.ANY_SERIES
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_any_series(kwargs["x"]):
+            return False
+        return True
+
+
+class AnyNumericInput(InputTypeBaseMixin):
+    """Mixin class for any numeric input segments."""
+
+    _input_type = IOType.ANY_NUMERIC
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_any_numeric(kwargs["x"]):
+            return False
+        return True
+
+
+class AnyInput(InputTypeBaseMixin):
+    """Mixin class for any input segments."""
+
+    _input_type = IOType.ANY
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._ensure_input_exists(**kwargs):
+            return False
+        if not self._validate_any(kwargs["x"]):
+            return False
+        return True
+
+
+class OtherInput(InputTypeBaseMixin):
+    """Mixin class for other input segments."""
+
+    _input_type = IOType.OTHER
+
+    @abstractmethod
+    def _validate_other(self, x: Any) -> bool:
+        """Validate that the input is other type."""
+        msg = "Other input type must be implemented by subclass."
+        raise NotImplementedError(msg)
+
+    def validate_input(self, **kwargs) -> bool:
+        """Validate the input."""
+        if not self._validate_other(kwargs["x"]):
+            return False
+        return True

src/mopipe/segments/io.py → src/mopipe/core/segments/io.py

@@ -2,6 +2,7 @@
 import typing as t
 from enum import Enum, auto
 
+import numpy as np
 import pandas as pd
 
 
@@ -16,7 +17,10 @@ class IOType(Enum):
     MULTIVARIATE_SERIES = auto()
     SINGLE_VALUE = auto()
     MULTIPLE_VALUES = auto()
+    SINGLE_NUMERIC_VALUE = auto()
     ANY = auto()
+    ANY_SERIES = auto()
+    ANY_NUMERIC = auto()
     OTHER = auto()
 
 
@@ -54,6 +58,31 @@ def _validate_multiple_values(self, x: t.Any) -> bool:
                 return True
         return False
 
+    def _validate_single_numeric_value(self, x: t.Any) -> bool:
+        """Validate that the input is a single numeric value."""
+        if isinstance(x, (int, float, np.number)):
+            return True
+        return False
+
+    def _validate_any_series(self, x: t.Any) -> bool:
+        """Validate that the input is any series."""
+        if self._validate_series_or_dataframe(x):
+            if self._validate_shape(x, row_min=1, col_min=1):
+                return True
+        return False
+
+    def _validate_any_numeric(self, x: t.Any) -> bool:
+        """Validate that the input is any numeric value."""
+        if isinstance(x, (int, float)):
+            return True
+        if self._validate_series(x):
+            if x.dtype in [int, float]:
+                return True
+        if self._validate_dataframe(x):
+            if x.dtypes.apply(lambda x: x in [int, float]).all():
+                return True
+        return False
+
     def _validate_any(self, x: t.Any) -> bool:  # noqa: ARG002
         """Validate that the input is anything."""
         return True