Merge pull request #72 from Cumulocity-IoT/feature/jupyter-support

Measurements and Series handling fixes and improvements

Author: chsou

c8y_api/model/_base.py

@@ -4,7 +4,7 @@
 
 import logging
 from typing import Any, Iterable, Set
-from urllib.parse import quote_plus, urlencode
+from urllib.parse import urlencode
 
 from collections.abc import MutableMapping, MutableSequence
 from deprecated import deprecated
@@ -61,6 +61,9 @@ def __init__(self, dictionary: dict, on_update=None):
         self.__dict__['_property_items'] = dictionary
         self.__dict__['_property_on_update'] = on_update
 
+    def __repr__(self):
+        return f'{type(self).__name__}({self.__dict__["_property_items"]})'
+
     def has(self, name: str):
         """Check whether a key is present in the dictionary."""
         return name in self.__dict__['_property_items']
@@ -108,6 +111,9 @@ def __init__(self, values: list, on_update=None):
         self.__dict__['_property_items'] = values
         self.__dict__['_property_on_update'] = on_update
 
+    def __repr__(self):
+        return f'{type(self).__name__}({self.__dict__["_property_items"]})'
+
     def __getitem__(self, i):
         item = self.__dict__['_property_items'][i]
         if isinstance(item, dict):
@@ -423,7 +429,7 @@ def __setitem__(self, name: str, fragment: str | bool | int | float | dict | lis
         (specified as nested dictionary).::
 
             obj['c8y_SimpleValue'] = 14
-            obj['c8y_ComplexValue'] = { 'x': 1, 'y': 2, 'text': 'message'}
+            obj['c8y_ComplexValue'] = { ('x', 1, 'y': 2), 'text': 'message'}
 
         Args:
             name (str):  Name of the custom fragment.
@@ -642,7 +648,6 @@ def _map_params(
             name=None,
             fragment=None,
             source=None,  # noqa (type)
-            value=None,
             series=None,
             owner=None,
             device_id=None,
@@ -669,7 +674,7 @@ def _map_params(
             reverse=None,
             page_size=None,
             page_number=None,  # (must not be part of the prepared query)
-            **kwargs) -> dict:
+            **kwargs) -> list[tuple]:
         assert not page_number
 
         def multi(*xs):
@@ -716,6 +721,7 @@ def multi(*xs):
             'owner': owner,
             'source': source,
             'fragmentType': fragment,
+            # 'series': series,
             'deviceId': device_id,
             'agentId': agent_id,
             'bulkId': bulk_id,
@@ -730,13 +736,19 @@ def multi(*xs):
             'lastUpdatedTo': updated_to,
             'withSourceAssets': with_source_assets,
             'withSourceDevices': with_source_devices,
-            'revert': str(reverse) if reverse is not None else None,
+            'revert': str(reverse).lower() if reverse is not None else None,
             'pageSize': page_size}.items() if v is not None}
         params.update({_StringUtil.to_pascal_case(k): v for k, v in kwargs.items() if v is not None})
-        return params
+        tuples = list(params.items())
+        if series:
+            if isinstance(series, list):
+                tuples += [('series', s) for s in series]
+            else:
+                tuples.append(('series', series))
+        return tuples
 
     def _prepare_query(self, resource: str = None, expression: str = None, **kwargs):
-        encoded = quote_plus(expression) if expression else urlencode(self._map_params(**kwargs))
+        encoded = expression or urlencode(self._map_params(**kwargs))
         if not encoded:
             return resource or self.resource
         return (resource or self.resource) + '?' + encoded

c8y_api/model/measurements.py

