Feat/event publisher

.github/workflows/publish.yml

@@ -15,8 +15,6 @@ env:
 
 jobs:
   publish-tag:
-    # On a tag push to a public repo: build and push that one tag plus :latest.
-    # On a private repo: this job is skipped; the tag still exists in git.
     if: |
       github.event_name == 'push' &&
       github.event.repository.private == false
@@ -38,12 +36,11 @@ jobs:
           push: true
           tags: |
             ${{ env.IMAGE }}:${{ github.ref_name }}
-            ${{ env.IMAGE }}:latest
+            ${{ !contains(github.ref_name, '-') && format('{0}:latest', env.IMAGE) || '' }}
           cache-from: type=gha
           cache-to: type=gha,mode=max
 
   backfill:
-    # On private->public flip or manual re-run: build every existing v* tag.
     if: github.event_name == 'public' || github.event_name == 'workflow_dispatch'
     runs-on: ubuntu-latest
     steps:
@@ -68,10 +65,24 @@ jobs:
           for tag in $tags; do
             echo "::group::$tag"
             git checkout "$tag"
-            docker build -t "$IMAGE:$tag" .
-            docker push "$IMAGE:$tag"
+            # Only "manifest unknown" means unpublished; any other registry error stops the job rather than risk overwriting a tag.
+            if inspect=$(docker manifest inspect "$IMAGE:$tag" 2>&1); then
+              echo "$IMAGE:$tag is published, skipped"
+            elif grep -q "manifest unknown" <<<"$inspect"; then
+              docker build -t "$IMAGE:$tag" .
+              docker push "$IMAGE:$tag"
+            else
+              echo "$inspect" >&2
+              exit 1
+            fi
             echo "::endgroup::"
           done
-          latest=$(echo "$tags" | tail -1)
+          # latest points at the newest tag without a hyphen, so a pre-release never becomes latest.
+          latest=$(echo "$tags" | { grep -v -- '-' || true; } | tail -1)
+          if [[ -z "$latest" ]]; then
+            echo "No release tag without a hyphen, latest left unchanged."
+            exit 0
+          fi
+          docker pull "$IMAGE:$latest"
           docker tag "$IMAGE:$latest" "$IMAGE:latest"
           docker push "$IMAGE:latest"

Dockerfile

@@ -1,4 +1,4 @@
-FROM ubuntu:20.04
+FROM ubuntu:20.04@sha256:8feb4d8ca5354def3d8fce243717141ce31e2c428701f6682bd2fafe15388214
 
 # Avoid interactive prompts
 ARG DEBIAN_FRONTEND=noninteractive
@@ -22,10 +22,13 @@ ENV LANG=en_US.UTF-8
 ENV LANGUAGE=en_US:en
 ENV LC_ALL=en_US.UTF-8
 
+# Package versions for both interpreters
+COPY docker/constraints-py3.txt docker/constraints-py2.txt /tmp/
+
 # Install Python packages
 # qibullet pulls in pybullet, numpy, cv2, etc.
 # flask for the shim server
-RUN pip3 install --no-cache-dir \
+RUN pip3 install --no-cache-dir -c /tmp/constraints-py3.txt \
     qibullet \
     flask \
     numpy \
@@ -34,8 +37,9 @@ RUN pip3 install --no-cache-dir \
 # --- Legacy Support for Physical Robot (Python 2.7) ---
 RUN apt-get update && apt-get install -y python2 libpython2.7 curl && \
     curl https://bootstrap.pypa.io/pip/2.7/get-pip.py --output get-pip.py && \
