pepperplus/pepperplus-ri
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev' into feat/ri2cb-robot-connections

Browse Source
This commit is contained in:
Björn Otgaar
2025-10-22 11:48:21 +02:00
parent da14a67791 4da83a0a7e
commit 8990af88fb
22 changed files with 1039 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
.githooks/commit-msg Normal file
View File

@@ -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

17
.githooks/pre-commit Normal file
View File

@@ -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

9
.githooks/prepare-commit-msg Normal file
View File

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

219
.gitignore vendored Normal file
View File

@@ -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

137
README.md
View File

@@ -2,6 +2,7 @@
### Linux (or WSL) ### Linux (or WSL)
Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell: Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell:
<<<<<<< HEAD
```bash ```bash
pyenv install 2.7 pyenv install 2.7
pyenv shell 2.7 pyenv shell 2.7
@@ -81,4 +82,138 @@ Assuming you have the virtual environment activated (`source .venv/bin/activate`
python main.py --qi-url tcp://localhost:<port> python main.py --qi-url tcp://localhost:<port>
``` ```
where `<port>` is the port on which your robot is running. where `<port>` is the port on which your robot is running.
=======
The robot interface is a high-level API for controlling the robot. It implements the API as designed: https://utrechtuniversity.youtrack.cloud/articles/N25B-A-14/RI-CB-Communication.
This is an implementation for the Pepper robot, using the Pepper SDK and Python 2.7 as required by the SDK.
## Installation
### Linux (or WSL)
Start off by installing [Pyenv](https://github.com/pyenv/pyenv?tab=readme-ov-file#installation) and walk through the steps outlined there (be sure to also add it to PATH). Also install the [Python build requirements](https://github.com/pyenv/pyenv/wiki#suggested-build-environment). Afterwards, install Python 2.7 and activate it for your current shell:
```bash
pyenv install 2.7
pyenv shell 2.7
```
You can check that this worked by typing
```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
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
>>>>>>> origin/dev

9
requirements.txt
View File

@@ -1,2 +1,9 @@
pyzmq<16 pyzmq<16
pyaudio<=0.2.11 <<<<<<< HEAD
pyaudio<=0.2.11
=======
pyaudio<=0.2.11
pytest<5
pytest-mock<3.0.0
pytest-cov<3.0.0
>>>>>>> origin/dev

0
src/robot_interface/__init__.py Normal file
View File

0
src/robot_interface/endpoints/__init__.py Normal file
View File

49
src/robot_interface/endpoints/actuation_receiver.py Normal file
View File

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

56
src/robot_interface/endpoints/main_receiver.py Normal file
View File

@@ -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."}

21
src/robot_interface/endpoints/receiver_base.py Normal file
View File

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

72
src/robot_interface/endpoints/socket_base.py Normal file
View File

@@ -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
}

49
src/robot_interface/endpoints/video_sender.py Normal file
View File

@@ -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.")

78
src/robot_interface/main.py Normal file
View File

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

63
src/robot_interface/state.py Normal file
View File

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

0
src/robot_interface/utils/__init__.py Normal file
View File

25
src/robot_interface/utils/qi_utils.py Normal file
View File

@@ -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

31
src/robot_interface/utils/timeblock.py Normal file
View File

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

0
test/unit/__init__.py Normal file
View File

74
test/unit/test_actuation_receiver.py Normal file
View File

@@ -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."

79
test/unit/test_main_receiver.py Normal file
View File

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

37
test/unit/test_time_block.py Normal file
View File

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