Repository: kangasta/fdbk

Issue:
Data tools should not aggregate data if len(data) < aggregate_to

Discussion:

Patch:
diff --git a/fdbk/_db_connection.py b/fdbk/_db_connection.py
index a92fd00..af155fe 100644
--- a/fdbk/_db_connection.py
+++ b/fdbk/_db_connection.py
@@ -212,7 +212,8 @@ class DBConnection:
             until=None,
             limit=None,
             aggregate_to=None,
-            aggregate_with=None):
+            aggregate_with=None,
+            aggregate_always=False):
         '''Get summary of the topic data
 
         Args:
@@ -222,6 +223,8 @@ class DBConnection:
             limit: Number of entries to include from the most recent
             aggregate_to: Aggregate data into specified number of data points
             aggregate_with: Aggregate data with speficied function
+            aggregate_always: Aggregate data even if datas length is
+                shorter than aggregate_to value. Disabled by default.
 
         Returns:
             Dictionary with summary of the topic
@@ -242,7 +245,7 @@ class DBConnection:
         }
 
         results, warnings = run_data_tools(
-            topic_d, data_d, aggregate_to, aggregate_with)
+            topic_d, data_d, aggregate_to, aggregate_with, aggregate_always,)
         summary_d["warnings"].extend(warnings)
 
         results, warnings = post_process(results)
@@ -258,9 +261,15 @@ class DBConnection:
             until=None,
             limit=None,
             aggregate_to=None,
-            aggregate_with=None):
+            aggregate_with=None,
+            aggregate_always=False):
         data_d = self.get_data(topic_d.get("id"), since, until, limit)
-        return run_data_tools(topic_d, data_d, aggregate_to, aggregate_with)
+        return run_data_tools(
+            topic_d,
+            data_d,
+            aggregate_to,
+            aggregate_with,
+            aggregate_always)
 
     def _run_data_tools_for_many(self,
                                  topic_ids=None,
@@ -269,7 +278,8 @@ class DBConnection:
                                  until=None,
                                  limit=None,
                                  aggregate_to=None,
-                                 aggregate_with=None):
+                                 aggregate_with=None,
+                                 aggregate_always=False):
         executor = ThreadPoolExecutor()
         warnings = []
 
@@ -293,7 +303,14 @@ class DBConnection:
         }
 
         jobs = []
-        params = (since, until, limit, aggregate_to, aggregate_with,)
+        params = (
+            since,
+            until,
+            limit,
+            aggregate_to,
+            aggregate_with,
+            aggregate_always,
+        )
         for topic_d in topics.values():
             jobs.append(
                 executor.submit(
@@ -324,7 +341,8 @@ class DBConnection:
             until=None,
             limit=None,
             aggregate_to=None,
-            aggregate_with=None):
+            aggregate_with=None,
+            aggregate_always=False):
         '''Get overview of the data
 
         Args:
@@ -337,6 +355,8 @@ class DBConnection:
             limit: Number of entries to include from the most recent
             aggregate_to: Aggregate data into specified number of data points
             aggregate_with: Aggregate data with speficied function
+            aggregate_always: Aggregate data even if datas length is
+                shorter than aggregate_to value. Disabled by default.
 
         Returns:
             Dictionary with overview of the topics data
@@ -351,7 +371,8 @@ class DBConnection:
             until=until,
             limit=limit,
             aggregate_to=aggregate_to,
-            aggregate_with=aggregate_with)
+            aggregate_with=aggregate_with,
+            aggregate_always=aggregate_always)
 
 
 ConnectionClass = DBConnection
diff --git a/fdbk/data_tools/_aggregate.py b/fdbk/data_tools/_aggregate.py
index 2283bbe..861cbe5 100644
--- a/fdbk/data_tools/_aggregate.py
+++ b/fdbk/data_tools/_aggregate.py
@@ -21,7 +21,20 @@ def _get_keys(data_point):
     return [key for key in data_point if key != 'timestamp']
 
 
