Update for failing tests

btrdb/stream.py

@@ -18,6 +18,7 @@
 import re
 import json
 import uuid as uuidlib
+import warnings
 from copy import deepcopy
 from collections.abc import Sequence
 
@@ -841,7 +842,10 @@ class StreamSetBase(Sequence):
 
     def __init__(self, streams):
         self._streams = streams
-        self._btrdb = self._streams[0]._btrdb
+        try:
+            self._btrdb = self._streams[0]._btrdb
+        except Exception as e:
+            warnings.warn(f"Issue setting btrdb object from stream: {e}")
         self._pinned_versions = None
 
         self.filters = []
@@ -967,7 +971,7 @@ def earliest(self):
             lambda s: s.nearest(start, version=versions.get(s.uuid, 0), backward=False),
             self._streams,
         )
-        for point in earliest_points_gen:
+        for point, _ in earliest_points_gen:
             earliest.append(point)
 
         return tuple(earliest)
@@ -994,7 +998,7 @@ def latest(self):
             lambda s: s.nearest(start, version=versions.get(s.uuid, 0), backward=True),
             self._streams,
         )
-        for point in latest_points_gen:
+        for point, _ in latest_points_gen:
             latest.append(point)
 
         return tuple(latest)

btrdb/transformers.py

@@ -188,8 +188,9 @@ def to_array(streamset, agg="mean"):
                 segment.append(point.value)
             else:
                 segment.append(getattr(point, agg))
-        results.append(segment)
-    return np.array(results)
+        results.append(np.array(segment))
+    print(results)
+    return np.array(results, dtype=object)
 
 
 def to_dict(streamset, agg="mean", name_callable=None):

tests/btrdb/test_stream.py

@@ -10,7 +10,7 @@
 """
 Testing package for the btrdb stream module
 """
-
+import concurrent.futures
 ##########################################################################
 ## Imports
 ##########################################################################
@@ -58,6 +58,8 @@ def stream1():
     type(stream).name = PropertyMock(return_value="gala")
     stream.tags = Mock(return_value={"name": "gala", "unit": "volts"})
     stream.annotations = Mock(return_value=({"owner": "ABC", "color": "red"}, 11))
+    stream._btrdb = Mock()
+    stream._btrdb._executor = concurrent.futures.ThreadPoolExecutor()
     return stream
 
 
@@ -72,6 +74,8 @@ def stream2():
     type(stream).name = PropertyMock(return_value="blood")
     stream.tags = Mock(return_value={"name": "blood", "unit": "amps"})
     stream.annotations = Mock(return_value=({"owner": "ABC", "color": "orange"}, 22))
+    stream._btrdb = Mock()
+    stream._btrdb._executor = Mock()
     return stream
 
 
@@ -1181,7 +1185,7 @@ def test_earliest(self, stream1, stream2):
         Assert earliest returns correct time code
         """
         streams = StreamSet([stream1, stream2])
-        assert streams.latest() == (RawPoint(time=10, value=1), RawPoint(time=20, value=1))
+        assert streams.earliest() == (RawPoint(time=10, value=1), RawPoint(time=20, value=1))
 
 
     def test_latest(self, stream1, stream2):