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,93 +1,134 @@
 # 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.
 
 
-## Getting started
-
-To make it easy for you to get started with GitLab, here's a list of recommended next steps.
-
-Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
-
-## Add your files
-
-- [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files
-- [ ] [Add files using the command line](https://docs.gitlab.com/topics/git/add_files/#add-files-to-a-git-repository) or push an existing Git repository with the following command:
-
-```
-cd existing_repo
-git remote add origin https://git.science.uu.nl/ics/sp/2025/n25b/pepperplus-ri.git
-git branch -M main
-git push -uf origin main
-```
-
-## Integrate with your tools
-
-- [ ] [Set up project integrations](https://git.science.uu.nl/ics/sp/2025/n25b/pepperplus-ri/-/settings/integrations)
-
-## Collaborate with your team
-
-- [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/)
-- [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html)
-- [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically)
-- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/)
-- [ ] [Set auto-merge](https://docs.gitlab.com/user/project/merge_requests/auto_merge/)
-
-## Test and Deploy
-
-Use the built-in continuous integration in GitLab.
-
-- [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/)
-- [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/ee/user/application_security/sast/)
-- [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html)
-- [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/)
-- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html)
-
-***
-
-# Editing this README
-
-When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
-
-## Suggestions for a good README
-
-Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
-
-## Name
-Choose a self-explaining name for your project.
-
-## Description
-Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
-
-## Badges
-On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
-
-## Visuals
-Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
 
 ## Installation
-Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
+
+### Linux (or WSL)
+
+Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell:
+
+```bash
+pyenv install 2.7
+pyenv shell 2.7
+```
+
+You can check that this worked by typing
+
+```bash
+python -V
+```
+
+Which should return `Python 2.7.18`.
+
+Next, `cd` into this repository and create (and activate) a virtual environment:
+
+```bash
+cd <path to project>/
+python -m pip install virtualenv
+python -m virtualenv .venv
+source .venv/bin/activate
+```
+
+Install the required packages with
+
+```bash
+pip install -r requirements.txt
+```
+
+Now we need to install the NaoQi SDK into our virtual environment, which we need to do manually. Begin by downloading the SDK:
+
+```bash
+wget https://community-static.aldebaran.com/resources/2.5.10/Python%20SDK/pynaoqi-python2.7-2.5.7.1-linux64.tar.gz
+```
+
+Next, move into the `site-packages` directory and extract the file you just downloaded:
+
+```bash
+cd .venv/lib/python2.7/site-packages/
+tar xvfz <path to SDK>/pynaoqi-python2.7-2.5.7.1-linux64.tar.gz
+rm <path to SDK>/pynaoqi-python2.7-2.5.7.1-linux64.tar.gz
+```
+
+Lastly, we need to inform our virtual environment where to find our newly installed package:
+
+```bash
+echo <path to project>/.venv/lib/python2.7/site-packages/pynaoqi-python2.7-2.5.7.1-linux64/lib/python2.7/site-packages/ > pynaoqi-python2.7.pth
+```
+
+That's it! Verify that it works with
+
+```bash
+python -c "import qi; print(qi)"
+```
+
+You should now be able to run this project.
+
+### macOS
+
+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 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
+```
+
+Then resume the steps from above.
+
+
 
 ## Usage
-Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
 
-## Support
-Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
+On Linux and macOS:
 
-## Roadmap
-If you have ideas for releases in the future, it is a good idea to list them in the README.
+```shell
+PYTHONPATH=src python -m robot_interface.main
+```
 
-## Contributing
-State if you are open to contributions and what your requirements are for accepting them.
+On Windows:
 
-For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
+```shell
+$env:PYTHONPATH="src"; python -m robot_interface.main
+```
 
-You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
+With both, if you want to connect to the actual robot (or simulator), pass the `--qi-url` argument.
 
-## Authors and acknowledgment
-Show your appreciation to those who have contributed to the project.
 
-## License
-For open source projects, say how it is licensed.
 
-## Project status
-If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.
+## 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

requirements.txt

@@ -0,0 +1,5 @@
+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,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

@@ -0,0 +1,64 @@
+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
+        self.is_speaking = False  # type: Boolean
+
+    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,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()