diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 0f2c1cf..49379ed 100644 --- a/.github/workflows/publish.yml +++ b/.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" diff --git a/Dockerfile b/Dockerfile index 7955c87..f4cac0b 100644 --- a/Dockerfile +++ b/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 diff --git a/README.md b/README.md index d6e85de..ae2eafd 100644 --- a/README.md +++ b/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. diff --git a/config/events.txt b/config/events.txt new file mode 100644 index 0000000..72e8ffc --- /dev/null +++ b/config/events.txt @@ -0,0 +1 @@ +* diff --git a/docker/constraints-py2.txt b/docker/constraints-py2.txt new file mode 100644 index 0000000..682f807 --- /dev/null +++ b/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 diff --git a/docker/constraints-py3.txt b/docker/constraints-py3.txt new file mode 100644 index 0000000..4b5ee3c --- /dev/null +++ b/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 diff --git a/entrypoint.sh b/entrypoint.sh index 7d63d8d..489d6f4 100755 --- a/entrypoint.sh +++ b/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 diff --git a/py3-naoqi-bridge/event_publisher.py b/py3-naoqi-bridge/event_publisher.py new file mode 100644 index 0000000..7727d2c --- /dev/null +++ b/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()) diff --git a/py3-naoqi-bridge/tests/py2/test_event_publisher.py b/py3-naoqi-bridge/tests/py2/test_event_publisher.py new file mode 100644 index 0000000..6501b27 --- /dev/null +++ b/py3-naoqi-bridge/tests/py2/test_event_publisher.py @@ -0,0 +1,272 @@ +"""Tests for the event publisher: patterns, message building, the robot signal and event wiring and the ZMQ stream.""" +import json +import os +import shutil +import tempfile +import threading +import time +import unittest + +import zmq + +from event_publisher import HeartbeatThread, PublisherThread, connect, event_message, read_patterns, start + + +class FakeSignal(object): + """A qi signal whose callbacks the test fires by hand; each connection is written to the shared call log.""" + + def __init__(self, log): + self.callbacks = [] + self.log = log + + def connect(self, callback): + self.log.append("connect") + self.callbacks.append(callback) + return len(self.callbacks) + + def fire(self, *args): + for callback in self.callbacks: + callback(*args) + + +class FakeSubscriber(object): + def __init__(self, log): + self.signal = FakeSignal(log) + + +class FakeMemory(object): + """ALMemory with a declared event list, event subscribers and a ping whose answer the test sets.""" + + def __init__(self, log=None, events=()): + self.events = list(events) + self.subscribers = {} + self.log = [] if log is None else log + self.answer = lambda: True + + def getEventList(self): + return list(self.events) + + def subscriber(self, event): + subscriber = FakeSubscriber(self.log) + self.subscribers.setdefault(event, []).append(subscriber) + return subscriber + + def fire(self, event, value): + for subscriber in self.subscribers.get(event, []): + subscriber.signal.fire(value) + + def ping(self): + self.log.append("ping") + return self.answer() + + +class FakeBehaviorManager(object): + def __init__(self, log): + self.behaviorStarted = FakeSignal(log) + self.behaviorStopped = FakeSignal(log) + + +class FakeSession(object): + """A qi session serving the two NAOqi modules the publisher listens to, with one call log for both.""" + + def __init__(self, events=()): + self.log = [] + self.memory = FakeMemory(self.log, events) + self.behavior_manager = FakeBehaviorManager(self.log) + + def service(self, name): + return {"ALMemory": self.memory, "ALBehaviorManager": self.behavior_manager}[name] + + +class RecordingPublisher(object): + def __init__(self): + self.messages = [] + + def enqueue(self, message): + self.messages.append(message) + + def sent(self): + """The topic and value of every message, in order.""" + return [(topic, payload["value"]) for topic, payload in self.messages] + + +class PatternTests(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.dir) + + def test_patterns_are_read_one_per_line_without_blanks_or_comments(self): + """Blank lines and lines starting with # are skipped, and surrounding spaces are stripped.""" + path = os.path.join(self.dir, "events.txt") + with open(path, "w") as f: + f.write("*\n\n# a note\n AutonomousLife/* \n") + self.assertEqual(read_patterns(path), ["*", "AutonomousLife/*"]) + + +class MessageTests(unittest.TestCase): + def test_a_value_json_cannot_encode_is_sent_as_its_repr(self): + """An object or a byte string that is not UTF-8 becomes its repr, so the message still goes out.""" + thing = object() + self.assertEqual(event_message("Some/Event", thing, 1.0)[1]["value"], repr(thing)) + self.assertEqual(event_message("Some/Event", "\xff", 1.0)[1]["value"], repr("\xff")) + + +class ConnectTests(unittest.TestCase): + def test_behaviour_starts_and_stops_are_forwarded_with_their_paths(self): + """Every behaviour start and stop is forwarded with its path.""" + session = FakeSession() + publisher = RecordingPublisher() + connect(session, publisher, ["*"]) + session.behavior_manager.behaviorStarted.fire("animations/Stand/Gestures/Me_8") + session.behavior_manager.behaviorStarted.fire("robot-page/behavior_1") + session.behavior_manager.behaviorStopped.fire("animations/Stand/Gestures/Me_8") + self.assertEqual(publisher.sent(), [ + ("ALBehaviorManager.behaviorStarted", "animations/Stand/Gestures/Me_8"), + ("ALBehaviorManager.behaviorStarted", "robot-page/behavior_1"), + ("ALBehaviorManager.behaviorStopped", "animations/Stand/Gestures/Me_8")]) + for _, payload in publisher.messages: + self.assertIsInstance(payload["t"], float) + + def test_matching_events_are_forwarded_with_their_raw_values(self): + """Events matching the patterns are forwarded with every value NAOqi raises; others are never subscribed.""" + session = FakeSession(events=["ALTextToSpeech/TextStarted", "AutonomousLife/State", "PeoplePerception/PeopleList"]) + publisher = RecordingPublisher() + connect(session, publisher, ["ALTextToSpeech/*", "AutonomousLife/State"]) + self.assertNotIn("PeoplePerception/PeopleList", session.memory.subscribers) + session.memory.fire("ALTextToSpeech/TextStarted", 1) + session.memory.fire("ALTextToSpeech/TextStarted", 0) + session.memory.fire("AutonomousLife/State", "solitary") + self.assertEqual(publisher.sent(), [ + ("ALTextToSpeech/TextStarted", 1), + ("ALTextToSpeech/TextStarted", 0), + ("AutonomousLife/State", "solitary")]) + + def test_an_event_declared_later_is_subscribed(self): + """An event an app declares after the start is announced by KeyAdded and then forwarded.""" + session = FakeSession() + publisher = RecordingPublisher() + connect(session, publisher, ["*"]) + session.memory.events.append("App/NewEvent") + session.memory.fire("ALMemory/KeyAdded", "App/NewEvent") + session.memory.fire("App/NewEvent", 7) + self.assertEqual(publisher.sent(), [("App/NewEvent", 7)]) + + def test_a_data_key_added_later_is_not_subscribed(self): + """KeyAdded also announces data keys, which are not in the event list and are left alone.""" + session = FakeSession() + connect(session, RecordingPublisher(), ["*"]) + session.memory.fire("ALMemory/KeyAdded", "App/SomeData") + self.assertNotIn("App/SomeData", session.memory.subscribers) + + def test_a_removed_event_is_subscribed_again_when_declared_again(self): + """KeyRemoved forgets the subscription, so a second declaration subscribes afresh.""" + session = FakeSession(events=["App/Event"]) + kept = connect(session, RecordingPublisher(), ["*"]) + session.memory.fire("ALMemory/KeyRemoved", "App/Event") + self.assertNotIn("App/Event", kept["events"]) + session.memory.fire("ALMemory/KeyAdded", "App/Event") + self.assertEqual(len(session.memory.subscribers["App/Event"]), 2) + self.assertIn("App/Event", kept["events"]) + + def test_returns_the_subscribers_to_keep_alive(self): + """The event subscribers and the two key watchers are returned, because a dropped subscriber object ends its subscription.""" + session = FakeSession(events=["ALTextToSpeech/TextStarted", "AutonomousLife/State"]) + kept = connect(session, RecordingPublisher(), ["*"]) + self.assertEqual(sorted(kept["events"]), ["ALTextToSpeech/TextStarted", "AutonomousLife/State"]) + self.assertEqual(len(kept["watchers"]), 2) + self.assertEqual(kept["watchers"], session.memory.subscribers["ALMemory/KeyAdded"] + session.memory.subscribers["ALMemory/KeyRemoved"]) + + +class HeartbeatTests(unittest.TestCase): + def run_heartbeat(self, memory, seconds=0.2): + """Run a heartbeat thread against memory for a while and return what it enqueued.""" + publisher = RecordingPublisher() + thread = HeartbeatThread(memory, publisher, started=42.0, heartbeat_s=0.01) + thread.start() + time.sleep(seconds) + thread.stop() + thread.join(timeout=1.0) + return publisher.messages + + def test_each_answer_from_the_robot_is_a_heartbeat_with_the_start_time(self): + """Every ping the robot answers is followed by one heartbeat carrying its time and the publisher's start time.""" + memory = FakeMemory() + messages = self.run_heartbeat(memory) + self.assertTrue(messages) + self.assertEqual(len(messages), memory.log.count("ping")) + for topic, payload in messages: + self.assertEqual((topic, payload["started"]), ("heartbeat", 42.0)) + self.assertIsInstance(payload["t"], float) + + def test_no_heartbeat_while_the_robot_does_not_answer(self): + """A ping that never returns, as on a link that silently drops the robot's packets, stops the heartbeats.""" + unblock = threading.Event() + memory = FakeMemory() + memory.answer = unblock.wait + try: + self.assertEqual(self.run_heartbeat(memory), []) + finally: + unblock.set() + + def test_a_failed_ping_is_skipped_and_pinging_goes_on(self): + """A ping that raises gives no heartbeat, and the next answer gives one.""" + memory = FakeMemory() + answers = iter([RuntimeError("no route to robot")]) + + def answer(): + error = next(answers, None) + if error: + raise error + return True + + memory.answer = answer + messages = self.run_heartbeat(memory) + self.assertTrue(messages) + self.assertEqual(len(messages), memory.log.count("ping") - 1) + + +class StartTests(unittest.TestCase): + def test_the_robot_is_pinged_only_after_its_signals_and_events_are_connected(self): + """The first ping comes after the last signal or event is connected, so a heartbeat always means events can arrive.""" + session = FakeSession(events=["ALTextToSpeech/TextStarted", "ALTextToSpeech/TextDone"]) + publisher, heartbeat, _ = start(session, started=1.0, patterns=["*"], port=15566, heartbeat_s=0.01) + try: + time.sleep(0.1) + finally: + heartbeat.stop() + publisher.stop() + heartbeat.join(timeout=1.0) + publisher.join(timeout=1.0) + self.assertIn("ping", session.log) + last_connect = max(i for i, entry in enumerate(session.log) if entry == "connect") + self.assertGreater(session.log.index("ping"), last_connect) + + +class StreamTests(unittest.TestCase): + PORT = 15565 + + def setUp(self): + self.thread = PublisherThread(port=self.PORT) + self.thread.start() + self.sub = zmq.Context.instance().socket(zmq.SUB) + self.sub.setsockopt(zmq.RCVTIMEO, 1000) + self.sub.connect("tcp://localhost:%d" % self.PORT) + + def tearDown(self): + self.sub.close() + self.thread.stop() + self.thread.join(timeout=2.0) + + def test_messages_arrive_as_topic_and_json(self): + """An enqueued message is sent as a topic frame and a JSON frame, and a subscription to a NAOqi prefix receives its events.""" + self.sub.setsockopt(zmq.SUBSCRIBE, b"ALTextToSpeech/") + time.sleep(0.2) + self.thread.enqueue(("ALTextToSpeech/TextStarted", {"t": 1.5, "value": 1})) + received_topic, body = self.sub.recv_multipart() + self.assertEqual((received_topic, json.loads(body)), (b"ALTextToSpeech/TextStarted", {"t": 1.5, "value": 1})) + + +if __name__ == "__main__": + unittest.main()