Skip to content

Commit 363232e

Browse files
Stack, StephenStack, Stephen
Stack, Stephen
authored and
Stack, Stephen
committed
Merge pull request digidotcom#14 in PYTHON/python-devicecloud from stream-docs to master
* commit '21c97400e26feb37ea66fbc227249f759d7be2ca': Streams/Docs: Fix some oddities in roll-up responses from DC and add cookbook recipe.
2 parents eb8d526 + 21c9740 commit 363232e

7 files changed

Lines changed: 270 additions & 68 deletions

File tree

‎.gitignore‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,9 @@
1414
# Virtualenv
1515
#-------------------------------------------------------------------------------
1616
env/
17+
env3/
18+
.env/
19+
.env3/
1720
.toxenv/
1821

1922
#-------------------------------------------------------------------------------
@@ -66,7 +69,6 @@ build/
6669
# Temporary files
6770
#-------------------------------------------------------------------------------
6871
tmp/
69-
.env/
7072

7173
#-------------------------------------------------------------------------------
7274
# Generated Files

‎devicecloud/examples/classroom_stream.py‎

Lines changed: 0 additions & 54 deletions
This file was deleted.

‎devicecloud/examples/cookbook_streams.py‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
from devicecloud.streams import STREAM_TYPE_STRING, DataPoint, STREAM_TYPE_INTEGER
2+
from devicecloud import DeviceCloud
3+
from getpass import getpass
4+
import datetime
5+
import pprint
6+
import random
7+
import time
8+
import json
9+
10+
11+
def get_authenticated_dc():
12+
while True:
13+
user = raw_input("username: ")
14+
password = getpass("password: ")
15+
dc = DeviceCloud(user, password)
16+
if dc.has_valid_credentials():
17+
print ("Credentials accepted!")
18+
return dc
19+
else:
20+
print ("Invalid username or password provided, try again")
21+
22+
23+
def get_or_create_classroom(datatype):
24+
dc = get_authenticated_dc()
25+
classroom = dc.streams.get_stream_if_exists('classroom')
26+
if not classroom:
27+
classroom = dc.streams.create_stream(
28+
stream_id='classroom',
29+
data_type=datatype,
30+
description='Stream representing a classroom of students',
31+
)
32+
return classroom
33+
34+
35+
def fill_classroom_with_student_ids(classroom):
36+
# fake data with wide range of timestamps
37+
now = time.time()
38+
one_day_in_seconds = 86400
39+
40+
datapoints = list()
41+
for student_id in xrange(100):
42+
deviation = random.randint(0, one_day_in_seconds)
43+
random_time = now + deviation
44+
datapoint = DataPoint(data=student_id,
45+
timestamp=datetime.datetime.fromtimestamp(random_time),
46+
data_type=STREAM_TYPE_INTEGER)
47+
datapoints.append(datapoint)
48+
49+
classroom.bulk_write_datapoints(datapoints)
50+
51+
52+
def example_1():
53+
classroom = get_or_create_classroom(STREAM_TYPE_STRING)
54+
55+
student = {
56+
'name': 'Bob',
57+
'student_id': 12,
58+
'age': 21,
59+
}
60+
datapoint = DataPoint(data=json.dumps(student))
61+
classroom.write(datapoint)
62+
63+
students = [
64+
{
65+
'name': 'James',
66+
'student_id': 13,
67+
'age': 22,
68+
},
69+
{
70+
'name': 'Henry',
71+
'student_id': 14,
72+
'age': 20,
73+
}
74+
]
75+
datapoints = [DataPoint(data=json.dumps(x)) for x in students]
76+
classroom.bulk_write_datapoints(datapoints)
77+
78+
most_recent_dp = classroom.get_current_value()
79+
print json.loads(most_recent_dp.get_data())['name']
80+
81+
82+
def example_2():
83+
# assume `fill_classroom_with_random_data()` has already been called
84+
85+
classroom = get_or_create_classroom(STREAM_TYPE_INTEGER)
86+
rollup_data = classroom.read(rollup_interval='hour', rollup_method='count')
87+
hourly_data = {}
88+
for dp in rollup_data:
89+
hourly_data[dp.get_timestamp().hour] = dp.get_data()
90+
pprint.pprint(hourly_data)
91+
92+
93+
example_2()
94+
print 'done.'

‎devicecloud/streams.py‎

