Merge pull request #30 from PredixDev/develop_tsfeedback

Account for time series attributes that may be numbers

Author: j12y

predix/data/timeseries.py

@@ -219,7 +219,7 @@ def get_datapoints(self, tags, start=None, end=None, order=None,
             - order: ascending (asc) or descending (desc)
             - limit: only return a few values (ie. 25)
             - qualities: data quality value (ie. [ts.GOOD, ts.UNCERTAIN])
-            - attributes: data attributes (ie. {'unit': 'mph'})
+            - attributes: dictionary of key-values (ie. {'unit': 'mph'})
             - measurement: tuple of operation and value (ie. ('gt', 30))
             - aggregations: summary statistics on data results (ie. 'avg')
             - post: POST query instead of GET (caching implication)
@@ -241,7 +241,7 @@ def get_datapoints(self, tags, start=None, end=None, order=None,
         # seems to be required all the time, so using sensible default.
         if not start:
             start = '1w-ago'
-            logging.info("Defaulting to timeseries data since %s" % (start))
+            logging.warning("Defaulting query for data with start date %s" % (start))
 
         # Start date can be absolute or relative, only certain legal values
         # but service will throw error if used improperly.  (ms, s, mi, h, d,
@@ -352,7 +352,12 @@ def get_latest(self, tags, post=False):
         and more complex query.
         """
         params = {}
-        params['tags'] = tags
+        if isinstance(tags, list):
+            params['tags'] = str.join(',', tags)
+        elif isinstance(tags, str):
+            params['tags'] = tags
+        else:
+            raise ValueError("Expect to get tags as a str or list")
 
         if post:
             return self._post_latest(params)
@@ -459,17 +464,24 @@ def queue(self, name, value, quality=None, timestamp=None,
             "datapoints": [[timestamp, value, quality]]
         }
 
-        # Attributes are specified for datapoint
-        if attributes:
+        # Attributes are extra details for a datapoint
+
+        if attributes is not None:
+            if not isinstance(attributes, dict):
+                raise ValueError("Attributes are expected to be a dictionary.")
+
             # Validate rules for attribute keys to provide guidance.
             invalid_value = ':;= '
             has_invalid_value = re.compile(r'[%s]' % (invalid_value)).search
             has_valid_key = re.compile(r'^[\w\.\/\-]+$').search
 
             for (key, val) in list(attributes.items()):
-                # Values cannot be NULL
-                if not val:
-                    raise ValueError("Attribute (%s) must have value." % (key))
+                # Values cannot be empty
+                if (val == '') or (val is None):
+                    raise ValueError("Attribute (%s) must have a non-empty value." % (key))
+
+                # Values should be treated as a string for regex validation
+                val = str(val)
 
                 # Values cannot contain certain arbitrary characters
                 if bool(has_invalid_value(val)):

test/integration/test_timeseries.py

@@ -14,7 +14,6 @@ class TestTimeSeries(unittest.TestCase):
     More Testing
     - [ ] Query on multiple tags
     - [ ] Get aggregations
-    - [ ] Get latest
     - [ ] Get values
     """
 
@@ -86,6 +85,16 @@ def test_ingest_attributes(self):
         self.assertEqual(values[0][2], 1)
         self.assertEqual(attrs, {'units':['F']})
 
+        # Test multiple attributes, including a number
+        ts.send('TAG4b', 61, attributes={'units': 'F', 'Number': 1})
+        time.sleep(2) # Allow time for ingestion to complete
+        res = ts.get_datapoints('TAG4b')
+        self.assertEqual('TAG4b', res['tags'][0]['name'])
+
+        # Test we get error if attributes is not a dictionary
+        with self.assertRaises(ValueError):
+            ts.send('TAG4b', 62, attributes="{'units': 'F'}")
+
     def test_ingest_timestamp(self):
         ts = self.app.get_timeseries()
 
@@ -102,6 +111,12 @@ def test_ingest_timestamp(self):
         values = res['tags'][0]['results'][0]['values']
         self.assertEqual(values[0][0], 306658800000)
 
+        # Verify can search with epoch dates, not just relative dates
+
+        res = ts.get_datapoints('TAG5', start=292185600000, end=313267200000)
+        values = res['tags'][0]['results'][0]['values']
+        self.assertEqual(values[0][1], 70)
+
     def test_query_filter_aggregation(self):
         ts = self.app.get_timeseries()
         tag = 'STACK1'
@@ -115,6 +130,15 @@ def test_query_filter_aggregation(self):
         result = query['tags'][0]['results'][0]['values'][0][1]
         self.assertEqual(result, sum(values[1:]))
 
+    def test_get_latest(self):
+        ts = self.app.get_timeseries()
+        ts.send('LATEST-1', 1)
+        ts.send('LATEST-2', 2)
+        time.sleep(2) # Allow time for ingestion
+
+        res = ts.get_latest(['LATEST-1', 'LATEST-2'])
+        self.assertEqual(len(res['tags']), 2)
+        self.assertEqual(res['tags'][0]['name'], 'LATEST-1')
 
 if __name__ == '__main__':
     if os.getenv('DEBUG'):