build: add qi lib to Docker

Uses Pyenv to build Python 2 from source, while keeping Debian for
needed packages. Might move to two build stages in the future.

ref: N25B-284

.dockerignore

@@ -0,0 +1,9 @@
+.git
+.githooks/
+.idea/
+.venv/
+test/
+typings/
+.dockerignore
+.gitignore
+README.md

.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
+

Dockerfile

@@ -0,0 +1,37 @@
+FROM debian:trixie AS build
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN apt-get update; apt-get install -y portaudio19-dev libzmq3-dev make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev curl git libncursesw5-dev xz-utils tk-dev libxml2-dev libxmlsec1-dev libffi-dev liblzma-dev alsa-utils musl-dev
+
+ENV HOME="/root"
+
+RUN git clone --depth=1 https://github.com/pyenv/pyenv.git ${HOME}/.pyenv
+ENV PYENV_ROOT="${HOME}/.pyenv"
+ENV PATH="${PYENV_ROOT}/shims:${PYENV_ROOT}/bin:${PATH}"
+
+ENV PYTHON_VERSION=2.7.18
+RUN pyenv install ${PYTHON_VERSION}; pyenv global ${PYTHON_VERSION}
+
+RUN python -m pip install virtualenv; python -m virtualenv .venv
+RUN /usr/bin/env bash -c 'source .venv/bin/activate && pip install -r ./requirements.txt'
+# RUN eval "$(pyenv init - bash)"; pyenv install 2.7; pyenv shell 2.7; python -m pip install virtualenv; python -m virtualenv .venv; source .venv/bin/activate; pip install -r requirements.txt
+
+# FROM debian:trixie
+#
+# WORKDIR /app
+#
+# COPY --from=build /app/.venv /app/.venv
+
+WORKDIR /app/.venv/lib/python2.7/site-packages
+RUN /usr/bin/env bash -c 'apt-get install -y wget && wget https://community-static.aldebaran.com/resources/2.5.10/Python%20SDK/pynaoqi-python2.7-2.5.7.1-linux64.tar.gz && tar xvfz ./pynaoqi-python2.7-2.5.7.1-linux64.tar.gz && rm pynaoqi-python2.7-2.5.7.1-linux64.tar.gz'
+RUN echo /app/.venv/lib/python2.7/site-packages/pynaoqi-python2.7-2.5.7.1-linux64/lib/python2.7/site-packages/ > pynaoqi-python2.7.pth
+WORKDIR /app
+
+COPY . .
+
+ENV PYTHONPATH=src
+
+CMD [ "/bin/bash", "-c", "source .venv/bin/activate && python -m robot_interface.main --qi-url tcp://172.17.0.1:43305" ]

README.md

@@ -1,13 +1,23 @@
-## Development environment
+# PepperPlus-RI
+
+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: 
+
+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,6 +34,18 @@ python -m virtualenv .venv
 source .venv/bin/activate
 ```
 
+We depend on PortAudio for the `pyaudio` package, so install it with:
+
+```bash
+sudo apt install -y portaudio19-dev
+```
+
+On WSL, also install:
+
+```bash
+sudo apt install -y libasound2-plugins
+```
+
 Install the required packages with
 
 ```bash
@@ -58,14 +80,69 @@ python -c "import qi; print(qi)"
 
 You should now be able to run this project.
 
-### MacOS
-...
+### macOS
 
-## 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
+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/).
 