Lines changed: 51 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,8 @@
1313
import six
1414
from devicecloud.apibase import APIBase
1515
from devicecloud import DeviceCloudException, DeviceCloudHttpException
16-
from devicecloud.util import conditional_write, to_none_or_dt, validate_type, isoformat
16+
from devicecloud.util import conditional_write, to_none_or_dt, validate_type, isoformat, \
17+
dc_utc_timestamp_to_dt
1718
from six import StringIO
1819

1920

@@ -69,6 +70,10 @@ class NoSuchStreamException(StreamException):
6970
"""Failure to find a stream based on a given id"""
7071

7172

73+
class InvalidRollupDatatype(StreamException):
74+
"""Roll-up's are only valid on numerical data types"""
75+
76+
7277
class StreamsAPI(APIBase):
7378
"""Provide interface for interacting with device cloud streams API
7479
@@ -278,6 +283,29 @@ def from_json(cls, stream, json_data):
278283
dp_id=json_data.get("id"),
279284
)
280285

286+
@classmethod
287+
def from_rollup_json(cls, stream, json_data):
288+
"""Rollup json data from the server looks slightly different
289+
290+
:param DataStream stream: The :class:`~DataStream` out of which this data is coming
291+
:param dict json_data: Deserialized JSON data from the device cloud about this device
292+
:raises ValueError: if the data is malformed
293+
:return: (:class:`~DataPoint`) newly created :class:`~DataPoint`
294+
"""
295+
dp = cls.from_json(stream, json_data)
296+
297+
# Special handling for timestamp
298+
timestamp = isoformat(dc_utc_timestamp_to_dt(int(json_data.get("timestamp"))))
299+
300+
# Special handling for data, all rollup data is float type
301+
type_converter = DSTREAM_TYPE_MAP[dp.get_data_type()]
302+
data = type_converter[0](float(json_data.get("data")))
303+
304+
# Update the special fields
305+
dp.set_timestamp(timestamp)
306+
dp.set_data(data)
307+
return dp
308+
281309
def __init__(self, data, stream_id=None, description=None, timestamp=None,
282310
quality=None, location=None, data_type=None, units=None, dp_id=None,
283311
customer_id=None, server_timestamp=None):
@@ -831,7 +859,7 @@ def read(self, start_time=None, end_time=None, use_client_timeline=True, newest_
831859
the result set.
832860
833861
:param start_time: The start time for the window of data points to read. None means
834-
that we should start with the old data available.
862+
that we should start with the oldest data available.
835863
:type start_time: :class:`datetime.datetime` or None
836864
:param end_time: The end time for the window of data points to read. None means
837865
that we should include all points received until this point in time.
@@ -868,6 +896,23 @@ def read(self, start_time=None, end_time=None, use_client_timeline=True, newest_
868896
:returns: A generator object which one can iterate over the DataPoints read.
869897
870898
"""
899+
900+
is_rollup = False
901+
if (rollup_interval is not None) or (rollup_method is not None):
902+
is_rollup = True
903+
numeric_types = [
904+
STREAM_TYPE_INTEGER,
905+
STREAM_TYPE_LONG,
906+
STREAM_TYPE_FLOAT,
907+
STREAM_TYPE_DOUBLE,
908+
STREAM_TYPE_STRING,
909+
STREAM_TYPE_BINARY,
910+
STREAM_TYPE_UNKNOWN,
911+
]
912+
913+
if self.get_data_type(use_cached=True) not in numeric_types:
914+
raise InvalidRollupDatatype('Rollups only support numerical DataPoints')
915+
871916
# Validate function inputs
872917
start_time = to_none_or_dt(validate_type(start_time, datetime.datetime, type(None)))
873918
end_time = to_none_or_dt(validate_type(end_time, datetime.datetime, type(None)))
@@ -929,5 +974,8 @@ def read(self, start_time=None, end_time=None, use_client_timeline=True, newest_
929974
result_size = int(result["resultSize"]) # how many are actually included here?
930975
query_parameters["pageCursor"] = result.get("pageCursor") # will not be present if result set is empty
931976
for item_info in result.get("items", []):
932-
data_point = DataPoint.from_json(self, item_info)
977+
if is_rollup:
978+
data_point = DataPoint.from_rollup_json(self, item_info)
979+
else:
980+
data_point = DataPoint.from_json(self, item_info)
933981
yield data_point

‎devicecloud/test/test_streams.py‎

Lines changed: 32 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,6 @@
1515
from devicecloud.test.test_utilities import HttpTestBase
1616
from devicecloud import DeviceCloudHttpException
1717

18-
19-
2018
# Example HTTP Responses
2119
import httpretty
2220
import six
@@ -658,9 +656,11 @@ def test_rollup_interval_half(self):
658656
self.prepare_response("GET", "/ws/DataPoint/test", GET_DATA_POINTS_ONE)
659657
test_stream = self.dc.streams.get_stream("test")
660658
points = list(test_stream.read(rollup_interval=ROLLUP_INTERVAL_HALF))
661-
self.assertEqual(httpretty.httpretty.latest_requests[-2].querystring["rollupInterval"][0], "half")
659+
self.assertEqual(httpretty.httpretty.latest_requests[-1].querystring["rollupInterval"][0], "half")
662660

663661
def test_rollup_interval_invalid(self):
662+
self.prepare_response("GET", "/ws/DataStream/test", GET_TEST_DATA_STREAM)
663+
self.prepare_response("GET", "/ws/DataPoint/test", GET_DATA_POINTS_ONE)
664664
test_stream = self.dc.streams.get_stream("test")
665665
self.assertRaises(ValueError, six.next, test_stream.read(rollup_interval='invalid'))
666666

@@ -669,9 +669,10 @@ def test_rollup_method_count(self):
669669
self.prepare_response("GET", "/ws/DataPoint/test", GET_DATA_POINTS_ONE)
670670
test_stream = self.dc.streams.get_stream("test")
671671
points = list(test_stream.read(rollup_method=ROLLUP_METHOD_COUNT))
672-
self.assertEqual(httpretty.httpretty.latest_requests[-2].querystring["rollupMethod"][0], "count")
672+
self.assertEqual(httpretty.httpretty.latest_requests[-1].querystring["rollupMethod"][0], "count")
673673

674674
def test_rollup_method_invalid(self):
675+
self.prepare_response("GET", "/ws/DataStream/test", GET_TEST_DATA_STREAM)
675676
test_stream = self.dc.streams.get_stream("test")
676677
self.assertRaises(ValueError, six.next, test_stream.read(rollup_method='invalid'))
677678

@@ -754,6 +755,33 @@ def test_repr(self):
754755
dp = stream.get_current_value()
755756
repr(dp)
756757

758+
def test_rollup_datapoint(self):
759+
self.prepare_response("GET", "/ws/DataStream/test", GET_TEST_DATA_STREAM)
760+
example_json = {
761+
"id": "07d77854-0557-11e4-ab44-fa163e7ebc6b",
762+
"timestamp": "1404683207981",
763+
"timestampISO": "2014-07-06T21:46:47.981Z",
764+
"serverTimestamp": "1404683207981",
765+
"serverTimestampISO": "2014-07-06T21:46:47.981Z",
766+
"data": "0.0",
767+
"description": "Test",
768+
"quality": "20",
769+
"location": "1.0,2.0,3.0"
770+
}
771+
stream = self._get_stream("test", with_cached_data=True)
772+
dp = DataPoint.from_rollup_json(stream, example_json)
773+
self.assertEqual(dp.get_data(), 0.0)
774+
orig_dt = dp.get_timestamp()
775+
dt_wo_ms = datetime.datetime(year=orig_dt.year,
776+
month=orig_dt.month,
777+
day=orig_dt.day,
778+
hour=orig_dt.hour,
779+
minute=orig_dt.minute,
780+
second=orig_dt.second,
781+
tzinfo=orig_dt.tzinfo)
782+
self.assertEqual(six.b(dt_wo_ms.isoformat()),
783+
six.b('2014-07-06T21:46:47+00:00'))
784+
757785

758786
if __name__ == "__main__":
759787
unittest.main()

‎devicecloud/util.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ def validate_type(input, *types):
6969
raise TypeError("Input expected to one of following types: %s" % (types, ))
7070
return input
7171

72+
7273
def isoformat(dt):
7374
"""Return an ISO-8601 formatted string from the provided datetime object"""
7475
if not isinstance(dt, datetime.datetime):
@@ -78,3 +79,8 @@ def isoformat(dt):
7879
raise ValueError("naive datetime objects are not allowed beyond the library boundaries")
7980

8081
return dt.isoformat().replace("+00:00", "Z") # nicer to look at
82+
83+
84+
def dc_utc_timestamp_to_dt(dc_timestamp_in_milleseconds):
85+
"""Return a UTC datetime object"""
86+
return arrow.Arrow.utcfromtimestamp(dc_timestamp_in_milleseconds / 1000).datetime

0 commit comments

Comments
 (0)