strand-braid-user: update Python scripts and book

Use improvements from the recent work in
strand_cam_subscriber.py

strand-braid-user/scripts/.gitignore

@@ -1 +1,3 @@
 strand-cam-cookies.json
+braid-cookies.json
+checkerboard.svg

strand-braid-user/scripts/braid_retransmit_udp.py

@@ -1,43 +1,55 @@
 #!/usr/bin/env python
 
-# This script listens to the HTTP JSON Event Stream of flydra2 and transmits
+# This script listens to the HTTP JSON Event Stream of braid and transmits
 # pose information over UDP in a simple text format.
 
-# This is an example of listening for the live stream of flydra2. In version 1
-# of the flydra2 pose api, in addition to the `Update` events, flydra2 also has
-# `Birth` and `Death` events. The `Birth` event returns the same data as an
-# `Update` event, whereas the `Death` event sends just `obj_id`.
+# This is an example of listening for the live stream of braid. There are 3
+# event types. In addition to the `Update` events, there are also `Birth` and
+# `Death` events. The `Birth` event returns the same data as an `Update` event,
+# whereas the `Death` event sends just `obj_id`.
 
-from __future__ import print_function
 import argparse
 import requests
 import json
-import time
 import socket
+import os
 
 DATA_PREFIX = "data: "
+COOKIE_JAR_FNAME = "braid-cookies.json"
 
-
-class Flydra2Proxy:
-    def __init__(self, flydra2_url):
-        self.flydra2_url = flydra2_url
+class BraidProxy:
+    def __init__(self, braid_url):
+        self.braid_url = braid_url
         self.session = requests.session()
-        r = self.session.get(self.flydra2_url)
-        assert r.status_code == requests.codes.ok
+
+        # If we have a cookie jar, load the cookies before initial request. This
+        # allows using a URL without a token.
+        if os.path.isfile(COOKIE_JAR_FNAME):
+            with open(COOKIE_JAR_FNAME, 'r') as f:
+                cookies = requests.utils.cookiejar_from_dict(json.load(f))
+                self.session.cookies.update(cookies)
+
+        r = self.session.get(self.braid_url)
+        r.raise_for_status()
+
+        # Store cookies
+        with open(COOKIE_JAR_FNAME, 'w') as f:
+            json.dump(requests.utils.dict_from_cookiejar(self.session.cookies), f)
+
 
     def run(self, udp_host, udp_port):
         addr = (udp_host, udp_port)
         print("sending flydra data to UDP %s" % (addr,))
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        events_url = self.flydra2_url + "events"
+        events_url = self.braid_url + "events"
         r = self.session.get(
             events_url, stream=True, headers={"Accept": "text/event-stream"},
         )
         for chunk in r.iter_content(chunk_size=None, decode_unicode=True):
             data = parse_chunk(chunk)
             # print('chunk value: %r'%data)
             version = data.get("v", 1)  # default because missing in first release
-            assert version == 2  # check the data version
+            assert version in (2, 3)  # check the data version
 
             try:
                 update_dict = data["msg"]["Update"]
@@ -63,7 +75,7 @@ def main():
     parser = argparse.ArgumentParser()
 
     parser.add_argument(
-        "--flydra2-url", default="http://127.0.0.1:8397/", help="URL of flydra2 server"
+        "--braid-url", default="http://127.0.0.1:8397/", help="URL of braid server"
     )
 
     parser.add_argument(
@@ -76,8 +88,8 @@ def main():
         help="UDP host to send pose information",
     )
     args = parser.parse_args()
-    flydra2 = Flydra2Proxy(args.flydra2_url)
-    flydra2.run(udp_host=args.udp_host, udp_port=args.udp_port)
+    braid = BraidProxy(args.braid_url)
+    braid.run(udp_host=args.udp_host, udp_port=args.udp_port)
 
 
 if __name__ == "__main__":

strand-braid-user/scripts/change-tracking-settings-demo.py

@@ -1,13 +1,14 @@
 #!/usr/bin/env python
-from __future__ import print_function
 import argparse
 import requests
 import json
 import time
 import threading
+import urllib
+import os
 
-DATA_PREFIX = "data: "
-
+DATA_PREFIX = b"data: "
+COOKIE_JAR_FNAME = "strand-cam-cookies.json"
 
 def maintain_state_copy(event_iterator, shared_state):
     for chunk in event_iterator:
@@ -17,9 +18,10 @@ def maintain_state_copy(event_iterator, shared_state):
 
 
 def parse_chunk(chunk):
-    lines = chunk.strip().split("\n")
+    lines = chunk.strip().split(b"\n")
+    print(lines)
     assert len(lines) == 2
