upgraded to newer tiled substription api

pyproject.toml

@@ -6,6 +6,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "arroyoSAS"
 version = "0.1.0"
+requires-python = ">=3.11,<3.12"
 description = "A package to perform computations and suggestions during a time-resolved GISAXS experiment."
 authors = [{ name = "Dylan McReynolds", email = "dmcreynolds@lbl.gov" }]
 dependencies = [
@@ -23,7 +24,7 @@ dependencies = [
     "websockets",
     "zarr",
     "transformers",
-    "tiled[client]",
+    "tiled[all]",
     "redis"]
 
 [project.optional-dependencies]

src/_test/test_tiled_websocket_bluesky.py

@@ -8,6 +8,7 @@
 import numpy as np
 import pytest
 
+from arroyosas.schemas import RawFrameEvent, SASStart
 from arroyosas.tiled.tiled_websocket_bluesky import TiledClientListener, tiled_ws_listener_factory
 
 
@@ -85,110 +86,141 @@ def test_log_message_to_json_no_current_dir(self, listener):
         # No file created, no error
 
     def test_on_new_run_creates_subscription(self, listener, mock_tiled_client):
-        data = {"key": "run_uid_123"}
-        with (
-            patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription") as mock_sub_cls,
-            patch.object(listener, "publish_start"),
-        ):
-            mock_sub = MagicMock()
-            mock_sub_cls.return_value = mock_sub
-            listener.on_new_run(MagicMock(), data)
-
-            mock_sub_cls.assert_called_once()
-            mock_sub.add_callback.assert_called_once_with(listener.on_streams_namespace)
-            mock_sub.start.assert_called_once()
+        update = MagicMock()
+        update.key = "run_uid_123"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "run_uid_123"}
+
+        run_node = MagicMock()
+        run_sub = MagicMock()
+        run_sub.child_created = MagicMock()
+        run_node.subscribe.return_value = run_sub
+        update.child.return_value = run_node
+
+        with patch.object(listener, "publish_start") as mock_publish_start:
+            listener.on_new_run(update)
+
+        run_node.subscribe.assert_called_once_with(start=0)
+        run_sub.child_created.add_callback.assert_called_once_with(listener.on_streams_namespace)
+        run_sub.start.assert_called_once()
+        mock_publish_start.assert_called_once_with(run_node, {"key": "run_uid_123"})
 
     def test_on_new_run_calls_publish_start(self, listener):
-        data = {"key": "run_001"}
-        with (
-            patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription") as mock_sub_cls,
-            patch.object(listener, "publish_start") as mock_pub,
-        ):
-            mock_sub_cls.return_value = MagicMock()
-            listener.on_new_run(MagicMock(), data)
-            mock_pub.assert_called_once_with(data)
+        update = MagicMock()
+        update.key = "run_001"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "run_001"}
+
+        run_node = MagicMock()
+        run_sub = MagicMock()
+        run_sub.child_created = MagicMock()
+        run_node.subscribe.return_value = run_sub
+        update.child.return_value = run_node
+
+        with patch.object(listener, "publish_start") as mock_pub:
+            listener.on_new_run(update)
+
+        mock_pub.assert_called_once_with(run_node, {"key": "run_001"})
 
     def test_on_streams_namespace(self, listener, mock_tiled_client):
-        sub = MagicMock()
-        sub.segments = ["run_uid_123"]
-        with patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription") as mock_sub_cls:
-            mock_sub = MagicMock()
-            mock_sub_cls.return_value = mock_sub
-            listener.on_streams_namespace(sub, {"key": "streams"})
-            mock_sub.add_callback.assert_called_once_with(listener.on_new_stream)
+        update = MagicMock()
+        update.key = "streams"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "streams"}
+
+        streams_node = MagicMock()
+        streams_sub = MagicMock()
+        streams_sub.child_created = MagicMock()
+        streams_node.subscribe.return_value = streams_sub
+        update.child.return_value = streams_node
+
+        listener.on_streams_namespace(update)
+
+        streams_node.subscribe.assert_called_once_with(start=0)
+        streams_sub.child_created.add_callback.assert_called_once_with(listener.on_new_stream)
 
     def test_on_new_stream(self, listener):
-        sub = MagicMock()
-        sub.segments = ["run_uid", "streams"]
-        with patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription") as mock_sub_cls:
-            mock_sub = MagicMock()
-            mock_sub_cls.return_value = mock_sub
-            listener.on_new_stream(sub, {"key": "primary"})
-            mock_sub.add_callback.assert_called_once_with(listener.on_node_in_stream)
+        update = MagicMock()
+        update.key = "primary"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "primary"}
+
+        stream_node = MagicMock()
+        stream_sub = MagicMock()
+        stream_sub.child_created = MagicMock()
+        stream_node.subscribe.return_value = stream_sub
+        update.child.return_value = stream_node
+
+        listener.on_new_stream(update)
+
+        stream_node.subscribe.assert_called_once_with(start=0)
+        stream_sub.child_created.add_callback.assert_called_once_with(listener.on_node_in_stream)
 
     def test_on_node_in_stream_matching_target(self, listener):
