add new arroyo listener

docker-compose.yml

@@ -31,6 +31,18 @@ services:
     networks:
       gisax_net1:
 
+  lse_operator:
+    command: python -m arroyogisaxs.app.kse_operator_cli
+    build:
+      context: .
+    restart: unless-stopped
+    ports:
+      - "8021:8021"
+    volumes:
+      - .:/app:Z
+      - ./settings.yaml:/app/settings.yaml:Z
+    networks:
+      gisax_net1:
   viz_operator:
     command: python -m arroyosas.app.viz_operator_cli
     build:
@@ -79,6 +91,8 @@ services:
       - ./persist/kvrocks/data:/data
     networks:
       gisax_net1:
+    ports:
+      - "127.0.0.1:6666:6666"
 
 networks:
   gisax_net1:

notebooks/query.ipynb

@@ -24,31 +24,16 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "scans = c[\"smi\"][\"raw\"].search(Key(\"num_points\") > 9).search(Contains(\"detectors\", \"pil900KW\")).values().tail(10)"
+    "# scans = c[\"smi\"][\"raw\"].search(Key(\"num_points\") > 9).search(Contains(\"detectors\", \"pil900KW\")).values().tail(10)\n",
+    "\n",
+    "scans = c[\"smi\"][\"raw\"].search(Key(\"scan_id\") == \"844179\")"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 5,
+   "execution_count": 3,
    "metadata": {},
-   "outputs": [
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/8a0b8a58-8c26-4e35-8237-38b1b8308816     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/8a0b8a58-8c26-4e35-8237-38b1b8308816/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/227d58a4-0150-4116-aa56-ba1acdde8aaf     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/227d58a4-0150-4116-aa56-ba1acdde8aaf/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/b3cee03f-d59c-4df4-89df-6ff49dd1961f     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/b3cee03f-d59c-4df4-89df-6ff49dd1961f/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/1d4b7707-4dcd-41b7-b979-f9ebcef16a38     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/1d4b7707-4dcd-41b7-b979-f9ebcef16a38/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/982609e5-c0c4-4b25-bfd2-74b17552a76a     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/982609e5-c0c4-4b25-bfd2-74b17552a76a/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/4eae4d6a-030c-4b1b-886c-aac0a705ea63     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/4eae4d6a-030c-4b1b-886c-aac0a705ea63/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/d1dc32ee-1685-473d-bf9f-5100b6af8df2     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/d1dc32ee-1685-473d-bf9f-5100b6af8df2/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/8a18c06b-0f7e-4a98-9b41-1fa3fe521a70     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/8a18c06b-0f7e-4a98-9b41-1fa3fe521a70/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/ee5a2b25-b2f9-4966-a61c-e932f2401c4c     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/ee5a2b25-b2f9-4966-a61c-e932f2401c4c/primary/data/pil900KW_image\n",
-      "https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/6050f8bf-01b8-4a9b-9ad2-fb8f41a35701     https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/6050f8bf-01b8-4a9b-9ad2-fb8f41a35701/primary/data/pil900KW_image\n"
-     ]
-    }
-   ],
+   "outputs": [],
    "source": [
     "for scan in scans:\n",
     "    # print(f\"{scan.metadata['start']['uid']}  {scan.metadata['stop']['num_events']}\")\n",
@@ -115,7 +100,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "tiled",
+   "display_name": "smi",
    "language": "python",
    "name": "python3"
   },
@@ -129,7 +114,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.10"
+   "version": "3.11.11"
   }
  },
  "nbformat": 4,

notebooks/tiled.ipynb

@@ -2,20 +2,9 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": 3,
    "metadata": {},
-   "outputs": [
-    {
-     "data": {
-      "text/plain": [
-       "(600,)"
-      ]
-     },
-     "execution_count": 1,
-     "metadata": {},
-     "output_type": "execute_result"
-    }
-   ],
+   "outputs": [],
    "source": [
     "import os\n",
     "\n",
@@ -25,9 +14,9 @@
     "dotenv.load_dotenv()\n",
     "c = from_uri(\"https://tiled.nsls2.bnl.gov\", api_key=os.getenv('TILED_LIVE_API_KEY'))\n",
     "smi = c['smi']\n",
-    "scan = smi['raw']['c21d472b-4241-4f48-8d62-a0b56e1d471d']\n",
-    "# scan = smi['raw'][820000]\n",
-    "scan['primary']['data']['time'].shape"
+    "# scan = smi['raw']['c21d472b-4241-4f48-8d62-a0b56e1d471d']\n",
+    "# # scan = smi['raw'][820000]\n",
+    "# scan['primary']['data']['time'].shape"
    ]
   },
   {
@@ -220,7 +209,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "gisaxs",
+   "display_name": "smi",
    "language": "python",
    "name": "python3"
   },

pyproject.toml