@@ -716,19 +716,11 @@ def get_series(
 
         See also: https://cumulocity.com/api/core/#operation/getMeasurementSeriesResource
         """
-        # The 'series' parameter has to be added through a hack; it
-        # may be a list and because 'series' is by default converted to
-        # the 'valueFragmentSeries' parameter
-
-        if series:
-            series = series if isinstance(series, str) else ','.join(series)
-
         base_query = self._prepare_query(
             resource=f'{self.resource}/series',
             expression=expression,
             source=source,
-            # this is a non-mapped parameter
-            aggregationType=aggregation,
+            aggregationType=aggregation,  # this is a non-mapped parameter
             series=series,
             before=before,
             after=after,
@@ -738,14 +730,81 @@ def get_series(
             **kwargs)
         return Series(self.c8y.get(base_query))
 
+    def collect_series(
+            self,
+            expression: str = None,
+            source: str = None,
+            aggregation: str = None,
+            series: str | Sequence[str] = None,
+            before: str | datetime = None,
+            after: str | datetime = None,
+            min_age: timedelta = None,
+            max_age: timedelta = None,
+            reverse: bool = None,
+            value: str = None,
+            timestamps: bool|str = None,
+            **kwargs
+    ):
+        """Query the database for series values.
+
+        This function is functionally the same as using the `get_series` function
+        with an immediate `collect` on the result.
+
+        Args:
+            expression (str):  Arbitrary filter expression which will be
+                passed to Cumulocity without change; all other filters
+                are ignored if this is provided
+            source (str):  Database ID of a source device
+            aggregation (str):  Aggregation type
+            series (str|Sequence[str]):  Series' to query and collect; If
+                multiple series are collected each element in the result will
+                be a tuple. If omitted, all available series are collected.
+            before (datetime|str):  Datetime object or ISO date/time string.
+                Only measurements assigned to a time before this date are
+                included.
+            after (datetime|str):  Datetime object or ISO date/time string.
+                Only measurements assigned to a time after this date are
+                included.
+            min_age (timedelta):  Timedelta object. Only measurements of
+                at least this age are included.
+            max_age (timedelta):  Timedelta object. Only measurements with
+                at most this age are included.
+            reverse (bool):  Invert the order of results, starting with the
+                most recent one.
+            value (str):  Which value (min/max) to collect. If omitted, both
+                values will be collected, grouped as 2-tuples.
+            timestamps (bool|str):  Whether each element in the result list will
+                be prepended with the corresponding timestamp. If True, the
+                timestamp string will be included; Use 'datetime' or 'epoch' to
+                parse the timestamp string.
+
+        Returns:
+            A simple list or list of tuples (potentially nested) depending on the
+            parameter combination.
+
+        See also: https://cumulocity.com/api/core/#operation/getMeasurementSeriesResource
+        """
+        result = self.get_series(
+            expression=expression,
+            source=source,
+            aggregation=aggregation,
+            series=series,
+            before=before,
+            after=after,
+            min_age=min_age,
+            max_age=max_age,
+            reverse=reverse,
+            **kwargs)
+        return result.collect(
+            series=series,
+            value=value,
+            timestamps=timestamps)
+
     def delete_by(
             self,
             expression: str = None,
             type: str = None,
             source: str | int = None,
-            # value_fragment_type: str = None,
-            # value_fragment_series: str = None,
-            # series: str = None,
             date_from: str | datetime = None,
             date_to: str | datetime = None,
             before: str | datetime = None,
@@ -785,9 +844,6 @@ def delete_by(
             expression=expression,
             type=type,
             source=source,
-            # value_fragment_type=value_fragment_type,
-            # value_fragment_series=value_fragment_series,
-            # series=series,
             date_from=date_from,
             date_to=date_to,
             before=before,

integration_tests/test_measurements.py

@@ -261,3 +261,37 @@ def test_collect_multiple_series(series_fixture, request):
     for i in range(0, len(series_names)):
         t = type(values[0][i])
         assert all(isinstance(v[i], t) for v in values if v[i])
+
+
+def test_get_and_collect_series(live_c8y, sample_series_device):
+    """Verify that get & collect works as expected."""
+    series = live_c8y.measurements.get_series(
+        source=sample_series_device.id,
+        series=sample_series_device.c8y_SupportedSeries,
+        aggregation=Measurements.AggregationType.HOURLY,
+        after='1970-01-01',
+        before='now'
+    )
+
+    # multiple series
+    collected = series.collect(sample_series_device.c8y_SupportedSeries)
+    directly_collected = live_c8y.measurements.collect_series(
+        source=sample_series_device.id,
+        series=sample_series_device.c8y_SupportedSeries,
+        aggregation=Measurements.AggregationType.HOURLY,
+        after='1970-01-01',
+        before='now'
+    )
+    assert collected == directly_collected
+
+    # single series
+    for series_name in sample_series_device.c8y_SupportedSeries:
+        collected = series.collect(series_name)
+        directly_collected = live_c8y.measurements.collect_series(
+            source=sample_series_device.id,
+            series=series_name,
+            aggregation=Measurements.AggregationType.HOURLY,
+            after='1970-01-01',
+            before='now'
+        )
+        assert collected == directly_collected

integration_tests/test_notification2.py

@@ -418,4 +418,3 @@ async def receive_notification(m:AsyncListener.Message):
     # (99) cleanup
     await asyncio.gather(*[l.close() for l in listeners])
     await asyncio.wait(listener_tasks)
-    await asyncio.all_tasks()