feat: ignore own speech

When we detect that Pepper is talking we stop sending audio.

ref: N25B-214

.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

@@ -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,13 +34,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:
-
-```shell
-sudo apt install portaudio19-dev
-```
-
-Then you can install the required packages with
+Install the required packages with
 
 ```bash
 pip install -r requirements.txt
@@ -64,21 +68,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.
-
-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>
+```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.
+
+
+
+## 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/audio_streaming.py

@@ -1,93 +0,0 @@
-import threading
-
-import pyaudio
-import zmq
-
-from state import state
-
-
-def choose_mic_interactive(audio):
-    """Choose a microphone to use. The `audio` parameter is an instance of PyAudio. Returns a dict."""
-    device_count = audio.get_device_count()
-    print("Found {} audio devices:".format(device_count))
-    for i in range(device_count):
-        print("- {}: {}".format(i, audio.get_device_info_by_index(i)["name"]))
-
-    microphone_index = None
-    while microphone_index is None:
-        chosen = input("Which device would you like to use?\n> ")
-        try:
-            chosen = int(chosen)
-            if chosen < 0 or chosen > device_count: raise ValueError()
-            microphone_index = chosen
-        except ValueError:
-            print("Please enter a number between 0 and {}".format(device_count))
-
-    chosen_microphone = audio.get_device_info_by_index(microphone_index)
-    print("Chose microphone \"{}\"".format(chosen_microphone["name"]))
-    return chosen_microphone
-
-
-def choose_mic_default(audio):
-    """Choose a microphone to use based on defaults. The `audio` parameter is a PyAudio. Returns a dict."""
-    default_device = audio.get_default_input_device_info()
-    return default_device
-
-
-class AudioStreaming:
-    def __init__(self, port=5557):
-        self.port = port
-        self.audio = pyaudio.PyAudio()
-        self.microphone = choose_mic_default(self.audio)
-        self.thread = None
-
-    def run(self):
-        self.thread = threading.Thread(target=self._stream)
-        self.thread.start()
-
-    def wait_until_done(self):
-        if not self.thread: return
-        self.thread.join()
-
-    def _stream(self):
-        context = zmq.Context()
-        socket = context.socket(zmq.PUB)
-        socket.bind("tcp://*:{}".format(self.port))
-
-        chunk = 512  # 320 at 16000 Hz is 20ms, 512 is required for Silero-VAD
-
-        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)
-                socket.send(data)
-        finally:
-            stream.stop_stream()
-            stream.close()
-
-
-if __name__ == "__main__":
-    state.initialize()
-    try:
-        audio = AudioStreaming()
-        print("Starting audio streaming...")
-        audio.run()
-
-        import time
-        end = time.time() + 10
-        while not state.exit_event.is_set() and time.time() < end:
-            print "\rExiting in {:.2f} seconds".format(end - time.time()),
-            time.sleep(0.05)
-
-        state.exit_event.set()
-        audio.wait_until_done()
-    finally:
-        state.deinitialize()

src/robot_interface/endpoints/actuation_receiver.py

@@ -0,0 +1,66 @@
+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
+        self._al_memory = 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")
+        if not self._al_memory:
+            self._al_memory = state.qi_session.service("ALMemory")
+
+            # Subscribe to speech end event
+            self.status_subscriber = self._al_memory.subscriber("ALTextToSpeech/Status")  # self because garbage collect
+            self.status_subscriber.signal.connect(self._on_status_changed)
+
+        # Returns instantly. Messages received while speaking will be queued.
+        qi.async(self._tts_service.say, text)
+
+    @staticmethod
+    def _on_status_changed(value):  # value will contain either 'enqueued', 'started' or 'done' depending on the status
+        """Callback function for when the speaking status changes. Will change the is_speaking value of the state."""
+        if "started" in value:
+            logging.debug("Started speaking.")
+            state.is_speaking = True
+        if "done" in value:
+            logging.debug("Done speaking.")
+            state.is_speaking = False
+
+    def handle_message(self, message):
+        if message["endpoint"] == "actuate/speech":
+            self._handle_speech(message)

src/robot_interface/endpoints/audio_sender.py

@@ -0,0 +1,73 @@
+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_default
+
+
+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.audio = pyaudio.PyAudio()
+        self.microphone = choose_mic_default(self.audio)
+        self.thread = 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():
+                # Don't send audio if Pepper is speaking
+                if state.is_speaking:
+                    if stream.is_active(): stream.stop_stream()
+                    continue
+
+                if stream.is_stopped(): stream.start_stream()
+
+                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,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,95 @@
+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.")
+
+    import schedule
+    test_speaking_message = {"data": "Hi, my name is Pepper, and this is quite a long message."}
+    def test_speak():
+        logging.debug("Testing speech.")
+        actuation_receiver._handle_speech(test_speaking_message)
+
+    schedule.every(10).seconds.do(test_speak)
+
+    while True:
+        if state.exit_event.is_set(): break
+
+        schedule.run_pending()
+
+        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,9 @@ 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
+        self.is_speaking = False  # type: Boolean
 
     def initialize(self):
         if self.is_initialized:
@@ -27,10 +32,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/microphone.py

@@ -0,0 +1,69 @@
+from __future__ import unicode_literals  # So that `print` can print Unicode characters in names
+import logging
+
+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

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,95 @@
+import random
+import sys
+from StringIO import StringIO
+
+from robot_interface.utils.microphone import choose_mic_default, choose_mic_interactive, 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"]

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_default")
+    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_default")
+    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_default")
+    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_default")
+    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()