chore: cleanup for merging/ switching to later branches

Implement negotiation and actuation endpoints

See merge request ics/sp/2025/n25b/pepperplus-ri!5

.githooks/commit-msg

@@ -0,0 +1,16 @@
+#!/bin/sh
+
+commit_msg_file=$1
+commit_msg=$(cat "$commit_msg_file")
+
+if echo "$commit_msg" | grep -Eq "^(feat|fix|refactor|perf|style|test|docs|build|chore|revert): .+"; then
+  if echo "$commit_msg" | grep -Eq "^(ref|close):\sN25B-.+"; then
+    exit 0
+  else
+    echo "❌ Commit message invalid! Must end with [ref/close]: N25B-000"
+    exit 1
+  fi
+else
+  echo "❌ Commit message invalid! Must start with <type>: <description>"
+  exit 1
+fi

.githooks/pre-commit

@@ -0,0 +1,17 @@
+#!/bin/sh
+
+# Get current branch
+branch=$(git rev-parse --abbrev-ref HEAD)
+
+if echo "$branch" | grep -Eq "(dev|main)"; then
+  echo 0
+fi
+
+# allowed pattern <type/>
+if echo "$branch" | grep -Eq "^(feat|fix|refactor|perf|style|test|docs|build|chore|revert)\/\w+(-\w+){0,5}$"; then
+  exit 0
+else
+  echo "❌ Invalid branch name: $branch"
+  echo "Branch must be named <type>/<description-of-branch> (must have one to six words separated by a dash)"
+  exit 1
+fi

.githooks/prepare-commit-msg

@@ -0,0 +1,9 @@
+#!/bin/sh
+
+echo "#<type>: <description>
+
+#[optional body]
+
+#[optional footer(s)]
+
+#[ref/close]: <issue identifier>" > $1

.gitignore

@@ -0,0 +1,219 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#   Usually these files are written by a python script from a template
+#   before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+# Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+# uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+# poetry.lock
+# poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+# pdm.lock
+# pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+# pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# Redis
+*.rdb
+*.aof
+*.pid
+
+# RabbitMQ
+mnesia/
+rabbitmq/
+rabbitmq-data/
+
+# ActiveMQ
+activemq-data/
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#   JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#   and can be added to the global gitignore or merged into this file.  For a more nuclear
+#   option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/
+
+# Abstra
+#   Abstra is an AI-powered process automation framework.
+#   Ignore directories containing user credentials, local state, and settings.
+#   Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#   Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
+#   that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#   and can be added to the global gitignore or merged into this file. However, if you prefer,
+#   you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/
+
+# Streamlit
+.streamlit/secrets.toml
+
+.DS_Store
+

README.md

@@ -2,12 +2,24 @@
 ### Linux (or WSL)
 Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell: 
 
+The robot interface is a high-level API for controlling the robot. It implements the API as designed: https://utrechtuniversity.youtrack.cloud/articles/N25B-A-14/RI-CB-Communication.
+
+This is an implementation for the Pepper robot, using the Pepper SDK and Python 2.7 as required by the SDK.
+
+
+
+## Installation
+
+### Linux (or WSL)
+
+Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell:
+
 ```bash
 pyenv install 2.7
 pyenv shell 2.7
 ```
 
-You can check that this worked by typing 
+You can check that this worked by typing
 
 ```bash
 python -V
@@ -24,13 +36,7 @@ python -m virtualenv .venv
 source .venv/bin/activate
 ```
 
