feat: added basic list of 10 gestures

ref: N25B-402

src/robot_interface/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

src/robot_interface/core/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

src/robot_interface/core/config.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from __future__ import unicode_literals
 
 from robot_interface.utils.get_config import get_config
@@ -68,7 +61,7 @@ class VideoConfig(object):
     ):
         self.camera_index = get_config(camera_index, "VIDEO__CAMERA_INDEX", 0, int)
         self.resolution = get_config(resolution, "VIDEO__RESOLUTION", 2, int)
-        self.color_space = get_config(color_space, "VIDEO__COLOR_SPACE", 13, int)
+        self.color_space = get_config(color_space, "VIDEO__COLOR_SPACE", 11, int)
         self.fps = get_config(fps, "VIDEO__FPS", 15, int)
         self.stream_name = get_config(stream_name, "VIDEO__STREAM_NAME", "Pepper Video")
         self.image_buffer = get_config(image_buffer, "VIDEO__IMAGE_BUFFER", 6, int)

src/robot_interface/endpoints/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

src/robot_interface/endpoints/actuation_receiver.py

@@ -1,19 +1,11 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from __future__ import unicode_literals  # So that we can log texts with Unicode characters
 import logging
-from threading import Thread
 
-import Queue
 import zmq
 
 from robot_interface.endpoints.receiver_base import ReceiverBase
 from robot_interface.state import state
+
 from robot_interface.core.config import settings
 from robot_interface.endpoints.gesture_settings import GestureTags
 
@@ -40,9 +32,6 @@ class ActuationReceiver(ReceiverBase):
         self.socket.setsockopt_string(zmq.SUBSCRIBE, u"")  # Causes block if given in options
         self._tts_service = None
         self._animation_service = None
-        self._message_queue = Queue.Queue()
-        self.message_thread = Thread(target=self._handle_messages)
-        self.message_thread.start()
 
     def _handle_speech(self, message):
         """
@@ -69,26 +58,8 @@ class ActuationReceiver(ReceiverBase):
         if not self._tts_service:
             self._tts_service = state.qi_session.service("ALTextToSpeech")
 
-        if message.get("is_priority"):
+        # Returns instantly. Messages received while speaking will be queued.
-            # Bypass queue and speak immediately
+        getattr(qi, "async")(self._tts_service.say, text)
-            self.clear_queue()
-            self._message_queue.put(text)
-            logging.debug("Force speaking immediately: {}".format(text))
-        else: 
-            self._message_queue.put(text)
-
-    def clear_queue(self):
-        """
-        Safely drains all pending messages from the queue.
-        """
-        logging.info("Message queue size: {}".format(self._message_queue.qsize()))
-        try:
-            while True:
-                # Remove items one by one without waiting
-                self._message_queue.get_nowait()
-        except Queue.Empty:
-            pass
-        logging.info("Message queue cleared.")
 
     def _handle_gesture(self, message, is_single):
         """
@@ -151,19 +122,6 @@ class ActuationReceiver(ReceiverBase):
         if message["endpoint"] == "actuate/gesture/single":
             self._handle_gesture(message, True)
 
-    def _handle_messages(self):
-        while not state.exit_event.is_set():
-            try:
-                text = self._message_queue.get(timeout=0.1)
-                state.is_speaking = True
-                self._tts_service.say(text)
-            except Queue.Empty:
-                state.is_speaking = False
-            except RuntimeError:
-                logging.error("Lost connection to Pepper. Please check if you're connected to the "
-                              "local WiFi and restart this application.")
-                state.exit_event.set()        
-
     def endpoint_description(self):
         """
         Extend the default endpoint description with gesture tags.
@@ -171,5 +129,6 @@ class ActuationReceiver(ReceiverBase):
         """
         desc = super(ActuationReceiver, self).endpoint_description()
         desc["gestures"] = GestureTags.tags
+        desc["basic_gestures"] = GestureTags.basic_gestures
         desc["single_gestures"] = GestureTags.single_gestures
         return desc

src/robot_interface/endpoints/audio_sender.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from __future__ import unicode_literals  # So that `logging` can use Unicode characters in names
 import threading
 import logging
@@ -96,7 +89,6 @@ class AudioSender(SocketBase):
         try:
             while not state.exit_event.is_set():
                 data = stream.read(chunk)