-def aggregate(data, aggregate_to, aggregate_with=None):
+def aggregate(data, aggregate_to, aggregate_with=None, aggregate_always=False):
+    '''Aggregate data to less data points
+
+    Args:
+        data: Data before aggregation
+        aggregate_to: Number of data points to aggregate data to.
+        aggregate_with: Value function to use to when combining data-points.
+            Defaults to average.
+        aggregate_always: If true, data is aggregated even if datas length is
+            shorter than aggregate_to value. Disabled by default.
+
+    Returns:
+        List of aggregated data-points
+    '''
     if not aggregate_with:
         aggregate_with = 'average'
 
@@ -32,6 +45,9 @@ def aggregate(data, aggregate_to, aggregate_with=None):
         warnings.append(no_data())
         return ([], warnings,)
 
+    if len(data) <= aggregate_to and not aggregate_always:
+        return (data, warnings,)
+
     if aggregate_with not in VALUE_FUNCS:
         warnings.append(method_not_supported(aggregate_with))
         return ([], warnings,)
diff --git a/fdbk/data_tools/_utils.py b/fdbk/data_tools/_utils.py
index d55fe52..97b0b85 100644
--- a/fdbk/data_tools/_utils.py
+++ b/fdbk/data_tools/_utils.py
@@ -184,7 +184,12 @@ def post_process(statistics):
     return _process(funcs, statistics)
 
 
-def run_data_tools(topic_d, data, aggregate_to=None, aggregate_with=None):
+def run_data_tools(
+        topic_d,
+        data,
+        aggregate_to=None,
+        aggregate_with=None,
+        aggregate_always=False):
     results = []
     warnings = []
 
@@ -194,7 +199,7 @@ def run_data_tools(topic_d, data, aggregate_to=None, aggregate_with=None):
 
     if aggregate_to:
         chart_data, aggregate_warnings = aggregate(
-            data, aggregate_to, aggregate_with)
+            data, aggregate_to, aggregate_with, aggregate_always)
         warnings.extend(aggregate_warnings)
     else:
         chart_data = data
diff --git a/fdbk/server/_server_handlers.py b/fdbk/server/_server_handlers.py
index 2bd815a..32a3538 100644
--- a/fdbk/server/_server_handlers.py
+++ b/fdbk/server/_server_handlers.py
@@ -4,6 +4,10 @@
 from dateutil.parser import isoparse
 
 
+def _parse_boolean(param):
+    return str(param).lower() == 'true'
+
+
 def _parse_param(param, parser):
     try:
         return parser(param)
@@ -22,8 +26,13 @@ def parse_filter_parameters(args, include_aggregate=False):
         return query
 
     aggregate = dict(
-        aggregate_to=_parse_param(args.get('aggregate_to'), int),
-        aggregate_with=args.get('aggregate_with')
+        aggregate_to=_parse_param(
+            args.get('aggregate_to'),
+            int),
+        aggregate_with=args.get('aggregate_with'),
+        aggregate_always=_parse_param(
+            args.get('aggregate_always'),
+            _parse_boolean),
     )
 
     return {**aggregate, **query}


Test patch:
diff --git a/tst/test_data_tools.py b/tst/test_data_tools.py
index 0f3f273..08470b7 100644
--- a/tst/test_data_tools.py
+++ b/tst/test_data_tools.py
@@ -4,11 +4,23 @@ from unittest.mock import Mock, patch
 from fdbk.data_tools import aggregate, functions, run_data_tools
 from fdbk.utils.messages import method_not_supported, no_data
 
-def _test_timestamp(i):
+def _test_timestamp(i, timestamps=None):
+    if timestamps:
+        return timestamps[i]
     return f'2020-08-23T00:{i // 60:02}:{i % 60:02}Z'
 
-def generate_test_data(N=10):
-    return [dict(number=i, number2=i*2, letter=chr(ord('A') + i), timestamp=_test_timestamp(i)) for i in range(N)]
+def generate_test_data(N=10, timestamps=None):
+    return [dict(number=i, number2=i*2, letter=chr(ord('A') + i), timestamp=_test_timestamp(i, timestamps)) for i in range(N)]
+
+AGGREGATE_ALWAYS_TOPIC = dict(
+    name="Aggregate test",
+    fields=["number", "number2", "letter"],
+    data_tools=[{"field":"number", "method":"line"},])
+AGGREGATE_ALWAYS_DATA = generate_test_data(4, [
+    '2020-09-12T00:00:00Z',
+    '2020-09-12T00:00:01Z',
+    '2020-09-12T00:00:02Z',
+    '2020-09-13T00:00:00Z',])
 
 class DataToolsTest(TestCase):
     def test_summary_funcs_return_none_on_empty_data(self):