-To be able to install the PyAudio Python package, you'll need to have the `portaudio` system package installed. On Debian or Ubuntu:
+Install the required packages with
-
-```shell
-sudo apt install portaudio19-dev
-```
-
-Then you can install the required packages with
 
 ```bash
 pip install -r requirements.txt
@@ -64,21 +70,67 @@ python -c "import qi; print(qi)"
 
 You should now be able to run this project.
 
-### MacOS
+### macOS
 
-On ARM CPU's, pyenv doesn't want to install Python 2. You can download and install it from [the Python website](https://www.python.org/downloads/release/python-2718/).
+Similar to Linux, but don't bother installing `pyenv` as it won't be able to install Python 2 on Apple Silicon. Instead, install Python 2.7.18 from the [Python website](https://www.python.org/downloads/release/python-2718/).
 
-Create a virtual environment as described in the Linux section.
+Create the virtual environment as described above in the Linux section. Stop at the point where it shows you how to download the NaoQi SDK. Instead, use:
 
-Then build `portaudio` for x86_64 CPU's.
+```shell
-
+curl -OL https://community-static.aldebaran.com/resources/2.5.10/Python%20SDK/pynaoqi-python2.7-2.5.7.1-mac64.tar.gz
-Then follow the remaining installation instructions in the Linux section.
-
-## Running
-Assuming you have the virtual environment activated (`source .venv/bin/activate` on Linux) and that you have a virtual robot running on localhost you should be able to run this project by typing
-
-```bash
-python main.py --qi-url tcp://localhost:<port>
 ```
 
-where `<port>` is the port on which your robot is running.
+Then resume the steps from above.
+
+
+
+## Usage
+
+On Linux and macOS:
+
+```shell
+PYTHONPATH=src python -m robot_interface.main
+```
+
+On Windows:
+
+```shell
+$env:PYTHONPATH="src"; python -m robot_interface.main
+```
+
+With both, if you want to connect to the actual robot (or simulator), pass the `--qi-url` argument.
+
+
+
+## Testing
+
+To run the unit tests, on Linux and macOS:
+
+```shell
+PYTHONPATH=src pytest test/
+```
+
+On Windows:
+
+```shell
+$env:PYTHONPATH="src"; pytest test/
+```
+
+### Coverage
+
+For coverage, add `--cov=robot_interface` as an argument to `pytest`. 
+
+
+
+## GitHooks
+
+To activate automatic commits/branch name checks run:
+
+```shell
+git config --local core.hooksPath .githooks
+```
+
+If your commit fails its either:
+branch name != <type>/description-of-branch , 
+commit name !=  <type>: description of the commit.
+                <ref>: N25B-Num's

main.py

@@ -1,69 +0,0 @@
-import sys
-import logging
-logging.
-
-import zmq
-
-from src.audio_streaming import AudioStreaming
-from state import state
-
-
-def say(session, message):
-    tts = session.service("ALTextToSpeech")
-    tts.say(message)
-
-
-def listen_for_messages(session):
-    context = zmq.Context()
-    socket = context.socket(zmq.SUB)
-    socket.connect("tcp://localhost:5556")
-    socket.setsockopt_string(zmq.SUBSCRIBE, u"")  # u because Python 2 shenanigans
-
-    poller = zmq.Poller()
-    poller.register(socket, zmq.POLLIN)
-
-    logging.info("Listening for messages")
-    while not state.exit_event.is_set():
-        if not poller.poll(200): continue  # At most 200 ms delay after CTRL+C
-        # We now know there's a message waiting for us
-        message = socket.recv_string()
-        logging.debug("Received message: {}".format(message))
-
-        if session: say(session, message)
-
-
-def get_session():
-    if "--qi-url" not in sys.argv:
-        logging.info("No Qi URL argument given. Running in stand-alone mode.")
-        return None
-
-    try:
-        import qi
-    except ImportError:
-        logging.info("Unable to import qi. Running in stand-alone mode.")
-        return None
-
-    try:
-        app = qi.Application()
-        app.start()
-        return app.session
-    except RuntimeError:
-        logging.info("Unable to connect to the robot. Running in stand-alone mode.")
-        return None
-
-
-def main():
-    session = get_session()
-
-    audio_streamer = AudioStreaming()
-    audio_streamer.run()
-
-    listen_for_messages(session)  # Runs indefinitely, until CTRL+C
-
-
-if __name__ == "__main__":
-    try:
-        state.initialize()
-        main()
-    finally:
-        state.deinitialize()