-```bash
-python main.py --qi-url tcp://localhost:<port>
+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:
+
+```shell
+curl -OL https://community-static.aldebaran.com/resources/2.5.10/Python%20SDK/pynaoqi-python2.7-2.5.7.1-mac64.tar.gz
 ```
 
-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.
+
+There's also a `--microphone` argument that can be used to choose a microphone to use. If not given, the program will try the default microphone. If you don't know the name of the microphone, pass the argument with any value, and it will list the names of available microphones.
+
+
+
+## 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

asound.conf

@@ -0,0 +1,9 @@
+pcm.!default {
+  type hw
+  card 2
+}
+
+ctl.!default {
+  type hw
+  card 2
+}

install_deps.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env sh
+apk add portaudio-dev libzmq gcc musl-dev g++ alsa-utils
+
+pip install -r requirements.txt

main.py

@@ -1,27 +0,0 @@
-import qi
-import zmq
-import time
-
-
-def say(session, message):
-    tts = session.service("ALTextToSpeech")
-    tts.say(message)
-
-if __name__ == "__main__":
-    app = qi.Application()
-    app.start()
-    session = app.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
-
-    while True:
-        print("Listening for message")
-        message = socket.recv_string()
-        print("Received message: {}".format(message))
-
-        say(session, message)
-
-        time.sleep(1)

requirements.txt

@@ -1 +1,5 @@
-pyzmq<16
+pyzmq<16
+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,50 @@
+from __future__ import unicode_literals  # So that we can log texts with Unicode characters
+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/audio_sender.py

@@ -0,0 +1,72 @@
+from __future__ import unicode_literals  # So that `logging` can use Unicode characters in names
+import threading
+import logging
+
+import pyaudio
+import zmq
+
+from robot_interface.endpoints.socket_base import SocketBase
+from robot_interface.state import state
+from robot_interface.utils.microphone import choose_mic
+
+
+logger = logging.getLogger(__name__)
+
+
+class AudioSender(SocketBase):
+    def __init__(self, zmq_context, port=5558):
+        super(AudioSender, self).__init__(str("audio"))  # Convert future's unicode_literal to str
+        self.create_socket(zmq_context, zmq.PUB, port)
+        self.thread = None
+
+        try:
+            self.audio = pyaudio.PyAudio()
+            self.microphone = choose_mic(self.audio)
+        except IOError as e:
+            logger.warning("PyAudio is not available.", exc_info=e)
+            self.audio = None
+            self.microphone = None
+
+    def start(self):
+        """
+        Start sending audio in a different thread.
+        """
+        if not self.microphone:
+            logger.info("Not listening: no microphone available.")
+            return
+
+        logger.info("Listening with microphone \"{}\".".format(self.microphone["name"]))
+        self.thread = threading.Thread(target=self._stream)
+        self.thread.start()
+
+    def wait_until_done(self):
+        """
+        Wait until the audio thread is done. Will only be done if `state.exit_event` is set, so
+        make sure to set that before calling this method or it will block.
+        """
+        if not self.thread: return
+        self.thread.join()
+        self.thread = None
+
+    def _stream(self):
+        chunk = 512  # 320 at 16000 Hz is 20ms, 512 is required for Silero-VAD
+
+        # Docs say this only raises an error if neither `input` nor `output` is True
+        stream = self.audio.open(
+            format=pyaudio.paFloat32,
+            channels=1,
+            rate=16000,
+            input=True,
+            input_device_index=self.microphone["index"],
+            frames_per_buffer=chunk,
+        )
+
+        try:
+            while not state.exit_event.is_set():
+                data = stream.read(chunk)
+                self.socket.send(data)
+        except IOError as e:
+            logger.error("Stopped listening: failed to get audio from microphone.", exc_info=e)
+        finally:
+            stream.stop_stream()
+            stream.close()

src/robot_interface/endpoints/main_receiver.py

@@ -0,0 +1,59 @@
+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,69 @@
+from abc import ABCMeta
+import os
+
+
+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=False):
+        """
+        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
+        host = os.environ.get("CB_HOST", "localhost")
+        if bind:
+            self.socket.bind("tcp://{}:{}".format(host, port))
+        else:
+            self.socket.connect("tcp://{}:{}".format(host, 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,48 @@
+import zmq
+import threading
+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,84 @@
+import logging
+
+from robot_interface.endpoints.audio_sender import AudioSender
+
+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)
+    audio_sender = AudioSender(context)
+    state.sockets.append(audio_sender)
+
+    video_sender.start_video_rcv()
+    audio_sender.start()
+
+    # 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

@@ -0,0 +1,63 @@
+import logging
+import signal
+import threading
+
+from robot_interface.utils.qi_utils import get_qi_session
+
+
+class State(object):
+    """
+    Do not create an instance of this class directly: use the instance `state` below. This state must be initiated once,
+    probably when your program starts.
+
+    This class is used to share state between threads. For example, when the program is quit, that all threads can
+    detect this via the `exit_event` property being set.
+    """
+    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:
+            logging.warn("Already initialized")
+            return
+
+        self.exit_event = threading.Event()
+        def handle_exit(_, __):
+            logging.info("Exiting.")
+            self.exit_event.set()
+        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):
+        # Enforce that the state is initialized before accessing any property (aside from the basic ones)
+        if name in ("initialize", "deinitialize", "is_initialized", "__dict__", "__class__"):
+            return object.__getattribute__(self, name)
+
+        if not object.__getattribute__(self, "is_initialized"):
+            # Special case for the exit_event: if the event is set, return it without an error
+            if name == "exit_event":
+                exit_event = object.__getattribute__(self, "exit_event")
+                if exit_event and exit_event.is_set(): return exit_event
+
+            raise RuntimeError("State must be initialized before accessing '%s'" % name)
+
+        return object.__getattribute__(self, name)
+
+
+# Must call `.initialize` before use
+state = State()

src/robot_interface/utils/microphone.py

@@ -0,0 +1,120 @@
+from __future__ import unicode_literals  # So that `print` can print Unicode characters in names
+import logging
+import sys
+
+logger = logging.getLogger(__name__)
+
+
+def get_microphones(audio):
+    """
+    Get audio devices which have input channels.
+
+    :param audio: An instance of PyAudio to use.
+    :type audio: pyaudio.PyAudio
+
+    :return: An interator of PaAudio dicts containing information about the microphone devices.
+    :rtype: Iterator[dict]
+    """
+    for i in range(audio.get_device_count()):
+        device = audio.get_device_info_by_index(i)
+        if device["maxInputChannels"] > 0:
+            yield device
+
+
+def choose_mic_interactive(audio):
+    """
+    Choose a microphone to use, interactively in the CLI.
+
+    :param audio: An instance of PyAudio to use.
+    :type audio: pyaudio.PyAudio
+
+    :return: A dictionary from PyAudio containing information about the microphone to use, or None
+    if there is no microphone.
+    :rtype: dict | None
+    """
+    microphones = list(get_microphones(audio))
+    if len(microphones) == 0: return None
+
+    print("Found {} microphones:".format(len(microphones)))
+    for i, mic in enumerate(microphones):
+        print("- {}: {}".format(i, mic["name"]))
+
+    chosen_microphone = None
+    while chosen_microphone is None:
+        chosen = raw_input("Which device would you like to use?\n> ")
+        try:
+            chosen = int(chosen)
+            if chosen < 0 or chosen >= len(microphones): raise ValueError()
+            chosen_microphone = microphones[chosen]
+        except ValueError:
+            print("Please enter a number between 0 and {}".format(len(microphones)-1))
+
+    logger.info("Chose microphone \"{}\"".format(chosen_microphone["name"]))
+    return chosen_microphone
+
+
+def choose_mic_default(audio):
+    """
+    Get the system's default microphone to use.
+
+    :param audio: An instance of PyAudio to use.
+    :type audio: pyaudio.PyAudio
+
+    :return: A dictionary from PyAudio containing information about the microphone to use, or None
+    if there is no microphone.
+    :rtype: dict | None
+    """
+    try:
+        return audio.get_default_input_device_info()
+    except IOError:
+        return None
+
+
+def choose_mic_arguments(audio):
+    """
+    Get a microphone to use from command line arguments.
+
+    :param audio: An instance of PyAudio to use.
+    :type audio: pyaudio.PyAudio
+
+    :return: A dictionary from PyAudio containing information about the microphone to use, or None
+    if there is no microphone satisfied by the arguments.
+    :rtype: dict | None
+    """
+    microphone_name = None
+    for i, arg in enumerate(sys.argv):
+        if arg == "--microphone" and len(sys.argv) > i+1:
+            microphone_name = sys.argv[i+1].strip()
+        if arg.startswith("--microphone="):
+            microphone_name = arg[13:].strip()
+
+    if not microphone_name: return None
+
+    available_mics = list(get_microphones(audio))
+    for mic in available_mics:
+        if mic["name"] == microphone_name:
+            return mic
+
+    available_mic_names = [mic["name"] for mic in available_mics]
+    logger.warning("Microphone \"{}\" not found. Choose one of {}"
+                   .format(microphone_name, available_mic_names))
+
+    return None
+
+
+def choose_mic(audio):
+    """
+    Get a microphone to use. Firstly, tries to see if there's an application argument specifying the
+    microphone to use. If not, get the default microphone.
+
+    :param audio: An instance of PyAudio to use.
+    :type audio: pyaudio.PyAudio
+
+    :return: A dictionary from PyAudio containing information about the microphone to use, or None
+    if there is no microphone.
+    :rtype: dict | None
+    """
+    chosen_mic = choose_mic_arguments(audio)
+    if chosen_mic: return chosen_mic
+
+    return choose_mic_default(audio)

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/common/microphone_utils.py

@@ -0,0 +1,152 @@
+from __future__ import unicode_literals  # So that we can format strings with Unicode characters
+import random
+import sys
+from StringIO import StringIO
+
+from robot_interface.utils.microphone import (
+    choose_mic_default,
+    choose_mic_interactive,
+    choose_mic_arguments,
+    choose_mic,
+    get_microphones,
+)
+
+
+class MicrophoneUtils(object):
+    """Shared tests for any PyAudio-like implementation, e.g. mock and real."""
+
+    def test_choose_mic_default(self, pyaudio_instance):
+        """
+        The result must contain at least "index", as this is used to identify the microphone.
+        The "name" is used for logging, so it should also exist.
+        It must have one or more channels.
+        Lastly it must be capable of sending at least 16000 samples per second.
+        """
+        result = choose_mic_default(pyaudio_instance)
+        assert "index" in result
+        assert isinstance(result["index"], (int, long))
+
+        assert "name" in result
+        assert isinstance(result["name"], (str, unicode))
+
+        assert "maxInputChannels" in result
+        assert isinstance(result["maxInputChannels"], (int, long))
+        assert result["maxInputChannels"] > 0
+
+        assert "defaultSampleRate" in result
+        assert isinstance(result["defaultSampleRate"], float)
+        assert result["defaultSampleRate"] >= 16000
+
+    def test_choose_mic_interactive_input_not_int(self, pyaudio_instance, mocker):
+        """
+        First mock an input that's not an integer, then a valid integer. There should be no errors.
+        """
+        mock_input = mocker.patch("__builtin__.raw_input", side_effect=["not an integer", "0"])
+        fake_out = StringIO()
+        mocker.patch.object(sys, "stdout", fake_out)
+
+        result = choose_mic_interactive(pyaudio_instance)
+        assert "index" in result
+        assert isinstance(result["index"], (int, long))
+        assert result["index"] == 0
+
+        assert mock_input.called
+
+        assert any(p.startswith("Please enter a number") for p in fake_out.getvalue().splitlines())
+
+    def test_choose_mic_interactive_negative_index(self, pyaudio_instance, mocker):
+        """
+        Make sure that the interactive method does not allow negative integers as input.
+        """
+        mock_input = mocker.patch("__builtin__.raw_input", side_effect=["-1", "0"])
+        fake_out = StringIO()
+        mocker.patch.object(sys, "stdout", fake_out)
+
+        result = choose_mic_interactive(pyaudio_instance)
+        assert "index" in result
+        assert isinstance(result["index"], (int, long))
+        assert result["index"] == 0
+
+        assert mock_input.called
+
+        assert any(p.startswith("Please enter a number") for p in fake_out.getvalue().splitlines())
+
+    def test_choose_mic_interactive_index_too_high(self, pyaudio_instance, mocker):
+        """
+        Make sure that the interactive method does not allow indices higher than the highest mic index.
+        """
+        real_count = len(list(get_microphones(pyaudio_instance)))
+        mock_input = mocker.patch("__builtin__.raw_input", side_effect=[str(real_count), "0"])
+        fake_out = StringIO()
+        mocker.patch.object(sys, "stdout", fake_out)
+
+        result = choose_mic_interactive(pyaudio_instance)
+        assert "index" in result
+        assert isinstance(result["index"], (int, long))
+
+        assert mock_input.called
+
+        assert any(p.startswith("Please enter a number") for p in fake_out.getvalue().splitlines())
+
+    def test_choose_mic_interactive_random_index(self, pyaudio_instance, mocker):
+        """
+        Get a random index from the list of available mics, make sure it's correct.
+        """
+        microphones = list(get_microphones(pyaudio_instance))
+        random_index = random.randrange(len(microphones))
+        mocker.patch("__builtin__.raw_input", side_effect=[str(random_index)])
+
+        result = choose_mic_interactive(pyaudio_instance)
+        assert "index" in result
+        assert isinstance(result["index"], (int, long))
+        assert result["index"] == microphones[random_index]["index"]
+
+    def test_choose_mic_no_arguments(self, pyaudio_instance, mocker):
+        mocker.patch.object(sys, "argv", [])
+
+        result = choose_mic_arguments(pyaudio_instance)
+
+        assert result is None
+
+    def test_choose_mic_arguments(self, pyaudio_instance, mocker):
+        for mic in get_microphones(pyaudio_instance):
+            mocker.patch.object(sys, "argv", ["--microphone", mic["name"]])
+
+            result = choose_mic_arguments(pyaudio_instance)
+
+            assert result is not None
+            assert result == mic
+
+    def test_choose_mic_arguments_eq(self, pyaudio_instance, mocker):
+        for mic in get_microphones(pyaudio_instance):
+            mocker.patch.object(sys, "argv", ["--microphone={}".format(mic["name"])])
+
+            result = choose_mic_arguments(pyaudio_instance)
+
+            assert result is not None
+            assert result == mic
+
+    def test_choose_mic_arguments_not_exits(self, pyaudio_instance, mocker):
+        mocker.patch.object(sys, "argv", ["--microphone", "Surely this microphone doesn't exist"])
+
+        result = choose_mic_arguments(pyaudio_instance)
+
+        assert result is None
+
+    def test_choose_mic_with_argument(self, pyaudio_instance, mocker):
+        mic = next(get_microphones(pyaudio_instance))
+        mocker.patch.object(sys, "argv", ["--microphone", mic["name"]])
+
+        result = choose_mic(pyaudio_instance)
+
+        assert result is not None
+        assert result == mic
+
+    def test_choose_mic_no_argument(self, pyaudio_instance, mocker):
+        default_mic = choose_mic_default(pyaudio_instance)
+        mocker.patch.object(sys, "argv", [])
+
+        result = choose_mic(pyaudio_instance)
+
+        assert result is not None
+        assert result == default_mic

test/integration/test_microphone_utils.py

@@ -0,0 +1,20 @@
+import pyaudio
+
+import pytest
+
+from common.microphone_utils import MicrophoneUtils
+
+
+@pytest.fixture
+def pyaudio_instance():
+    audio = pyaudio.PyAudio()
+    try:
+        audio.get_default_input_device_info()
+        return audio
+    except IOError:
+        pytest.skip("No microphone available to test with.")
+
+
+class TestAudioIntegration(MicrophoneUtils):
+    """Run shared audio behavior tests with the mock implementation."""
+    pass

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_audio_sender.py

@@ -0,0 +1,104 @@
+# coding=utf-8
+import os
+import time
+
+import mock
+import pytest
+import zmq
+
+from robot_interface.endpoints.audio_sender import AudioSender
+
+
+@pytest.fixture
+def zmq_context():
+    context = zmq.Context()
+    yield context
+
+
+def test_no_microphone(zmq_context, mocker):
+    mock_info_logger = mocker.patch("robot_interface.endpoints.audio_sender.logger.info")
+    mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
+    mock_choose_mic.return_value = None
+
+    sender = AudioSender(zmq_context)
+    assert sender.microphone is None
+
+    sender.start()
+    assert sender.thread is None
+    mock_info_logger.assert_called()
+
+    sender.wait_until_done()  # Should return early because we didn't start a thread
+
+
+def test_unicode_mic_name(zmq_context, mocker):
+    mocker.patch("robot_interface.endpoints.audio_sender.threading")
+    mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
+    mock_choose_mic.return_value = {"name": u"• Some Unicode name"}
+
+    sender = AudioSender(zmq_context)
+    assert sender.microphone is not None
+
+    # `.start()` logs the name of the microphone. It should not give an error if it contains Unicode
+    #  symbols.
+    sender.start()
+    assert sender.thread is not None
+
+    sender.wait_until_done()  # Should return instantly because we didn't start a real thread
+
+
+def _fake_read(num_frames):
+    return os.urandom(num_frames * 4)
+
+
+def test_sending_audio(mocker):
+    mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
+    mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
+
+    mock_state = mocker.patch("robot_interface.endpoints.audio_sender.state")
+    mock_state.exit_event.is_set.side_effect = [False, True]
+
+    mock_zmq_context = mock.Mock()
+    send_socket = mock.Mock()
+
+    # If there's something wrong with the microphone, it will raise an IOError when `read`ing.
+    stream = mock.Mock()
+    stream.read = _fake_read
+
+    sender = AudioSender(mock_zmq_context)
+    sender.socket.send = send_socket
+    sender.audio.open = mock.Mock()
+    sender.audio.open.return_value = stream
+
+    sender.start()
+    sender.wait_until_done()
+
+    send_socket.assert_called()
+
+
+def _fake_read_error(num_frames):
+    raise IOError()
+
+
+def test_break_microphone(mocker):
+    mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
+    mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
+
+    mock_state = mocker.patch("robot_interface.endpoints.audio_sender.state")
+    mock_state.exit_event.is_set.side_effect = [False, True]
+
+    mock_zmq_context = mock.Mock()
+    send_socket = mock.Mock()
+
+    # If there's something wrong with the microphone, it will raise an IOError when `read`ing.
+    stream = mock.Mock()
+    stream.read = _fake_read_error
+
+    sender = AudioSender(mock_zmq_context)
+    sender.socket.send = send_socket
+    sender.audio.open = mock.Mock()
+    sender.audio.open.return_value = stream
+
+    sender.start()
+    sender.wait_until_done()
+
+    send_socket.assert_not_called()

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_microphone_utils.py

@@ -0,0 +1,85 @@
+# coding=utf-8
+import mock
+import pytest
+
+from common.microphone_utils import MicrophoneUtils
+from robot_interface.utils.microphone import choose_mic_default, choose_mic_interactive
+
+
+class MockPyAudio:
+    def __init__(self):
+        # You can predefine fake device info here
+        self.devices = [
+            {
+                "index": 0,
+                "name": u"Someone’s Microphone",  # Using a Unicode ’ character
+                "maxInputChannels": 2,
+                "maxOutputChannels": 0,
+                "defaultSampleRate": 44100.0,
+                "defaultLowInputLatency": 0.01,
+                "defaultLowOutputLatency": 0.01,
+                "defaultHighInputLatency": 0.1,
+                "defaultHighOutputLatency": 0.1,
+                "hostApi": 0,
+            },
+            {
+                "index": 1,
+                "name": u"Mock Speaker 1",
+                "maxInputChannels": 0,
+                "maxOutputChannels": 2,
+                "defaultSampleRate": 48000.0,
+                "defaultLowInputLatency": 0.01,
+                "defaultLowOutputLatency": 0.01,
+                "defaultHighInputLatency": 0.1,
+                "defaultHighOutputLatency": 0.1,
+                "hostApi": 0,
+            },
+        ]
+
+    def get_device_count(self):
+        """Return the number of available mock devices."""
+        return len(self.devices)
+
+    def get_device_info_by_index(self, index):
+        """Return information for a given mock device index."""
+        if 0 <= index < len(self.devices):
+            return self.devices[index]
+        else:
+            raise IOError("Invalid device index: {}".format(index))
+
+    def get_default_input_device_info(self):
+        """Return info for a default mock input device."""
+        for device in self.devices:
+            if device.get("maxInputChannels", 0) > 0:
+                return device
+        raise IOError("No default input device found")
+
+
+@pytest.fixture
+def pyaudio_instance():
+    return MockPyAudio()
+
+
+def _raise_io_error():
+    raise IOError()
+
+
+class TestAudioUnit(MicrophoneUtils):
+    """Run shared audio behavior tests with the mock implementation."""
+    def test_choose_mic_default_no_mic(self):
+        mock_pyaudio = mock.Mock()
+        mock_pyaudio.get_device_count = mock.Mock(return_value=0L)
+        mock_pyaudio.get_default_input_device_info = _raise_io_error
+
+        result = choose_mic_default(mock_pyaudio)
+
+        assert result is None
+
+    def test_choose_mic_interactive_no_mic(self):
+        mock_pyaudio = mock.Mock()
+        mock_pyaudio.get_device_count = mock.Mock(return_value=0L)
+        mock_pyaudio.get_default_input_device_info = _raise_io_error
+
+        result = choose_mic_interactive(mock_pyaudio)
+
+        assert result is None

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()