ENH: Add wrapper for new endpoint for fetching aggregated timeseries (#120)

* ESS-2234

Author: branislav-jenco-4ss

datareservoirio/_utils.py

@@ -63,3 +63,33 @@ def as_binary_csv(self):
             df.to_csv(fp, lineterminator="\n", **kwargs)
             csv = fp.getvalue()
         return csv
+
+
+# Translation of user input parameters of the samples/aggregate method for more convenient use (matching pandas)
+
+function_translation = {"std": "Stdev", "mean": "Avg"}
+
+period_translation = {
+    "hours": "h",
+    "hour": "h",
+    "hr": "h",
+    "h": "h",
+    "minutes": "m",
+    "minute": "m",
+    "min": "m",
+    "m": "m",
+    "seconds": "s",
+    "second": "s",
+    "sec": "s",
+    "s": "s",
+    "milliseconds": "ms",
+    "millisecond": "ms",
+    "millis": "ms",
+    "milli": "ms",
+    "ms": "ms",
+    "microseconds": "microsecond",
+    "microsecond": "microsecond",
+    "micros": "microsecond",
+    "micro": "microsecond",
+    "tick": "tick",
+}

datareservoirio/client.py

@@ -3,8 +3,10 @@
 import warnings
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
 from functools import wraps
 from operator import itemgetter
+from urllib.parse import urlencode
 from uuid import uuid4
 
 import pandas as pd
@@ -19,6 +21,7 @@
 )
 
 from ._logging import log_decorator
+from ._utils import function_translation, period_translation
 from .globalsettings import environment
 from .storage import Storage
 
@@ -285,7 +288,7 @@ def delete(self, series_id):
         )
 
     def _timer(func):
-        """Decorator used to log latency of the ``get`` method"""
+        """Decorator used to log latency of the ``get`` and ``get_samples_aggregate`` method"""
 
         @wraps(func)
         def wrapper(self, series_id, start=None, end=None, **kwargs):