requirements.txt

@@ -1,2 +1,5 @@
 pyzmq<16
-pyaudio<=0.2.11
+pyaudio<=0.2.11
+pytest<5
+pytest-mock<3.0.0
+pytest-cov<3.0.0

src/robot_interface/endpoints/actuation_receiver.py

@@ -0,0 +1,49 @@
+import logging
+
+import zmq
+
+from robot_interface.endpoints.receiver_base import ReceiverBase
+from robot_interface.state import state
+
+
+class ActuationReceiver(ReceiverBase):
+    def __init__(self, zmq_context, port=5557):
+        """
+        The actuation receiver endpoint, responsible for handling speech and gesture requests.
+
+        :param zmq_context: The ZeroMQ context to use.
+        :type zmq_context: zmq.Context
+
+        :param port: The port to use.
+        :type port: int
+        """
+        super(ActuationReceiver, self).__init__("actuation")
+        self.create_socket(zmq_context, zmq.SUB, port)
+        self.socket.setsockopt_string(zmq.SUBSCRIBE, u"")  # Causes block if given in options
+        self._tts_service = None
+
+    def _handle_speech(self, message):
+        text = message.get("data")
+        if not text:
+            logging.warn("Received message to speak, but it lacks data.")
+            return
+
+        if not isinstance(text, (str, unicode)):
+            logging.warn("Received message to speak but it is not a string.")
+            return
+
+        logging.debug("Received message to speak: {}".format(text))
+
+        if not state.qi_session: return
+        # If state has a qi_session, we know that we can import qi
+        import qi  # Takes a while only the first time it's imported
+
+        if not self._tts_service:
+            self._tts_service = state.qi_session.service("ALTextToSpeech")
+
+        # Returns instantly. Messages received while speaking will be queued.
+        qi.async(self._tts_service.say, text)
+
+    def handle_message(self, message):
+        if message["endpoint"] == "actuate/speech":
+            self._handle_speech(message)

src/robot_interface/endpoints/main_receiver.py

@@ -0,0 +1,56 @@
+import zmq
+
+from robot_interface.endpoints.receiver_base import ReceiverBase
+from robot_interface.state import state
+
+
+class MainReceiver(ReceiverBase):
+    def __init__(self, zmq_context, port=5555):
+        """
+        The main receiver endpoint, responsible for handling ping and negotiation requests.
+
+        :param zmq_context: The ZeroMQ context to use.
+        :type zmq_context: zmq.Context
+
+        :param port: The port to use.
+        :type port: int
+        """
+        super(MainReceiver, self).__init__("main")
+        self.create_socket(zmq_context, zmq.REP, port, bind=False)
+
+    @staticmethod
+    def _handle_ping(message):
+        """A simple ping endpoint. Returns the provided data."""
+        return {"endpoint": "ping", "data": message.get("data")}
+
+    @staticmethod
+    def _handle_port_negotiation(message):
+        endpoints = [socket.endpoint_description() for socket in state.sockets]
+
+        return {"endpoint": "negotiate/ports", "data": endpoints}
+
+    @staticmethod
+    def _handle_negotiation(message):
+        """
+        Handle a negotiation request. Will respond with ports that can be used to connect to the robot.
+
+        :param message: The negotiation request message.
+        :type message: dict
+
+        :return: A response dictionary with a 'ports' key containing a list of ports and their function.
+        :rtype: dict[str, list[dict]]
+        """
+        # In the future, the sender could send information like the robot's IP address, etc.
+
+        if message["endpoint"] == "negotiate/ports":
+            return MainReceiver._handle_port_negotiation(message)
+
+        return {"endpoint": "negotiate/error", "data": "The requested endpoint is not implemented."}
+
+    def handle_message(self, message):
+        if message["endpoint"] == "ping":
+            return self._handle_ping(message)
+        elif message["endpoint"].startswith("negotiate"):
+            return self._handle_negotiation(message)
+
+        return {"endpoint": "error", "data": "The requested endpoint is not supported."}

