chore: replace print with logging and make robot conditional

All print statements in the main program, and components used by the main program, have been replaced with appropriate logging statements. The connection to the robot now only gets made when it's possible, otherwise only the microphone will be run.

ref: N25B-119

.githooks/commit-msg

@@ -1,16 +0,0 @@
-#!/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

@@ -1,17 +0,0 @@
-#!/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

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

.gitignore

@@ -1,219 +0,0 @@
-# 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,15 +1,5 @@
-# PepperPlus-RI
+## Development environment
-
-The robot interface is a high-level API for controlling the robot. It implements the API as designed: https://utrechtuniversity.youtrack.cloud/articles/N25B-A-14/RI-CB-Communication.
-
-This is an implementation for the Pepper robot, using the Pepper SDK and Python 2.7 as required by the SDK.
-
-
-
-## Installation
-
 ### Linux (or WSL)
-
 Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell: 
 
 ```bash
@@ -34,7 +24,13 @@ python -m virtualenv .venv
 source .venv/bin/activate
 ```
 
-Install the required packages with
+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
 
 ```bash
 pip install -r requirements.txt
@@ -68,67 +64,21 @@ python -c "import qi; print(qi)"
 
 You should now be able to run this project.
 
-### macOS
+### 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/).
+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/).
 
-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:
+Create a virtual environment as described in the Linux section.
 
-```shell
+Then build `portaudio` for x86_64 CPU's.
-curl -OL https://community-static.aldebaran.com/resources/2.5.10/Python%20SDK/pynaoqi-python2.7-2.5.7.1-mac64.tar.gz
+
+Then follow the remaining installation instructions in the Linux section.
+
+## Running
+Assuming you have the virtual environment activated (`source .venv/bin/activate` on Linux) and that you have a virtual robot running on localhost you should be able to run this project by typing
+
+```bash
+python main.py --qi-url tcp://localhost:<port>
 ```
 
-Then resume the steps from above.
+where `<port>` is the port on which your robot is running.
-
-
-
-## 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

@@ -0,0 +1,69 @@
+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,5 +1,2 @@
 pyzmq<16
 pyaudio<=0.2.11
-pytest<5
-pytest-mock<3.0.0
-pytest-cov<3.0.0

src/audio_streaming.py

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

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

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

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

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

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

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

@@ -1,95 +0,0 @@
-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/utils/microphone.py

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

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

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

state.py

@@ -2,8 +2,6 @@ import logging
 import signal
 import threading
 
-from robot_interface.utils.qi_utils import get_qi_session
-
 
 class State(object):
     """
@@ -16,9 +14,6 @@ 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:
@@ -32,16 +27,10 @@ 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):

test/common/microphone_utils.py

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

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

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

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

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

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

@@ -1,37 +0,0 @@
-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()