-    python2 get-pip.py && \
-    python2 -m pip install flask requests "pyzmq<20" && \
+    echo "40ee07eac6674b8d60fce2bbabc148cf0e2f1408c167683f110fd608b8d6f416  get-pip.py" | sha256sum -c - && \
+    python2 get-pip.py "pip==20.3.4" "setuptools==44.1.1" "wheel==0.37.1" && \
+    python2 -m pip install -c /tmp/constraints-py2.txt flask requests "pyzmq<20" && \
     rm -rf /var/lib/apt/lists/*
 
 # pynaoqi SDK is supplied at runtime by bind-mounting the user's licit copy at
@@ -53,6 +57,9 @@ RUN mkdir .qibullet
 # Copy Bridge Code
 COPY --chown=pepperdev:pepperdev py3-naoqi-bridge /home/pepperdev/py3-naoqi-bridge
 
+# Copy Config
+COPY --chown=pepperdev:pepperdev config /home/pepperdev/config
+
 # Copy Simulation Code
 COPY --chown=pepperdev:pepperdev src /home/pepperdev/src
 

README.md

@@ -76,6 +76,8 @@ We ship two ZMQ subscribers alongside the client for low-latency data streaming:
 - `clients.vision_client.VisionClient`: greyscale QVGA video on port `5559` (currently physical only).
 - `clients.state_client.StateClient`: head joints (full body on the roadmap) at 50 Hz on port `5560`, with time-interpolated lookup (both robot and sim).
 
+When `NAOQI_IP` points at a NAOqi, the bridge also publishes on port `5565` a message whenever a behaviour, such as an animation, starts or stops, and whenever NAOqi raises an ALMemory event listed in `config/events.txt`. Each message has the NAOqi name as its topic and `{"t", "value"}` as its payload, and a `heartbeat` goes out once a second. List one event name or `fnmatch` pattern per line in `config/events.txt`; the default `*` covers every event, and you can mount your own file over it.
+
 ## Physical robot setup
 
 You will need to install the SoftBank Robotics pynaoqi SDK once. PepperBox cannot ship it for legal reasons, but our `setup.sh` script fetches it directly from Aldebaran's CDN for you, with a Wayback Machine fallback, and verifies the SHA256. 

config/events.txt

@@ -0,0 +1 @@
+*

docker/constraints-py2.txt

@@ -0,0 +1,13 @@
+# Python 2 package versions of the pepper-box image, from the pip freeze of v2.3.1.
+certifi==2021.10.8
+chardet==4.0.0
+click==7.1.2
+Flask==1.1.4
+idna==2.10
+itsdangerous==1.1.0
+Jinja2==2.11.3
+MarkupSafe==1.1.1
+pyzmq==19.0.2
+requests==2.27.1
+urllib3==1.26.20
+Werkzeug==1.0.1

docker/constraints-py3.txt

@@ -0,0 +1,14 @@
+# Python 3 package versions of the pepper-box image, from the pip freeze of v2.3.1.
+blinker==1.8.2
+click==8.1.8
+flask==3.0.3
+importlib-metadata==8.5.0
+itsdangerous==2.2.0
+jinja2==3.1.6
+MarkupSafe==2.1.5
+numpy==1.24.4
+pybullet==3.2.7
+pyzmq==27.1.0
+qibullet==1.4.6
+werkzeug==3.0.6
+zipp==3.20.2

entrypoint.sh

@@ -61,5 +61,7 @@ EOF
     fi
 
     python2 py3-naoqi-bridge/video_streamer.py &
+    # Restart the event publisher whenever it exits.
+    (while true; do python2 py3-naoqi-bridge/event_publisher.py; sleep 2; done) &
     python2 py3-naoqi-bridge/shim_server.py
 fi

py3-naoqi-bridge/event_publisher.py

@@ -0,0 +1,195 @@
+# py3-naoqi-bridge/event_publisher.py (Python 2)
+"""NAOqi behaviour signals and ALMemory events -> ZMQ PUB :5565, with a heartbeat each time the robot answers a ping."""
+from __future__ import print_function
+import fnmatch
+import json
+import os
+import sys
+import threading
+import time
+import Queue  # Py2
+import zmq
+
+
+EVENT_PUB_PORT = 5565
+HEARTBEAT_S = 1.0
+BEHAVIOR_SIGNALS = ("behaviorStarted", "behaviorStopped")
+EVENTS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config", "events.txt")
+
+
+def read_patterns(path):
+    """Event name patterns from the config file, one per line; blank lines and lines starting with # are skipped."""
+    with open(path) as f:
+        lines = [line.strip() for line in f]
+    return [line for line in lines if line and not line.startswith("#")]
+
+
+def matches(name, patterns):
+    """True when the event name matches any of the fnmatch patterns, case-sensitively."""
+    return any(fnmatch.fnmatchcase(name, pattern) for pattern in patterns)
+
+
+def event_message(topic, value, t):
+    """Topic and payload for one NAOqi signal or event; a value JSON cannot encode is sent as its repr."""
+    try:
+        json.dumps(value)
+    except (TypeError, ValueError, UnicodeDecodeError):
+        value = repr(value)
+    return (topic, {"t": t, "value": value})
+
+
+def connect(session, publisher, patterns):
+    """Forward the behaviour signals and every ALMemory event matching patterns, including events declared later; returns the subscribers to keep alive."""
+    def forward(topic, value):
+        publisher.enqueue(event_message(topic, value, time.time()))
+
+    behavior_manager = session.service("ALBehaviorManager")
+    for name in BEHAVIOR_SIGNALS:
+        topic = "ALBehaviorManager." + name
+        getattr(behavior_manager, name).connect(lambda value, topic=topic: forward(topic, value))
+
+    memory = session.service("ALMemory")
+    lock = threading.Lock()
+    subscribers = {}
+
+    def subscribe(event):
+        with lock:
+            if event in subscribers or not matches(event, patterns):
+                return
+            subscriber = memory.subscriber(event)
+            subscriber.signal.connect(lambda value, topic=event: forward(topic, value))
+            subscribers[event] = subscriber
+
+    def on_key_added(key):
+        """Handle ALMemory/KeyAdded, which ALMemory raises with the name of every new key, event or data; subscribe only if it is an event."""
+        if key in memory.getEventList():
+            subscribe(key)
+
+    def on_key_removed(key):
+        """Handle ALMemory/KeyRemoved, which ALMemory raises with the name of every deleted key; forget its subscription."""
+        with lock:
+            subscribers.pop(key, None)
+
+    watchers = [memory.subscriber("ALMemory/KeyAdded"), memory.subscriber("ALMemory/KeyRemoved")]
+    watchers[0].signal.connect(on_key_added)
+    watchers[1].signal.connect(on_key_removed)
+    for event in memory.getEventList():
+        subscribe(event)
+    return {"watchers": watchers, "events": subscribers}
+
+
+class PublisherThread(threading.Thread):
+    """Owns the PUB socket and sends queued messages as topic and JSON frames."""
+
+    def __init__(self, port=EVENT_PUB_PORT):
+        threading.Thread.__init__(self)
+        self.daemon = True
+        self.queue = Queue.Queue(maxsize=256)
+        self._stop = threading.Event()
+        self._sock = zmq.Context.instance().socket(zmq.PUB)
+        self._sock.bind("tcp://*:%d" % port)
+
+    def enqueue(self, message):
+        try:
+            self.queue.put_nowait(message)
+        except Queue.Full:
+            # Drop rather than block the NAOqi dispatcher.
+            sys.stderr.write("event_publisher: queue full; dropped one message\n")
+
+    def stop(self):
+        self._stop.set()
+
+    def _send(self, topic, payload):
+        try:
+            self._sock.send_multipart([topic.encode("utf-8"), json.dumps(payload).encode("utf-8")])
+        except zmq.ZMQError as e:
+            sys.stderr.write("event_publisher: zmq send error: %s\n" % e)
+
+    def run(self):
+        while not self._stop.is_set():
+            try:
+                topic, payload = self.queue.get(timeout=0.05)
+            except Queue.Empty:
+                continue
+            self._send(topic, payload)
+        self._sock.close()
+
+
+class HeartbeatThread(threading.Thread):
+    """Pings the robot every heartbeat_s and enqueues a heartbeat for each answer, so heartbeats stop when the robot stops answering."""
+
+    def __init__(self, memory, publisher, started, heartbeat_s=HEARTBEAT_S):
+        threading.Thread.__init__(self)
+        self.daemon = True
+        self._memory = memory
+        self._publisher = publisher
+        self._started = started
+        self._heartbeat_s = heartbeat_s
+        self._stop = threading.Event()
+
+    def stop(self):
+        self._stop.set()
+
+    def run(self):
+        while not self._stop.is_set():
+            try:
+                self._memory.ping()
+                self._publisher.enqueue(("heartbeat", {"t": time.time(), "started": self._started}))
+            except Exception as e:
+                sys.stderr.write("event_publisher: robot ping failed: %s\n" % e)
+            self._stop.wait(self._heartbeat_s)
+
+
+def start(session, started, patterns, port=EVENT_PUB_PORT, heartbeat_s=HEARTBEAT_S):
+    """Connect the robot's signals and events, then start publishing and pinging; returns the two threads and the subscribers to keep alive."""
+    publisher = PublisherThread(port=port)
+    subscribers = connect(session, publisher, patterns)
+    heartbeat = HeartbeatThread(session.service("ALMemory"), publisher, started, heartbeat_s)
+    publisher.start()
+    heartbeat.start()
+    return publisher, heartbeat, subscribers
+
+
+def main():
+    try:
+        import qi
+    except ImportError:
+        sys.stderr.write("event_publisher: qi not importable; exiting.\n")
+        return 2
+
+    try:
+        patterns = read_patterns(EVENTS_FILE)
+    except IOError as e:
+        sys.stderr.write("event_publisher: cannot read %s: %s\n" % (EVENTS_FILE, e))
+        return 5
+
+    naoqi_ip = os.environ.get("NAOQI_IP", "127.0.0.1")
+    naoqi_port = int(os.environ.get("NAOQI_PORT", "9559"))
+    app = qi.Application(["event_publisher", "--qi-url=tcp://%s:%d" % (naoqi_ip, naoqi_port)])
+    try:
+        app.start()
+    except RuntimeError as e:
+        sys.stderr.write("event_publisher: NAOqi unreachable at %s:%d: %s\n" % (naoqi_ip, naoqi_port, e))
+        return 3
+
+    session = app.session
+    dropped = threading.Event()
+    session.disconnected.connect(lambda reason: dropped.set())
+    publisher, heartbeat, kept = start(session, started=time.time(), patterns=patterns)
+    sys.stderr.write("event_publisher: publishing on :%d from %s:%d\n" % (EVENT_PUB_PORT, naoqi_ip, naoqi_port))
+
+    try:
+        while not dropped.wait(1.0):
+            pass
+        sys.stderr.write("event_publisher: robot session dropped; exiting\n")
+        return 4
+    except KeyboardInterrupt:
+        return 0
+    finally:
+        heartbeat.stop()
+        publisher.stop()
+        publisher.join(timeout=2.0)
+
+
+if __name__ == "__main__":
+    sys.exit(main())