src/robot_interface/endpoints/receiver_base.py

@@ -0,0 +1,21 @@
+from abc import ABCMeta, abstractmethod
+
+from robot_interface.endpoints.socket_base import SocketBase
+
+
+class ReceiverBase(SocketBase, object):
+    """Associated with a ZeroMQ socket."""
+    __metaclass__ = ABCMeta
+
+    @abstractmethod
+    def handle_message(self, message):
+        """
+        Handle a message with the receiver.
+
+        :param message: The message to handle, must contain properties "endpoint" and "data".
+        :type message: dict
+
+        :return: A response message or None if this type of receiver doesn't publish.
+        :rtype: dict | None
+        """
+        raise NotImplementedError()

src/robot_interface/endpoints/socket_base.py

@@ -0,0 +1,72 @@
+from abc import ABCMeta
+
+import zmq
+
+
+class SocketBase(object):
+    __metaclass__ = ABCMeta
+
+    name = None
+    socket = None
+
+    def __init__(self, identifier):
+        """
+        :param identifier: The identifier of the endpoint.
+        :type identifier: str
+        """
+        self.identifier = identifier
+        self.port = None  # Set later by `create_socket`
+        self.socket = None  # Set later by `create_socket`
+        self.bound = None  # Set later by `create_socket`
+
+    def create_socket(self, zmq_context, socket_type, port, options=[], bind=True):
+        """
+        Create a ZeroMQ socket.
+
+        :param zmq_context: The ZeroMQ context to use.
+        :type zmq_context: zmq.Context
+
+        :param socket_type: The type of socket to create. Use zmq constants, e.g. zmq.SUB or zmq.REP.
+        :type socket_type: int
+
+        :param port: The port to use.
+        :type port: int
+
+        :param options: A list of options to be set on the socket. The list contains tuples where the first element contains the option
+        and the second the value, for example (zmq.CONFLATE, 1).
+        :type options: list[tuple[int, int]]
+
+        :param bind: Whether to bind the socket or connect to it.
+        :type bind: bool
+        """
+        self.port = port
+        self.socket = zmq_context.socket(socket_type)
+
+        for option, arg in options:
+            self.socket.setsockopt(option,arg)
+
+        self.bound = bind
+        if bind:
+            self.socket.bind("tcp://*:{}".format(port))
+        else:
+            self.socket.connect("tcp://localhost:{}".format(port))
+
+    def close(self):
+        """Close the ZeroMQ socket."""
+        if not self.socket: return
+        self.socket.close()
+        self.socket = None
+
+    def endpoint_description(self):
+        """
+        Description of the endpoint. Used for negotiation.
+
+        :return: A dictionary with the following keys: id, port, bind. See API specification at:
+        https://utrechtuniversity.youtrack.cloud/articles/N25B-A-14/RI-CB-Communication#negotiation
+        :rtype: dict
+        """
+        return {
+            "id": self.identifier,
+            "port": self.port,
+            "bind": not self.bound
+        }

src/robot_interface/endpoints/video_sender.py

@@ -0,0 +1,49 @@
+import zmq
+import threading
+import qi
+import logging
+
+from robot_interface.endpoints.socket_base import SocketBase
+from robot_interface.state import state
+
+
+class VideoSender(SocketBase):
+    def __init__(self, zmq_context, port=5556):
+        super(VideoSender, self).__init__("video")
+        self.create_socket(zmq_context, zmq.PUB, port, [(zmq.CONFLATE,1)])
+    
+    def start_video_rcv(self):
+        """
+        Prepares arguments for retrieving video images from Pepper and starts video loop on a separate thread.
+        """
+        if not state.qi_session:
+            logging.info("No Qi session available. Not starting video loop.")
+            return
+
+        video = state.qi_session.service("ALVideoDevice")
+
+        camera_index = 0
+        kQVGA = 2
+        kRGB = 11
+        FPS = 15
+        vid_stream_name = video.subscribeCamera("Pepper Video", camera_index, kQVGA, kRGB, FPS)
+        thread = threading.Thread(target=self.video_rcv_loop, args=(video, vid_stream_name))
+        thread.start()
+
+    def video_rcv_loop(self, vid_service, vid_stream_name):
+        """
+        The main loop of retrieving video images from the robot.
+
+        :param vid_service: The video service object that the active Qi session is connected to. 
+        :type vid_service: Object (Qi service object)
+
+        :param vid_stream_name: The name of a camera subscription on the video service object vid_service
+        :type vid_stream_name: String
+        """
+        while not state.exit_event.is_set():
+            try:
+                img = vid_service.getImageRemote(vid_stream_name)
+                #Possibly limit images sent if queuing issues arise
+                self.socket.send(img[6])
+            except:
+                logging.warn("Failed to retrieve video image from robot.")