@@ -72,7 +84,7 @@ class DataToolsTest(TestCase):
 
     def test_aggregate_empty_window(self):
         data = generate_test_data(5)
-        aggregated, warnings = aggregate(data, 10, 'max')
+        aggregated, warnings = aggregate(data, 10, 'max', aggregate_always=True)
 
         for i in range(1, 5):
             self.assertNotEqual(aggregated[i].get('timestamp'), _test_timestamp(i))
@@ -80,8 +92,22 @@ class DataToolsTest(TestCase):
             self.assertEqual(aggregated[i].get('number2'), i * 2)
 
     def test_aggregate_unknown_data_tool(self):
-        data = generate_test_data(5)
+        data = generate_test_data(15)
         aggregated, warnings = aggregate(data, 10, 'horse')
 
         self.assertEqual(aggregated, [])
         self.assertEqual(warnings, [method_not_supported('horse')])
+
+    def test_aggregate_always(self):
+        data = AGGREGATE_ALWAYS_DATA
+        aggregated, warnings = aggregate(data, 5)
+
+        self.assertEqual(aggregated, data)
+        self.assertEqual(warnings, [])
+
+        aggregated, warnings = aggregate(data, 5, aggregate_always=True)
+
+        self.assertEqual(len(aggregated), 2)
+        self.assertEqual(aggregated[0]["number"], 1)
+        self.assertEqual(aggregated[1]["number"], 3)
+        self.assertEqual(warnings, [])
diff --git a/tst/test_server_handlers.py b/tst/test_server_handlers.py
index c247bfd..9cc773d 100644
--- a/tst/test_server_handlers.py
+++ b/tst/test_server_handlers.py
@@ -1,4 +1,5 @@
 from datetime import datetime
+from dateutil.parser import isoparse
 from dateutil.tz import tzutc
 
 from unittest import TestCase
@@ -8,6 +9,8 @@ from fdbk import DictConnection
 from fdbk.server import parse_filter_parameters, ServerHandlers, generate_app
 from fdbk.server._server_handlers import _get_overwrite
 
+from test_data_tools import AGGREGATE_ALWAYS_DATA, AGGREGATE_ALWAYS_TOPIC
+
 class ServerHandlersTest(TestCase):
     def _assert_status(self, expected_status, function, *args, **kwargs):
         data, status = function(*args, **kwargs)
@@ -100,7 +103,6 @@ class ServerHandlersTest(TestCase):
         data = response["statistics"][0]["payload"]["data"]["datasets"][0]["data"]
         self.assertLessEqual(len(data), 3)
 
-
     def test_add_data_validations(self):
         s = ServerHandlers(DictConnection())
 
@@ -115,3 +117,22 @@ class ServerHandlersTest(TestCase):
         self._assert_status(404, s.get_topic, None)
         topic_id = self._create_topic(s, dict(name="topic", fields=["number"]))
         self._assert_status(200, s.get_topic, topic_id)
+
+    def test_aggregate(self):
+        s = ServerHandlers(DictConnection())
+        topic_id = self._create_topic(s, AGGREGATE_ALWAYS_TOPIC)
+        for data in AGGREGATE_ALWAYS_DATA:
+            data['timestamp'] = isoparse(data['timestamp']).replace(tzinfo=None)
+            self._assert_status(200, s.add_data, topic_id, data)
+
+        tests = [
+            (s.get_summary, (topic_id, dict(aggregate_to="5", aggregate_with='sum')), 4),
+            (s.get_summary, (topic_id, dict(aggregate_to="5", aggregate_with='sum', aggregate_always="tRuE")), 2),
+            (s.get_overview, (None, dict(aggregate_to="5", aggregate_with='sum')), 4),
+            (s.get_overview, (None, dict(aggregate_to="5", aggregate_with='sum', aggregate_always="True")), 2),
+        ]
+
+        for fn, params, count in tests:
+            data = self._assert_status(200, fn, *params)
+            aggregated = data['statistics'][0]['payload']['data']['datasets'][0]['data']
+            self.assertEqual(len(aggregated), count)