-    if lines[0] != "event: bui_backend":
+    if lines[0] != b"event: strand-cam":
         return None
     assert lines[1].startswith(DATA_PREFIX)
     buf = lines[1][len(DATA_PREFIX) :]
@@ -31,31 +33,46 @@ class StrandCamProxy:
     def __init__(self, strand_cam_url):
         if not strand_cam_url.endswith("/"):
             strand_cam_url = strand_cam_url + "/"
-        self.callback_url = strand_cam_url + "callback"
+        self.callback_url = urllib.parse.urljoin(strand_cam_url, "callback")
 
-        # Setup initial session
         self.session = requests.session()
+        # If we have a cookie jar, load the cookies before initial request. This
+        # allows using a URL without a token.
+        if os.path.isfile(COOKIE_JAR_FNAME):
+            with open(COOKIE_JAR_FNAME, 'r') as f:
+                cookies = requests.utils.cookiejar_from_dict(json.load(f))
+                self.session.cookies.update(cookies)
+
+        # Setup initial session
         r = self.session.get(strand_cam_url)
-        assert r.status_code == requests.codes.ok
+        r.raise_for_status()
+
+        # Store cookies
+        with open(COOKIE_JAR_FNAME, 'w') as f:
+            json.dump(requests.utils.dict_from_cookiejar(self.session.cookies), f)
 
         # Create iterator which is updated with each new event
-        events_url = strand_cam_url + "strand-cam-events"
+        events_url = urllib.parse.urljoin(strand_cam_url, "strand-cam-events")
         r = self.session.get(
             events_url, stream=True, headers={"Accept": "text/event-stream"},
         )
-        assert r.status_code == requests.codes.ok
+        r.raise_for_status()
         event_iterator = r.iter_content(chunk_size=None)
 
         # Send this iterator to a new thread
         self.shared_state = {}
         thread = threading.Thread(
             target=maintain_state_copy, args=(event_iterator, self.shared_state)
         )
-        thread.setDaemon(True)
+        thread.daemon = True
         thread.start()
 
     def get_led1_state(self):
-        return self.shared_state["led_box_device_state"]["ch1"]
+        led_box_device_state = self.shared_state["led_box_device_state"]
+        if led_box_device_state is None:
+            raise RuntimeError("Strand Cam does not include LED state, "
+                               "are you using the flydratrax variant?")
+        return led_box_device_state["ch1"]
 
     def wait_until_first_update(self):
         while len(self.shared_state.keys()) == 0:
@@ -79,14 +96,8 @@ def send_config(self, mode):
         )
 
         params = {"ToCamera": {"CamArgSetLedProgramConfig": CamArgSetLedProgramConfig}}
-        body = json.dumps(params)
-        r = self.session.post(self.callback_url, data=body)
-        if r.status_code != requests.codes.ok:
-            print(
-                "error making request, status code {}".format(r.status_code),
-                file=sys.stderr,
-            )
-            sys.exit(1)
+        r = self.session.post(self.callback_url, json=params)
+        r.raise_for_status()
         print("made request with mode {mode}".format(mode=mode))
 
 
@@ -100,9 +111,13 @@ def main():
         help="URL of Strand Camera",
     )
     args = parser.parse_args()
+    print("1")
     strand_cam = StrandCamProxy(strand_cam_url=args.strand_cam_url)
+    print("2")
     strand_cam.wait_until_first_update()
+    print("3")
     while 1:
+        print("4")
         strand_cam.send_config(mode="Off")
         print("current LED state: ", strand_cam.get_led1_state())
         time.sleep(5.0)

strand-braid-user/scripts/convert_braidz_to_flydra_h5.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-from __future__ import print_function
 import sys
 import os
 import subprocess

strand-braid-user/scripts/draw_checkerboard_svg.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-from __future__ import division, print_function
 import argparse
 
 # 6x8 corners

strand-braid-user/scripts/record-mp4-video-braid-all-cams.py

@@ -3,33 +3,39 @@
 import json
 import time
 import sys
-from urllib.parse import urlparse
+import urllib
 import requests  # https://docs.python-requests.org/en/latest/user/install
+import os
+
+COOKIE_JAR_FNAME = "braid-cookies.json"
 
 
 class BraidProxy:
     def __init__(self, braid_url):
-        self.callback_url = urlparse(braid_url)._replace(path="callback").geturl()
+        self.callback_url = urllib.parse.urljoin(braid_url, "callback")
         # Setup initial session
         self.session = requests.session()
+
+        # If we have a cookie jar, load the cookies before initial request. This
+        # allows using a URL without a token.
+        if os.path.isfile(COOKIE_JAR_FNAME):
+            with open(COOKIE_JAR_FNAME, 'r') as f:
+                cookies = requests.utils.cookiejar_from_dict(json.load(f))
+                self.session.cookies.update(cookies)
+
         r = self.session.get(braid_url)