src/robot_interface/main.py

@@ -0,0 +1,78 @@
+import logging
+logging.basicConfig(level=logging.DEBUG)
+
+import zmq
+
+from robot_interface.endpoints.actuation_receiver import ActuationReceiver
+from robot_interface.endpoints.main_receiver import MainReceiver
+from robot_interface.endpoints.video_sender import VideoSender
+from robot_interface.state import state
+from robot_interface.utils.timeblock import TimeBlock
+
+
+def main_loop(context):
+    """
+    Run the main loop, handling all incoming requests like pings, negotiation, actuation, etc.
+
+    :param context: The ZeroMQ context to use.
+    :type context: zmq.Context
+    """
+    # When creating sockets, remember to add them to the `sockets` list of the state to ensure they're deinitialized
+    main_receiver = MainReceiver(context)
+    state.sockets.append(main_receiver)
+    actuation_receiver = ActuationReceiver(context)
+    state.sockets.append(actuation_receiver)
+
+    video_sender = VideoSender(context)
+    state.sockets.append(video_sender)
+
+    video_sender.start_video_rcv()
+
+    # Sockets that can run on the main thread. These sockets' endpoints should not block for long (say 50 ms at most).
+    receivers = [main_receiver, actuation_receiver]
+
+    poller = zmq.Poller()
+    for receiver in receivers:
+        poller.register(receiver.socket, zmq.POLLIN)
+
+    logging.debug("Starting main loop.")
+
+    while True:
+        if state.exit_event.is_set(): break
+        socks = dict(poller.poll(100))
+
+        for receiver in receivers:
+            if receiver.socket not in socks: continue
+
+            message = receiver.socket.recv_json()
+            if not isinstance(message, dict) or "endpoint" not in message or "data" not in message:
+                logging.error("Received message of unexpected format: {}".format(message))
+                continue
+
+            def overtime_callback(time_ms):
+                logging.warn("Endpoint \"%s\" took too long (%.2f ms) on the main thread.",
+                             message["endpoint"], time_ms)
+
+            with TimeBlock(overtime_callback, 50):
+                response = receiver.handle_message(message)
+
+            if receiver.socket.getsockopt(zmq.TYPE) == zmq.REP:
+                receiver.socket.send_json(response)
+
+
+def main():
+    context = zmq.Context()
+
+    state.initialize()
+
+    try:
+        main_loop(context)
+    except KeyboardInterrupt:
+        logging.info("User interrupted.")
+    finally:
+        state.deinitialize()
+        context.term()
+
+
+if __name__ == "__main__":
+    main()

src/robot_interface/state.py

@@ -2,6 +2,8 @@ import logging
 import signal
 import threading
 