-                if (state.is_speaking): continue  # Do not send audio while the robot is speaking
                 self.socket.send(data)
         except IOError as e:
             logger.error("Stopped listening: failed to get audio from microphone.", exc_info=e)

src/robot_interface/endpoints/gesture_settings.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 class GestureTags:
     tags = ["above", "affirmative", "afford", "agitated", "all", "allright", "alright", "any", 
             "assuage", "assuage", "attemper", "back", "bashful", "beg", "beseech", "blank", 
@@ -21,6 +14,19 @@ class GestureTags:
             "think", "timid", "top", "unless", "up", "upstairs", "void", "warm", "winner", "yeah", 
             "yes", "yoo-hoo", "you", "your", "zero", "zestful"]
     
+    basic_gestures = [
+        "animations/Stand/Gestures/Hey_1",
+        "animations/Stand/Emotions/Neutral/Puzzled_1",
+        "animations/Stand/Gestures/Explain_4",
+        "animations/Stand/Gestures/You_1"
+        "animations/Stand/Emotions/Positive/Happy_1",
+        "animations/Stand/Emotions/Positive/Laugh_2",
+        "animations/Stand/Emotions/Neutral/Lonely_1",
+        "animations/Stand/Emotions/Negative/Surprise_1",
+        "animations/Stand/Emotions/Negative/Hurt_2",
+        "animations/Stand/Emotions/Negative/Angry_4",
+    ]
+
     single_gestures = [
         "animations/Stand/BodyTalk/Listening/Listening_1",
         "animations/Stand/BodyTalk/Listening/Listening_2",

src/robot_interface/endpoints/main_receiver.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import zmq
 
 from robot_interface.endpoints.receiver_base import ReceiverBase

src/robot_interface/endpoints/receiver_base.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from abc import ABCMeta, abstractmethod
 
 from robot_interface.endpoints.socket_base import SocketBase

src/robot_interface/endpoints/socket_base.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from abc import ABCMeta
 
 import zmq

src/robot_interface/endpoints/video_sender.py

@@ -1,21 +1,11 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import zmq
 import threading
 import logging
 
-import struct
-
 from robot_interface.endpoints.socket_base import SocketBase
 from robot_interface.state import state
 from robot_interface.core.config import settings
 
-
 class VideoSender(SocketBase):
     """
     Video sender endpoint, responsible for sending video frames.
@@ -28,7 +18,7 @@ class VideoSender(SocketBase):
     """
     def __init__(self, zmq_context, port=settings.agent_settings.video_sender_port):
         super(VideoSender, self).__init__("video")
-        self.create_socket(zmq_context, zmq.PUB, port, [(zmq.SNDHWM,3)])
+        self.create_socket(zmq_context, zmq.PUB, port, [(zmq.CONFLATE,1)])
     
     def start_video_rcv(self):
         """
@@ -61,23 +51,10 @@ class VideoSender(SocketBase):
         :param vid_stream_name: The name of a camera subscription on the video service object vid_service
         :type vid_stream_name: str
         """
-        try:
         while not state.exit_event.is_set():
             try:
                 img = vid_service.getImageRemote(vid_stream_name)
-                    if img is not None:
+                #Possibly limit images sent if queuing issues arise
-                        raw_data = img[6]
+                self.socket.send(img[settings.video_config.image_buffer])
-                        width = img[0]
-                        height = img[1]
-
-                        width_bytes = struct.pack('<I', width)
-                        height_bytes = struct.pack('<I', height)
-
-                        self.socket.send_multipart([width_bytes, height_bytes, raw_data])
-                except KeyboardInterrupt:
-                    logging.info("Video receiving loop interrupted by user.")
             except:
                 logging.warn("Failed to retrieve video image from robot.")
-        finally:
-            vid_service.unsubscribe(vid_stream_name)
-            logging.info("Unsubscribed from video stream.")

src/robot_interface/main.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import logging
 
 from robot_interface.endpoints.audio_sender import AudioSender

src/robot_interface/state.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import logging
 import signal
 import threading
@@ -37,7 +30,6 @@ class State(object):
         self.exit_event = None
         self.sockets = []  
         self.qi_session = None 
-        self.is_speaking = False
 
     def initialize(self):
         """

src/robot_interface/utils/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

src/robot_interface/utils/get_config.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import os
 from dotenv import load_dotenv
 

src/robot_interface/utils/microphone.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from __future__ import unicode_literals  # So that `print` can print Unicode characters in names
 import logging
 import sys

src/robot_interface/utils/qi_utils.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import logging
 import sys
 

src/robot_interface/utils/timeblock.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import time
 
 

test/common/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

test/common/microphone_utils.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from __future__ import unicode_literals  # So that we can format strings with Unicode characters
 import random
 import sys

test/conftest.py

@@ -1,17 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
-from mock import patch, MagicMock
-
-import pytest
-
-
-@pytest.fixture(autouse=True)
-def mock_zmq_context():
-    with patch("zmq.Context") as mock:
-        mock.instance.return_value = MagicMock()
-        yield mock

test/integration/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

test/integration/test_config.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from mock import patch, mock
 
 from robot_interface.core.config import Settings

test/integration/test_microphone_utils.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import pyaudio
 
 import pytest

test/unit/__init__.py

@@ -1,6 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""

test/unit/test_actuation_receiver.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import sys
 
 import mock
@@ -27,109 +20,46 @@ def zmq_context():
     yield context
 
 
-def test_force_speech_clears_queue(mocker):
+def test_handle_unimplemented_endpoint(zmq_context):
     """
-    Tests that a force speech message clears the existing queue
+    Tests that the ``ActuationReceiver.handle_message`` method can
-    and places the high-priority message at the front.
+    handle an unknown or unimplemented endpoint without raising an error.
     """
-    mocker.patch("threading.Thread")
+    receiver = ActuationReceiver(zmq_context)
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
+    # Should not error
-    
-    mock_qi = mock.Mock()
-    sys.modules["qi"] = mock_qi 
-
-    mock_tts_service = mock.Mock()
-    mock_state.qi_session.service.return_value = mock_tts_service
-
-    # Use Mock Context
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    receiver._message_queue.put("old_message_1")
-    receiver._message_queue.put("old_message_2")
-    
-    assert receiver._message_queue.qsize() == 2
-
-    force_msg = {
-        "endpoint": "actuate/speech", 
-        "data": "Emergency Notification",
-        "is_priority": True,
-    }
-    receiver.handle_message(force_msg)
-
-    assert receiver._message_queue.qsize() == 1
-    queued_item = receiver._message_queue.get()
-    assert queued_item == "Emergency Notification"
-
-
-def test_handle_unimplemented_endpoint(mocker):
-    """
-    Tests handling of unknown endpoints.
-    """
-    mocker.patch("threading.Thread")
-    
-    # Use Mock Context
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
     receiver.handle_message({
         "endpoint": "some_endpoint_that_definitely_does_not_exist",
         "data": None,
     })
 
 
-def test_speech_message_no_data(mocker):
+def test_speech_message_no_data(zmq_context, mocker):
     """
-    Tests that if the message data is empty, the receiver returns immediately
+    Tests that the message handler logs a warning when a speech actuation
-    WITHOUT attempting to access the global robot state or session.
+    request (`actuate/speech`) is received but contains empty string data.
     """
-    # 1. Prevent background threads from running
+    mock_warn = mocker.patch("logging.warn")
-    mocker.patch("threading.Thread")
 
-    # 2. Mock the global state object
+    receiver = ActuationReceiver(zmq_context)
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-
-    # 3. Create a PropertyMock to track whenever 'qi_session' is accessed
-    # We attach it to the class type of the mock so it acts like a real property
-    mock_session_prop = mock.PropertyMock(return_value=None)
-    type(mock_state).qi_session = mock_session_prop
-
-    # 4. Initialize Receiver (Mocking the context to avoid ZMQ errors)
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    # 5. Send empty data
     receiver.handle_message({"endpoint": "actuate/speech", "data": ""})
 
-    # 6. Assertion: 
+    mock_warn.assert_called_with(mock.ANY)
-    # Because the code does `if not text: return` BEFORE `if not state.qi_session`,
-    # the state property should NEVER be read.
-    mock_session_prop.assert_not_called()
 
 
-def test_speech_message_invalid_data(mocker):
+def test_speech_message_invalid_data(zmq_context, mocker):
     """
-    Tests that if the message data is not a string, the function returns.     
+    Tests that the message handler logs a warning when a speech actuation
-    :param mocker: Description
+    request (`actuate/speech`) is received with data that is not a string (e.g., a boolean).
     """
-    mocker.patch("threading.Thread")
+    mock_warn = mocker.patch("logging.warn")
-
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-
-    mock_session_prop = mock.PropertyMock(return_value=None)
-    type(mock_state).qi_session = mock_session_prop
-
-    # Use Mock Context
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
 
+    receiver = ActuationReceiver(zmq_context)
     receiver.handle_message({"endpoint": "actuate/speech", "data": True})
 
-    # Because the code does `if not text: return` BEFORE `if not state.qi_session`,
+    mock_warn.assert_called_with(mock.ANY)
-    # the state property should NEVER be read.
-    mock_session_prop.assert_not_called()
 
-def test_speech_no_qi(mocker):
+
+def test_speech_no_qi(zmq_context, mocker):
     """
     Tests the actuation receiver's behavior when processing a speech request
     but the global state does not have an active QI session.
@@ -139,21 +69,16 @@ def test_speech_no_qi(mocker):
     mock_qi_session = mock.PropertyMock(return_value=None)
     type(mock_state).qi_session = mock_qi_session
 
-    mock_tts_service = mock.Mock()
+    receiver = ActuationReceiver(zmq_context)
-
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    receiver._tts_service = mock_tts_service
-
     receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
 
-    receiver._tts_service.assert_not_called()
+    mock_qi_session.assert_called()
 
 
-def test_speech(mocker):
+def test_speech(zmq_context, mocker):
     """
     Tests the core speech actuation functionality by mocking the QI TextToSpeech
-    service and verifying that the received message is put into the queue.
+    service and verifying that it is called correctly.
     """
     mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
 
@@ -164,182 +89,17 @@ def test_speech(mocker):
     mock_state.qi_session = mock.Mock()
     mock_state.qi_session.service.return_value = mock_tts_service
 
-    mock_zmq_ctx = mock.Mock()
+    receiver = ActuationReceiver(zmq_context)
-    receiver = ActuationReceiver(mock_zmq_ctx)
-
     receiver._tts_service = None
     receiver._handle_speech({"endpoint": "actuate/speech", "data": "Some message to speak."})
 
-    assert receiver._message_queue.qsize() == 1
+    mock_state.qi_session.service.assert_called_once_with("ALTextToSpeech")
 
-    queued_item = receiver._message_queue.get()
+    getattr(mock_qi, "async").assert_called_once()
-    assert queued_item == "Some message to speak."
+    call_args = getattr(mock_qi, "async").call_args[0]
+    assert call_args[0] == mock_tts_service.say
+    assert call_args[1] == "Some message to speak."
 
-def test_speech_priority(mocker):
-    """
-    Tests that a priority speech message is handled correctly by clearing the queue
-    and placing the priority message at the front.
-    """
-    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
-
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-
-    receiver._message_queue.put("old_message_1")
-    receiver._message_queue.put("old_message_2")
-
-    assert receiver._message_queue.qsize() == 2
-
-    priority_msg = {
-        "endpoint": "actuate/speech", 
-        "data": "Urgent Message",
-        "is_priority": True,
-    }
-    receiver._handle_speech(priority_msg)
-
-    assert receiver._message_queue.qsize() == 1
-    queued_item = receiver._message_queue.get()
-    assert queued_item == "Urgent Message"
-
-def test_handle_messages_loop(mocker):
-    """
-    Tests the background consumer loop (_handle_messages) processing an item.
-    Runs SYNCHRONOUSLY to ensure coverage tools pick up the lines.
-    """
-    # Patch Thread so the real background thread NEVER starts automatically
-    mocker.patch("threading.Thread")
-
-    # Mock state
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-        
-    # Setup initial speaking state to False (covers "Started speaking" print)
-    mock_state.is_speaking = False
-    
-    # Mock the TextToSpeech service
-    mock_tts_service = mock.Mock()
-    mock_state.qi_session.service.return_value = mock_tts_service
-    
-    # Initialize receiver (Thread is patched, so no thread starts)
-    # Use Mock Context to avoid ZMQ errors
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    # Manually inject service (since lazy loading might handle it, but this is safer)
-    receiver._tts_service = mock_tts_service
-
-    # This ensures the while loop iterates exactly once
-    mock_state.exit_event.is_set.side_effect = [False, True]
-
-    # Put an item in the queue
-    receiver._message_queue.put("Hello World")
-
-    # RUN MANUALLY in the main thread
-    # This executes the code: while -> try -> get -> if print -> speaking=True -> say
-    receiver._handle_messages()
-
-    # Assertions
-    assert receiver._message_queue.empty()
-    mock_tts_service.say.assert_called_with("Hello World")
-    assert mock_state.is_speaking is True
-
-
-def test_handle_messages_queue_empty(mocker):
-    """
-    Tests the Queue.Empty exception handler in the consumer loop.
-    This covers the logic that resets 'state.is_speaking' to False.
-    """
-    # Prevent the real background thread from starting
-    mocker.patch("threading.Thread")
-
-    # Mock the state object
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-        
-    # Setup 'is_speaking' property mock
-    # We set return_value=True so the code enters the 'if state.is_speaking:' block.
-    # We use PropertyMock to track when this attribute is set.
-    type(mock_state).is_speaking = True
-    
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    # This ensures the while loop body runs exactly once for our test
-    mock_state.exit_event.is_set.side_effect = [False, True]
-
-    # Force get() to raise Queue.Empty immediately (simulate timeout)
-    # We patch the 'get' method on the specific queue instance of our receiver
-    #mocker.patch.object(receiver._message_queue, 'get', side_effect=Queue.Empty)
-    
-    # Run the loop logic manually (synchronously)
-    receiver._handle_messages()
-
-    # Final Assertion: Verify is_speaking was set to False
-    # The code execution order is: read (returns True) -> print -> set (to False)
-    # assert_called_with checks the arguments of the LAST call, which is the setter.
-    assert mock_state.is_speaking is False
-
-
-def test_handle_messages_runtime_error(mocker):
-    """
-    Tests the RuntimeError exception handler (e.g. lost WiFi connection).
-    Uses a Mock ZMQ context to avoid 'Address already in use' errors.
-    """
-    # Patch Thread so we don't accidentally spawn real threads
-    mocker.patch("threading.Thread")
-    
-    # Mock the state and logging
-    mock_state = mocker.patch("robot_interface.endpoints.actuation_receiver.state")
-        
-    # Use a MOCK ZMQ context. 
-    # This prevents the receiver from trying to bind to a real TCP port.
-    mock_zmq_ctx = mock.Mock()
-    
-    # Initialize receiver with the mock context
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    mock_state.exit_event.is_set.side_effect = [False, True]
-
-    receiver._message_queue.put("Test Message")
-    
-    # Setup: ...BUT the service raises RuntimeError when asked to speak
-    mock_tts = mock.Mock()
-    mock_tts.say.side_effect = RuntimeError("Connection lost")
-    receiver._tts_service = mock_tts
-    
-    # Run the loop logic manually
-    receiver._handle_messages()
-    
-    # Assertions
-    assert mock_state.exit_event.is_set.called
-
-def test_clear_queue(mocker):
-    """
-    Tests that the clear_queue method properly drains all items from the message queue.
-    """
-    mocker.patch("threading.Thread")
-    
-    # Use Mock Context
-    mock_zmq_ctx = mock.Mock()
-    receiver = ActuationReceiver(mock_zmq_ctx)
-    
-    # Populate the queue with multiple items
-    receiver._message_queue.put("msg1")
-    receiver._message_queue.put("msg2")
-    receiver._message_queue.put("msg3")
-    
-    assert receiver._message_queue.qsize() == 3
-
-    # Clear the queue
-    receiver.clear_queue()
-
-    # Assert the queue is empty
-    assert receiver._message_queue.qsize() == 0
 
 def test_gesture_no_data(zmq_context, mocker):
     receiver = ActuationReceiver(zmq_context)

test/unit/test_audio_sender.py

@@ -1,10 +1,4 @@
-# -*- coding: utf-8 -*-
+# coding=utf-8
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import os
 
 import mock
@@ -84,7 +78,6 @@ def test_sending_audio(mocker):
     mock_zmq_context = mock.Mock()
     send_socket = mock.Mock()
 
-    mock_state.is_speaking = False
     # If there's something wrong with the microphone, it will raise an IOError when `read`ing.
     stream = mock.Mock()
     stream.read = _fake_read
@@ -100,36 +93,6 @@ def test_sending_audio(mocker):
     send_socket.assert_called()
 
 
-def test_no_sending_if_speaking(mocker):
-    """
-    Tests the successful sending of audio data over a ZeroMQ socket.
-    """
-    mock_choose_mic = mocker.patch("robot_interface.endpoints.audio_sender.choose_mic")
-    mock_choose_mic.return_value = {"name": u"Some mic", "index": 0L}
-
-    mock_state = mocker.patch("robot_interface.endpoints.audio_sender.state")
-    mock_state.exit_event.is_set.side_effect = [False, True]
-
-    mock_zmq_context = mock.Mock()
-    send_socket = mock.Mock()
-    
-    mock_state.is_speaking = True
-
-    # 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_not_called()
-
-
 def _fake_read_error(num_frames):
     """
     Helper function to simulate an I/O error during microphone stream reading.

test/unit/test_get_config.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 from robot_interface.utils.get_config import get_config
 
 

test/unit/test_main.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import pytest
 import threading
 import zmq

test/unit/test_main_receiver.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import mock
 import pytest
 import zmq

test/unit/test_microphone_utils.py

@@ -1,10 +1,4 @@
-# -*- coding: utf-8 -*-
+# coding=utf-8
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import mock
 import pytest
 

test/unit/test_qi_utils.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import sys
 
 # Import module under test

test/unit/test_reciever_base.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import pytest
 from robot_interface.endpoints.receiver_base import ReceiverBase
 

test/unit/test_socket_base.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import mock
 import zmq
 from robot_interface.endpoints.socket_base import SocketBase

test/unit/test_state.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import threading
 import signal
 import pytest

test/unit/test_time_block.py

@@ -1,10 +1,3 @@
-# -*- coding: utf-8 -*-
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
-
 import time
 
 import mock

test/unit/test_video_sender.py

@@ -1,11 +1,5 @@
-# -*- coding: utf-8 -*-
+# coding=utf-8
-"""
-This program has been developed by students from the bachelor Computer Science at Utrecht
-University within the Software Project course.
-© Copyright Utrecht University (Department of Information and Computing Sciences)
-"""
 
-import struct
 import mock
 import pytest
 import zmq
@@ -54,10 +48,8 @@ def test_video_streaming(zmq_context, mocker):
     # Pepper's image buffer lives at index 6
     mocker.patch.object(settings.video_config, "image_buffer", 6)
 
-    test_width = 320
-    test_height = 240
     mock_video_service = mock.Mock()
-    mock_video_service.getImageRemote.return_value = [test_width, test_height, None, None, None, None, b"fake_img"]
+    mock_video_service.getImageRemote.return_value = [None]*6 + ["fake_img"]
 
     fake_session = mock.Mock()
     fake_session.service.return_value = mock_video_service
@@ -71,16 +63,12 @@ def test_video_streaming(zmq_context, mocker):
 
     sender = VideoSender(zmq_context)
     send_socket = mock.Mock()
-    sender.socket.send_multipart = send_socket
+    sender.socket.send = send_socket
 
     sender.start_video_rcv()
     sender.video_rcv_loop(mock_video_service, "stream_name")
 
-    send_socket.assert_called_with([
+    send_socket.assert_called_with("fake_img")
-    struct.pack('<I', 320),
-    struct.pack('<I', 240),
-    b"fake_img"
-])
 
 
 def test_video_receive_error(zmq_context, mocker):
@@ -103,30 +91,9 @@ def test_video_receive_error(zmq_context, mocker):
 
     sender = VideoSender(zmq_context)
     send_socket = mock.Mock()
-    sender.socket.send_multipart = send_socket
+    sender.socket.send = send_socket
 
     sender.start_video_rcv()
     sender.video_rcv_loop(mock_video_service, "stream_name")
 
     send_socket.assert_not_called()
-
-def test_video_loop_keyboard_interrupt(zmq_context, mocker):
-    """Video loop should handle KeyboardInterrupt gracefully and unsubscribe."""
-    _patch_basics(mocker)
-    _patch_exit_event(mocker)
-    
-    # We mock the video service to raise KeyboardInterrupt when accessed
-    mock_video_service = mock.Mock()
-    mock_video_service.getImageRemote.side_effect = KeyboardInterrupt
-    
-    # Mock logging to verify the specific interrupt message is logged
-    mock_logger = mocker.patch("robot_interface.endpoints.video_sender.logging")
-
-    sender = VideoSender(zmq_context)
-    
-    # Execute the loop
-    sender.video_rcv_loop(mock_video_service, "stream_name")
-
-    # Verify the 'finally' block executed (unsubscribe)
-    mock_video_service.unsubscribe.assert_called_with("stream_name")
-    mock_logger.info.assert_any_call("Unsubscribed from video stream.")