-        if r.status_code != requests.codes.ok:
-            print(f"request URL: {braid_url}")
-            print("request failed. response:")
-            print(r.text)
-            raise RuntimeError("connection to braid failed.")
+        r.raise_for_status()
+
+        # Store cookies
+        with open(COOKIE_JAR_FNAME, 'w') as f:
+            json.dump(requests.utils.dict_from_cookiejar(self.session.cookies), f)
+
 
     def send(self, cmd_dict):
-        body = json.dumps(cmd_dict)
         r = self.session.post(
-            self.callback_url, data=body, headers={"Content-Type": "application/json"}
+            self.callback_url, json=cmd_dict}
         )
-        if r.status_code != requests.codes.ok:
-            print(
-                "error making request, status code {}".format(r.status_code),
-                file=sys.stderr,
-            )
-            sys.exit(1)
+        r.raise_for_status()
 
 
 def main():

strand-braid-user/scripts/record-mp4-video.py

@@ -1,12 +1,13 @@
 #!/usr/bin/env python
-from __future__ import print_function
 import argparse
+import os
 import json
 import time
 import threading
-import sys
+import urllib
 import requests  # https://docs.python-requests.org/en/latest/user/install
 
+COOKIE_JAR_FNAME = "strand-cam-cookies.json"
 
 def maintain_state_copy(event_iterator, shared_state):
     for chunk in event_iterator:
@@ -31,21 +32,30 @@ def parse_chunk(chunk):
 
 class StrandCamProxy:
     def __init__(self, strand_cam_url):
-        if not strand_cam_url.endswith("/"):
-            strand_cam_url = strand_cam_url + "/"
-        self.callback_url = strand_cam_url + "callback"
+        self.callback_url = urllib.parse.urljoin(strand_cam_url, "callback")
 
-        # Setup initial session
         self.session = requests.session()
+        # If we have a cookie jar, load the cookies before initial request. This
+        # allows using a URL without a token.
+        if os.path.isfile(COOKIE_JAR_FNAME):
+            with open(COOKIE_JAR_FNAME, 'r') as f:
+                cookies = requests.utils.cookiejar_from_dict(json.load(f))
+                self.session.cookies.update(cookies)
+
+        # Pass any token given and setup cookies.
         r = self.session.get(strand_cam_url)
-        assert r.status_code == requests.codes.ok
+        r.raise_for_status()
+
+        # Store cookies
+        with open(COOKIE_JAR_FNAME, 'w') as f:
+            json.dump(requests.utils.dict_from_cookiejar(self.session.cookies), f)
 
         # Create iterator which is updated with each new event
-        events_url = strand_cam_url + "strand-cam-events"
+        events_url = urllib.parse.urljoin(strand_cam_url, "strand-cam-events")
         r = self.session.get(
             events_url, stream=True, headers={"Accept": "text/event-stream"},
         )
-        assert r.status_code == requests.codes.ok
+        r.raise_for_status()
         event_iterator = r.iter_content(chunk_size=None)
 
         # Send this iterator to a new thread
@@ -65,14 +75,8 @@ def wait_until_first_update(self):
 
     def send_to_camera(self, cmd_dict):
         params = {"ToCamera": cmd_dict}
-        body = json.dumps(params)
-        r = self.session.post(self.callback_url, data=body)
-        if r.status_code != requests.codes.ok:
-            print(
-                "error making request, status code {}".format(r.status_code),
-                file=sys.stderr,
-            )
-            sys.exit(1)
+        r = self.session.post(self.callback_url, json=params)
+        r.raise_for_status()
 
 
 def main():

strand-braid-user/users-guide/src/scripting-with-python.md

@@ -15,6 +15,16 @@ demo](https://github.com/strawlab/strand-braid/blob/main/strand-braid-user/scrip
 TODO: describe how to use and modify the [`record-mp4-video-braid-all-cams.py`
 demo](strand-braid-user/scripts/record-mp4-video-braid-all-cams.py).
 
+## Demo: save preview images to disk from Strand Camera using Python
+
+TODO: describe how to use and modify the [`strand_cam_subscriber.py`
+demo](https://github.com/strawlab/strand-braid/blob/main/strand-braid-user/scripts/strand_cam_subscriber.py).
+
+## Demo: listen to realtime 3D tracking data using Python
+
+TODO: describe how to use and modify the [`braid_retransmit_udp.py`
+demo](https://github.com/strawlab/strand-braid/blob/main/strand-braid-user/scripts/braid_retransmit_udp.py).
+
 ## Advanced: automating manual actions
 
 TODO: describe how to use the developer tools to watch the network requests from