@@ -55,11 +55,13 @@ lse = [
 
 all = [
     "arroyopy",
+    "bluesky",
     "dynaconf",
     "python-dotenv",
     "pandas",
     "msgpack",
     "numpy",
+    "ophyd",
     "Pillow",
     "pyzmq",
     "tqdm",
@@ -71,7 +73,7 @@ all = [
     "pandas",
     "python-dotenv",
     "pyarrow>=14.0.1",
-    
+
 ]
 
 [tool.isort]

scripts/tiled-ws/consumer.py

@@ -0,0 +1,47 @@
+from tiled.client import from_uri
+from tiled.client.stream import Subscription
+
+c = from_uri("https://tiled-dev.nsls2.bnl.gov")
+target = "pil2M_image"
+
+
+def on_new_run(sub, data):
+    uid = data["key"]
+    print(f"New run {uid}")
+    run_sub = Subscription(c.context, [uid], start=0)
+    run_sub.add_callback(on_streams_namespace)
+    run_sub.start()
+
+
+def on_streams_namespace(sub, data):
+    streams_sub = Subscription(c.context, sub.segments + ["streams"], start=0)
+    streams_sub.add_callback(on_new_stream)
+    streams_sub.start()
+
+
+def on_new_stream(sub, data):
+    stream_name = data["key"]
+    print(f"new stream {stream_name}")
+    stream_sub = Subscription(c.context, sub.segments + [stream_name], start=0)
+    stream_sub.add_callback(on_node_in_stream)
+    stream_sub.start()
+
+
+def on_node_in_stream(sub, data):
+    key = data["key"]
+    if key != target:
+        return
+    stream_sub = Subscription(c.context, sub.segments + [key], start=0)
+    stream_sub.add_callback(print)
+    stream_sub.start()
+
+
+catalog_sub = Subscription(c.context)
+catalog_sub.add_callback(on_new_run)
+catalog_sub.start()
+
+if __name__ == "__main__":
+    import time
+
+    while True:
+        time.sleep(1)

scripts/tiled-ws/producer.py

@@ -0,0 +1,16 @@
+import time
+
+from bluesky import RunEngine
+from bluesky.callbacks.tiled_writer import TiledWriter
+from bluesky.plans import count
+from ophyd.sim import img
+from tiled.client import from_uri
+
+RE = RunEngine()
+client = from_uri("http://localhost:8000", api_key="secret")
+tw = TiledWriter(client, batch_size=1)
+RE.subscribe(tw)
+
+while True:
+    RE(count([img], 10, delay=5))
+    time.sleep(5)

scripts/tiled-ws/tiled_ws.ipynb

@@ -0,0 +1,103 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "id": "ad9b72db",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "NOTICE TO USERS\n",
+      "\n",
+      "This is a Federal computer system (and/or it is directly connected to a BNL local network system) and is the property of the United States Government. It is for authorized use only. Users (authorized or unauthorized) have no explicit or implicit expectation of privacy.\n",
+      "\n",
+      "Any or all uses of this system and all files on this system may be intercepted, monitored, recorded, copied, audited, inspected, and disclosed to authorized site, Department of Energy, and law enforcement personnel, as well as authorized officials of other agencies, both domestic and foreign. By using this system, the user consents to such interception, monitoring, recording, copying, auditing, inspection, and disclosure at the discretion of authorized site or Department of Energy personnel.\n",
+      "\n",
+      "Unauthorized or improper use of this system may result in administrative disciplinary action and civil and criminal penalties. By continuing to use this system you indicate your awareness of and consent to these terms and conditions of use. LOG OFF IMMEDIATELY if you do not agree to the conditions stated in this warning.\n",
+      "\n",
+      "You have logged in as dmcreynol.\n",
+      "\n"
+     ]
+    }
+   ],
+   "source": [
+    "from tiled.client import from_uri\n",
+    "\n",
+    "c = from_uri(\"https://tiled-dev.nsls2.bnl.gov\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 7,
+   "id": "d717036d",
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "NameError",
+     "evalue": "name 'client' is not defined",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
+      "\u001b[31mNameError\u001b[39m                                 Traceback (most recent call last)",
+      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[7]\u001b[39m\u001b[32m, line 1\u001b[39m\n\u001b[32m----> \u001b[39m\u001b[32m1\u001b[39m \u001b[43mclient\u001b[49m\n",
+      "\u001b[31mNameError\u001b[39m: name 'client' is not defined"
+     ]
+    }
+   ],
+   "source": [
+    "client"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "051c3811",
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "<Container {'smi'}>"
+      ]
+     },
+     "execution_count": 8,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": []
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "d461b7d4",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "tiled_ws",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

settings.yaml

@@ -8,15 +8,16 @@ tiled_poller:
   poll_interval: 5 # seconds
   uri: https://tiled.nsls2.bnl.gov
   api_key: "@format {env[TILED_LIVE_API_KEY]}"
+  websocket_url: "ws://tiled:8000/stream"  # change it
   runs_segments:
     - smi
     - raw
   frames_segments:
-    - primary 
+    - primary
     - data
-    # - 
+    # -
     - pil1M_image
-    
+
 viz_operator:
   listener:
     zmq_address: tcp://tiled_poller:5000  #only safe in containers
@@ -47,7 +48,7 @@ lse_reducer:
       python_class: CNNAutoencoder
       python_file: ./models/cnn_autoencoder/model.py
       type: torch
-    
+
     - name: AE
       state_dict: ./models/ae_v1/ae_model_finetuned.npz
       python_class: CNNAutoencoder
@@ -69,7 +70,7 @@ lse_broker:
 lse_worker:
   broker:
     dealer_address: tcp://lse_broker:5556
-    
+
 
 frame_publisher:
   zmq_publish_address: tcp://*:5558
@@ -86,4 +87,14 @@ lse_operator:
     router_address: tcp://lse_broker:5555
     router_hwm: 100000
 
-dynaconf_merge: true
+tiled_websocket_listener:
+  runs_segments:
+    - smi
+    - raw
+  frames_segments:
+    - primary
+    - data
+  stream_name:
+    - pi21M_image
+
+dynaconf_merge: true

src/_test/test_tiled_poller.py

@@ -8,7 +8,7 @@
     RawFrameEvent,
     SerializableNumpyArrayModel,
 )
-from arroyosas.tiled.tiled import unsent_frame_numbers
+from arroyosas.tiled.tiled_poller import unsent_frame_numbers
 
 # import pytest_asyncio
 # from arroyopy.operator import Operator