added aggregate class method on SingleTimeSeries class

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "infrasys"
-version = "0.1.1"
+version = "0.1.2"
 description = ''
 readme = "README.md"
 requires-python = ">=3.10, <3.13"

src/infrasys/exceptions.py

@@ -35,3 +35,7 @@ class ISNotStored(ISBaseException):
 
 class ISOperationNotAllowed(ISBaseException):
     """Raised if the requested operation is not allowed."""
+
+
+class InconsistentTimeseriesAggregation(ISBaseException):
+    """Raised if attempting to aggregate inconsistent time series data."""

src/infrasys/time_series_models.py

@@ -4,7 +4,16 @@
 import importlib
 from datetime import datetime, timedelta
 from enum import Enum
-from typing import Any, Literal, Optional, Type, TypeAlias, Union, Sequence
+from typing import (
+    Any,
+    Literal,
+    Optional,
+    Self,
+    Type,
+    TypeAlias,
+    Union,
+    Sequence,
+)
 from uuid import UUID
 
 import numpy as np
@@ -21,7 +30,10 @@
 from typing_extensions import Annotated
 
 from infrasys.base_quantity import BaseQuantity
-from infrasys.exceptions import ISConflictingArguments
+from infrasys.exceptions import (
+    ISConflictingArguments,
+    InconsistentTimeseriesAggregation,
+)
 from infrasys.models import InfraSysBaseModelWithIdentifers, InfraSysBaseModel
 from infrasys.normalization import NormalizationModel
 
@@ -89,6 +101,73 @@ def check_data(
 
         return data
 
+    @classmethod
+    def aggregate(cls, ts_data: list[Self]) -> Self:
+        """Method to aggregate list of SingleTimeSeries data.
+
+        Parameters
+        ----------
+        ts_data
+            list of SingleTimeSeries data
+
+        Returns
+        -------
+        SingleTimeSeries
+
+        Raises
+        ------
+        InconsistentTimeseriesAggregation
+            Raised if incompatible timeseries data are passed.
+        """
+
+        # Extract unique properties from ts_data
+        unique_props = {
+            "length": {data.length for data in ts_data},
+            "resolution": {data.resolution for data in ts_data},
+            "start_time": {data.initial_time for data in ts_data},
+            "variable": {data.variable_name for data in ts_data},
+            "data_type": {type(data.data) for data in ts_data},
+        }
+
+        # Validate uniformity across properties
+        if any(len(prop) != 1 for prop in unique_props.values()):
+            inconsistent_props = {k: v for k, v in unique_props.items() if len(v) > 1}
+            msg = f"Inconsistent timeseries data: {inconsistent_props}"
+            raise InconsistentTimeseriesAggregation(msg)
+
+        # Aggregate data
+        is_quantity = issubclass(next(iter(unique_props["data_type"])), BaseQuantity)
+        magnitude_type = (
+            type(ts_data[0].data.magnitude)
+            if is_quantity
+            else next(iter(unique_props["data_type"]))
+        )
+
+        # Aggregate data based on magnitude type
+        if issubclass(magnitude_type, pa.Array):
+            new_data = sum(
+                [
+                    data.data.to_numpy() * (data.data.units if is_quantity else 1)
+                    for data in ts_data
+                ]
+            )
+        elif issubclass(magnitude_type, np.ndarray):
+            new_data = sum([data.data for data in ts_data])
+        elif issubclass(magnitude_type, list) and not is_quantity:
+            new_data = sum([np.array(data) for data in ts_data])
+        else:
+            msg = f"Unsupported data type for aggregation: {magnitude_type}"
+            raise TypeError(msg)
+
+        # Return new SingleTimeSeries instance
+        return SingleTimeSeries(
+            data=new_data,
+            variable_name=unique_props["variable"].pop(),
+            initial_time=unique_props["start_time"].pop(),
+            resolution=unique_props["resolution"].pop(),
+            normalization=None,
+        )
+
     @classmethod
     def from_array(
         cls,

tests/test_single_time_series.py

@@ -100,3 +100,31 @@ def test_normalization():
     assert ts.length == len(data)
     for i, val in enumerate(ts.data):
         assert val.as_py() == data[i] / max_val
+
+
+def test_normal_array_aggregate():
+    length = 10
+    initial_time = datetime(year=2020, month=1, day=1)
+    time_array = [initial_time + timedelta(hours=i) for i in range(length)]
+    data = [1.1, 2.2, 3.3, 4.5, 5.5]
+    variable_name = "active_power"
+    ts1 = ts2 = SingleTimeSeries.from_time_array(
+        data, variable_name, time_array, normalization=None
+    )
+    ts_agg = SingleTimeSeries.aggregate([ts1, ts2])
+    assert isinstance(ts_agg, SingleTimeSeries)
+    assert list([el.as_py() for el in ts_agg.data]) == [2.2, 4.4, 6.6, 9, 11]
+
+
+def test_pint_array_aggregate():
+    length = 10
+    initial_time = datetime(year=2020, month=1, day=1)
+    time_array = [initial_time + timedelta(hours=i) for i in range(length)]
+    data = ActivePower([1.1, 2.2, 3.3, 4.5, 5.5], "kilowatts")
+    variable_name = "active_power"
+    ts1 = ts2 = SingleTimeSeries.from_time_array(
+        data, variable_name, time_array, normalization=None
+    )
+    ts_agg = SingleTimeSeries.aggregate([ts1, ts2])
+    assert isinstance(ts_agg, SingleTimeSeries)
+    assert list(ts_agg.data.magnitude) == [2.2, 4.4, 6.6, 9, 11]