-        sub = MagicMock()
-        sub.segments = ["run_uid", "streams", "primary"]
-        data = {"key": "img", "sequence": 0}
-
-        with (
-            patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription") as mock_sub_cls,
-            patch.object(listener, "publish_event") as mock_pub,
-        ):
-            mock_sub = MagicMock()
-            mock_sub_cls.return_value = mock_sub
-            listener.on_node_in_stream(sub, data)
-            mock_pub.assert_called_once_with(sub, data)  # ← was publish_event(data)
+        update = MagicMock()
+        update.key = "img"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "img", "sequence": 0}
+
+        data_node = MagicMock()
+        data_sub = MagicMock()
+        data_sub.new_data = MagicMock()
+        data_node.subscribe.return_value = data_sub
+        update.child.return_value = data_node
+
+        listener.on_node_in_stream(update)
+
+        data_node.subscribe.assert_called_once_with(start=0)
+        data_sub.new_data.add_callback.assert_called_once_with(listener.on_event)
+        data_sub.start.assert_called_once()
 
     def test_on_node_in_stream_non_matching_target(self, listener):
-        sub = MagicMock()
-        sub.segments = ["run_uid", "streams", "primary"]
-        data = {"key": "other_key", "sequence": 0}
+        update = MagicMock()
+        update.key = "other_key"
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"key": "other_key", "sequence": 0}
+
+        listener.on_node_in_stream(update)
 
-        with (
-            patch("arroyosas.tiled.tiled_websocket_bluesky.Subscription"),
-            patch.object(listener, "publish_event") as mock_pub,
-        ):
-            listener.on_node_in_stream(sub, data)
-            mock_pub.assert_not_called()
+        update.child.assert_not_called()
 
     def test_on_event(self, listener):
-        sub = MagicMock()
-        data = {"key": "event_1", "sequence": 5}
+        update = MagicMock()
+        update.subscription = MagicMock()
+        update.model_dump.return_value = {"sequence": 5}
         listener.create_run_folder("run_on_event")
-        listener.on_event(sub, data)
+        with patch.object(listener, "publish_event") as mock_publish_event:
+            listener.on_event(update)
+
         # With create_run_logs=True, should log to JSON
         log_files = list(Path(listener.current_run_dir).glob("on_event_*.json"))
         assert len(log_files) == 1
+        mock_publish_event.assert_called_once_with(update)
 
     def test_publish_start(self, listener):
-        # publish_start calls SASStart(data=data) which fails validation (missing fields)
-        # The exception propagates from send_to_operator -> asyncio.run
         data = {"key": "run_001"}
-        with patch.object(listener, "send_to_operator"):
-            # The actual call will raise ValidationError; just verify send_to_operator is called
-            # (it's called even if message construction fails inside send_to_operator)
-            try:
-                listener.publish_start(data)
-            except Exception:
-                pass
-            # send_to_operator is called with whatever SASStart produces (or raises)
-            # The important thing is publish_start calls send_to_operator
-
-    def test_publish_event(self, listener, mock_tiled_client):  # ← add mock_tiled_client
-        sub = MagicMock()
-        sub.segments = ["run_uid", "streams", "primary"]
-        data = {"key": "img", "sequence": 7}
-
-        data_node = MagicMock()
-        data_node.__getitem__ = MagicMock(return_value=np.zeros((5, 5)))
-        mock_tiled_client.__getitem__ = MagicMock(return_value=data_node)
+        run_node = MagicMock()
+        run_node.metadata = {"run_name": "Run 1", "width": 5, "height": 7, "data_type": "uint16"}
 
         with patch.object(listener, "send_to_operator") as mock_send:
-            listener.publish_event(sub, data)  # ← pass sub
+            listener.publish_start(run_node, data)
             mock_send.assert_called_once()
-            from arroyosas.schemas import RawFrameEvent
+            msg = mock_send.call_args[0][0]
+            assert isinstance(msg, SASStart)
+            assert msg.run_id == "run_001"
+            assert msg.run_name == "Run 1"
+            assert msg.width == 5
+            assert msg.height == 7
+            assert msg.data_type == "uint16"
+
+    def test_publish_event(self, listener):
+        update = MagicMock()
+        update.data.return_value = np.zeros((5, 5))
+        update.sequence = 7
 
+        with patch.object(listener, "send_to_operator") as mock_send:
+            listener.publish_event(update)
+            mock_send.assert_called_once()
             msg = mock_send.call_args[0][0]
             assert isinstance(msg, RawFrameEvent)
             assert msg.frame_number == 7