@@ -411,6 +414,148 @@ def get(
 
         return series
 
+    @log_decorator("exception")
+    @_timer
+    @log_decorator("warning")
+    def get_samples_aggregate(
+        self,
+        series_id,
+        start=None,
+        end=None,
+        aggregation_period=None,
+        aggregation_function=None,
+        max_page_size=None,
+    ):
+        """
+        Retrieve a series from DataReservoir.io using the samples/aggregate endpoint.
+
+        Parameters
+        ----------
+        series_id : str
+            Identifier of the series to download
+        start: required
+            Start time (inclusive) of the aggregated series given as anything
+            pandas.to_datetime is able to parse. Date must be within the past 90 days.
+        end:
+            Stop time (exclusive) of the aggregated series given as anything
+            pandas.to_datetime is able to parse. Date must be within the past 90 days.
+        aggregation_function : str
+            One of "Avg", "Min", "Max", "Stdev".
+        aggregation_period : str
+            Used in combination with aggregation function to specify the period for aggregation.
+            Aggregation period is maximum 24 hours. Values can be in units of h, m, s, ms,
+            microsecond or tick. Use 100 ms instead of 0.1s for 10Hz.
+        max_page_size : optional
+            Maximum number of samples to return per page. The method automatically follows links
+            to next pages and returns the entire series. For advanced usage.
+        Returns
+        -------
+        pandas.Series
+            Series data
+        """
+        if not start:
+            # Required parameter
+            raise ValueError(
+                "You must specify the start date in ISO 8601 format, for example 2023-12-01"
+            )
+
+        if not end:
+            # Required parameter
+            raise ValueError(
+                "You must specify the end date in ISO 8601 format, for example 2023-12-31."
+            )
+
+        if not aggregation_period:
+            # Required parameter
+            raise ValueError(
+                "Aggregation period must be specified using integers and one of these units: h, m, s, ms, microsecond or tick, or their Pandas equivalents"
+            )
+
+        if not aggregation_function:
+            # Required parameter
+            raise ValueError(
+                "Aggregation function must be one of: Avg (mean), Min, Max, Stdev (std)"
+            )
+
+        # Translating some pandas terms to API terms
+        # Note the API is case insensitive so both min and Min will work
+        if aggregation_function in function_translation:
+            aggregation_function = function_translation[aggregation_function]
+
+        for period_unit in period_translation:
+            if (
+                aggregation_period.endswith(period_unit)
+                and aggregation_period[-len(period_unit) - 1].isnumeric()
+            ):
+                aggregation_period = (
+                    aggregation_period[: -len(period_unit)]
+                    + period_translation[period_unit]
+                )
+                break
+
+        start = pd.to_datetime(start, dayfirst=True, unit="ns", utc=True).isoformat()
+        end = pd.to_datetime(end, dayfirst=True, unit="ns", utc=True).isoformat()
+
+        params = {}
+
+        if max_page_size:
+            params["maxPageSize"] = max_page_size
+
+        params["aggregationPeriod"] = aggregation_period
+        params["aggregationFunction"] = aggregation_function
+        params["start"] = start
+        params["end"] = end
+
+        next_page_link = f"{environment.api_base_url}reservoir/timeseries/{series_id}/samples/aggregate?{urlencode(params)}"
+
+        df = (
+            pd.DataFrame(columns=("index", "values"))
+            .astype({"index": "int64"})
+            .astype({"values": "float64"}, errors="ignore")
+        )
+
+        @retry(
+            stop=stop_after_attempt(
+                4
+            ),  # Attempt!, not retry attempt. Attempt 2, is 1 retry
+            retry=retry_if_exception_type(
+                (
+                    ConnectionError,
+                    requests.exceptions.ChunkedEncodingError,
+                    requests.ReadTimeout,
+                    ConnectionRefusedError,
+                    requests.ConnectionError,
+                )
+            ),
+            wait=wait_chain(*[wait_fixed(0.1), wait_fixed(0.5), wait_fixed(30)]),
+        )
+        def get_samples_aggregate_page(url):
+            return self._auth_session.get(
+                url,
+                timeout=_TIMEOUT_DEAULT,
+            )
+
+        while next_page_link:
+            response = get_samples_aggregate_page(next_page_link)
+            response.raise_for_status()
+            response_json = response.json()
+            next_page_link = response_json.get("@odata.nextLink", None)
+
+            content = [
+                (pd.to_datetime(sample["Timestamp"], utc=True), sample["Value"])
+                for sample in response_json["value"]
+            ]
+
+            new_df = pd.DataFrame(
+                content, columns=("index", "values"), copy=False
+            ).astype({"values": "float64"}, errors="ignore")
+
+            df = pd.concat([df, new_df])
+
+        series = df.set_index("index").squeeze("columns").copy(deep=True)
+
+        return series
+
     def set_metadata(
         self,
         series_id,

docs/user_guide/advanced_config.rst

@@ -141,4 +141,11 @@ Instrumentation
 
 For monitoring purposes, the external logger can be enabled to report errors and performance metrics to 4insight Team.  
 
-To enable logging, environmental variable ``DRIO_PYTHON_APPINSIGHTS`` needs to be set to ``true``.
+To enable logging, environmental variable ``DRIO_PYTHON_APPINSIGHTS`` needs to be set to ``true``.
+
+Using the :py:mod:`max_page_size` parameter in :py:mod:`get_samples_aggregate` method
+-------------------------------------------------------------------------------------
+
+The :py:meth:`Client.get_samples_aggregate` method uses an endpoint that has support for paging of responses. This means that instead of making one big request, it might make a series of smaller requests traversing links to next pages returned in each partial response.
+
+Normally this is something you don't have to think about. In case you do want to change the maximum number of results returned in one page, you can use the parameter called ``max_page_size`` to alter this number. 

docs/user_guide/manage_series.rst

@@ -124,6 +124,26 @@ You can access any data you have ``TimeSeriesId`` (and authorization) for:
     :py:meth:`Client.get` returns :py:class:`pandas.Series`.
 
 
+Access existing data with aggregation
+-------------------------------------
+
+You can also access any data you have ``TimeSeriesId`` (and authorization) for with applied aggregation using:
+
+.. code-block:: python
+
+    # Get entire timeseries
+    timeseries = client.get_samples_aggregate(series_id, start='2018-01-01',
+                            end='2018-01-02', aggregation_period='15m',
+                            aggregation_function='Avg')
+
+.. note::
+
+    :py:meth:`Client.get_samples_aggregate` also returns :py:class:`pandas.Series`. The :py:mod:`start`, :py:mod:`end`, :py:mod:`aggregation_period` and :py:mod:`aggregation_function` parameters are required.   
+
+.. important::
+
+    Retrieving aggregated data is available only for the last 90 days.
+
 Delete data
 -----------
 
@@ -140,3 +160,5 @@ is removed from the `DataReservoir.io`_ inventory:
 .. _Pandas: https://pandas.pydata.org/
 
 
+
+

integration_tests/test_aggregate_samples.py

@@ -0,0 +1,115 @@
+import os
+import time
+from datetime import datetime, timedelta
+
+import numpy as np
+import pandas as pd
+
+import datareservoirio as drio
+
+
+def wait_for_data_to_appear(func, delay=3):
+    count = 0
+    while count == 0:
+        series = func()
+        count = len(series)
+        if count == 0:
+            time.sleep(delay)
+        else:
+            return series
+
+
+def test_non_paged(cleanup_series):
+    """
+    Integration test for the samples/aggregate endpoint.
+
+    Tests the following:
+        * Creates a non-empty timeseries in DataReservoir.io.
+        * Fetches it using the samples/aggregate endpoint.
+        * Checks that the data matches with what was created.
+    """
+
+    # Initialize client
+    auth_session = drio.authenticate.ClientAuthenticator(
+        os.getenv("DRIO_CLIENT_ID"), os.getenv("DRIO_CLIENT_SECRET")
+    )
+    client = drio.Client(auth_session, cache=False)
+
+    # Create some dummy data
+    # Relative from today, because we have the 3 month limitation
+    end = datetime.today().replace(hour=0, minute=0, second=0, microsecond=0)
+    start = end - timedelta(hours=1)
+    freq = pd.to_timedelta(1, "s")
+    index = pd.date_range(
+        start, end, freq=freq, tz="utc", inclusive="left", name="index"
+    )
+    series = pd.Series(data=np.random.random(len(index)), index=index, name="values")
+    response_create = client.create(series=series, wait_on_verification=True)
+    series_id = response_create["TimeSeriesId"]
+
+    # For cleaning up after test runs
+    cleanup_series.add(series_id)
+
+    # Get data from DataReservoir.io using the samples/aggregate endpoint
+    get_timeseries = lambda: client.get_samples_aggregate(
+        series_id,
+        start=start,
+        end=end,
+        aggregation_function="Avg",
+        aggregation_period="1s",
+    )
+
+    # Wait for data to be available
+    series_fetched = wait_for_data_to_appear(get_timeseries)
+
+    # Check downloaded data
+    pd.testing.assert_series_equal(series, series_fetched, check_freq=False)
+
+
+def test_paged(cleanup_series):
+    """
+    Integration test for the samples/aggregate endpoint.
+
+    Tests the following:
+        * Creates a non-empty timeseries in DataReservoir.io.
+        * Fetches it using the samples/aggregate endpoint and checks that the data is matches with what was created.
+        * uses a maxPageSize of 1000 which means the response will be paged (into 4 pages)
+    """
+
+    # Initialize client
+    auth_session = drio.authenticate.ClientAuthenticator(
+        os.getenv("DRIO_CLIENT_ID"), os.getenv("DRIO_CLIENT_SECRET")
+    )
+    client = drio.Client(auth_session, cache=False)
+
+    # Create some dummy data
+    # Relative from today, because we have the 3 month limitation
+    end = datetime.today().replace(hour=0, minute=0, second=0, microsecond=0)
+    start = end - timedelta(hours=1)
+    freq = pd.to_timedelta(1, "s")
+
+    index = pd.date_range(
+        start, end, freq=freq, tz="utc", inclusive="left", name="index"
+    )
+    series = pd.Series(data=np.random.random(len(index)), index=index, name="values")
+    response_create = client.create(series=series, wait_on_verification=True)
+    series_id = response_create["TimeSeriesId"]
+
+    # For cleaning up after test runs
+    cleanup_series.add(series_id)
+
+    # Get data from DataReservoir.io using the samples/aggregate endpoint
+    check_func = lambda: client.get_samples_aggregate(
+        series_id,
+        start=start,
+        end=end,
+        aggregation_function="Avg",
+        aggregation_period="1s",
+        max_page_size=1000,
+    )
+
+    # Wait for data to be available
+    series_fetched = wait_for_data_to_appear(check_func)
+
+    # Check downloaded data
+    pd.testing.assert_series_equal(series, series_fetched, check_freq=False)

pyproject.toml

@@ -48,6 +48,7 @@ version = {attr = "datareservoirio.__version__"}
 
 [tool.pytest.ini_options]
 pythonpath = [".", "src"]
+markers = ["response_irrelevant"]
 
 [tool.tox]
 legacy_tox_ini = """