+from robot_interface.utils.qi_utils import get_qi_session
+
 
 class State(object):
     """
@@ -14,6 +16,8 @@ class State(object):
     def __init__(self):
         self.is_initialized = False
         self.exit_event = None
+        self.sockets = []  # type: List[SocketBase]
+        self.qi_session = None  # type: None | ssl.SSLSession
 
     def initialize(self):
         if self.is_initialized:
@@ -27,10 +31,16 @@ class State(object):
         signal.signal(signal.SIGINT, handle_exit)
         signal.signal(signal.SIGTERM, handle_exit)
 
+        self.qi_session = get_qi_session()
+
         self.is_initialized = True
 
     def deinitialize(self):
         if not self.is_initialized: return
+
+        for socket in self.sockets:
+            socket.close()
+
         self.is_initialized = False
 
     def __getattribute__(self, name):

src/robot_interface/utils/qi_utils.py

@@ -0,0 +1,25 @@
+import logging
+import sys
+
+try:
+    import qi
+except ImportError:
+    qi = None
+
+
+def get_qi_session():
+    if qi is None:
+        logging.info("Unable to import qi. Running in stand-alone mode.")
+        return None
+
+    if "--qi-url" not in sys.argv:
+        logging.info("No Qi URL argument given. Running in stand-alone mode.")
+        return None
+
+    try:
+        app = qi.Application()
+        app.start()
+        return app.session
+    except RuntimeError:
+        logging.info("Unable to connect to the robot. Running in stand-alone mode.")
+        return None

src/robot_interface/utils/timeblock.py

@@ -0,0 +1,31 @@
+import time
+
+
+class TimeBlock(object):
+    """
+    A context manager that times the execution of the block it contains. If execution exceeds the
+    limit, or if no limit is given, the callback will be called with the time that the block took.
+    """
+    def __init__(self, callback, limit_ms=None):
+        """
+        :param callback: The callback function that is called when the block of code is over,
+        unless the code block did not exceed the time limit.
+        :type callback: Callable[[float], None]
+
+        :param limit_ms: The number of milliseconds the block of code is allowed to take. If it
+        exceeds this time, or if it's None, the callback function will be called with the time the
+        block took.
+        :type limit_ms: int | None
+        """
+        self.limit_ms = float(limit_ms) if limit_ms is not None else None
+        self.callback = callback
+        self.start = None
+
+    def __enter__(self):
+        self.start = time.time()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        elapsed = (time.time() - self.start) * 1000.0  # ms
+        if self.limit_ms is None or elapsed > self.limit_ms:
+            self.callback(elapsed)

test/unit/test_actuation_receiver.py

@@ -0,0 +1,74 @@
+import sys
+
+import mock
+import pytest
+import zmq
+
+from robot_interface.endpoints.actuation_receiver import ActuationReceiver
+
+
+@pytest.fixture
+def zmq_context():
+    context = zmq.Context()
+    yield context
+
+
+def test_handle_unimplemented_endpoint(zmq_context):
+    receiver = ActuationReceiver(zmq_context)
+    # Should not error
+    receiver.handle_message({
+        "endpoint": "some_endpoint_that_definitely_does_not_exist",
+        "data": None,
+    })
+
+
+def test_speech_message_no_data(zmq_context, mocker):
+    mock_warn = mocker.patch("logging.warn")
+
+    receiver = ActuationReceiver(zmq_context)
+    receiver.handle_message({"endpoint": "actuate/speech", "data": ""})
+
+    mock_warn.assert_called_with(mock.ANY)
+
+
+def test_speech_message_invalid_data(zmq_context, mocker):
+    mock_warn = mocker.patch("logging.warn")
+
+    receiver = ActuationReceiver(zmq_context)
+    receiver.handle_message({"endpoint": "actuate/speech", "data": True})
+
+    mock_warn.assert_called_with(mock.ANY)
+
+
+def test_speech_no_qi(zmq_context, mocker):
+    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
+
+    mock_qi_session = mock.PropertyMock(return_value=None)
+    type(mock_state).qi_session = mock_qi_session
+
+    receiver = ActuationReceiver(zmq_context)
+    receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
+
+    mock_qi_session.assert_called()
+
+
+def test_speech(zmq_context, mocker):
+    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
+
+    mock_qi = mock.Mock()
+    sys.modules["qi"] = mock_qi
+
+    mock_tts_service = mock.Mock()
+    mock_state.qi_session = mock.Mock()
+    mock_state.qi_session.service.return_value = mock_tts_service
+
+    receiver = ActuationReceiver(zmq_context)
+    receiver._tts_service = None
+    receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
+
+    mock_state.qi_session.service.assert_called_once_with("ALTextToSpeech")
+
+    mock_qi.async.assert_called_once()
+    call_args = mock_qi.async.call_args[0]
+    assert call_args[0] == mock_tts_service.say
+    assert call_args[1] == "Some message to speak."

test/unit/test_main_receiver.py

@@ -0,0 +1,79 @@
+import mock
+import pytest
+import zmq
+
+from robot_interface.endpoints.main_receiver import MainReceiver
+
+
+@pytest.fixture
+def zmq_context():
+    context = zmq.Context()
+    yield context
+
+
+def test_handle_ping(zmq_context):
+    receiver = MainReceiver(zmq_context)
+    response = receiver.handle_message({"endpoint": "ping", "data": "pong"})
+
+    assert "endpoint" in response
+    assert response["endpoint"] == "ping"
+    assert "data" in response
+    assert response["data"] == "pong"
+
+
+def test_handle_ping_none(zmq_context):
+    receiver = MainReceiver(zmq_context)
+    response = receiver.handle_message({"endpoint": "ping", "data": None})
+
+    assert "endpoint" in response
+    assert response["endpoint"] == "ping"
+    assert "data" in response
+    assert response["data"] == None
+
+
+@mock.patch("robot_interface.endpoints.main_receiver.state")
+def test_handle_negotiate_ports(mock_state, zmq_context):
+    receiver = MainReceiver(zmq_context)
+    mock_state.sockets = [receiver]
+
+    response = receiver.handle_message({"endpoint": "negotiate/ports", "data": None})
+
+    assert "endpoint" in response
+    assert response["endpoint"] == "negotiate/ports"
+    assert "data" in response
+    assert isinstance(response["data"], list)
+    for port in response["data"]:
+        assert "id" in port
+        assert isinstance(port["id"], str)
+        assert "port" in port
+        assert isinstance(port["port"], int)
+        assert "bind" in port
+        assert isinstance(port["bind"], bool)
+
+    assert any(port["id"] == "main" for port in response["data"])
+
+
+def test_handle_unimplemented_endpoint(zmq_context):
+    receiver = MainReceiver(zmq_context)
+    response = receiver.handle_message({
+        "endpoint": "some_endpoint_that_definitely_does_not_exist",
+        "data": None,
+    })
+
+    assert "endpoint" in response
+    assert response["endpoint"] == "error"
+    assert "data" in response
+    assert isinstance(response["data"], str)
+
+
+def test_handle_unimplemented_negotiation_endpoint(zmq_context):
+    receiver = MainReceiver(zmq_context)
+    response = receiver.handle_message({
+        "endpoint": "negotiate/but_some_subpath_that_definitely_does_not_exist",
+        "data": None,
+    })
+
+    assert "endpoint" in response
+    assert response["endpoint"] == "negotiate/error"
+    assert "data" in response
+    assert isinstance(response["data"], str)

test/unit/test_time_block.py

@@ -0,0 +1,37 @@
+import time
+
+import mock
+
+from robot_interface.utils.timeblock import TimeBlock
+
+
+class AnyFloat(object):
+    def __eq__(self, other):
+        return isinstance(other, float)
+
+
+def test_no_limit():
+    callback = mock.Mock()
+
+    with TimeBlock(callback):
+        pass
+
+    callback.assert_called_once_with(AnyFloat())
+
+
+def test_exceed_limit():
+    callback = mock.Mock()
+
+    with TimeBlock(callback, 0):
+        time.sleep(0.001)
+
+    callback.assert_called_once_with(AnyFloat())
+
+
+def test_within_limit():
+    callback = mock.Mock()
+
+    with TimeBlock(callback, 5):
+        pass
+
+    callback.assert_not_called()