Merge branch 'fix/send-video' into 'main'

fix: send video

See merge request ics/sp/2025/n25b/pepperplus-ri!28

src/robot_interface/__init__.py

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

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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
@@ -17,10 +24,6 @@ class AgentSettings(object):
     :vartype video_sender_port: int
     :ivar audio_sender_port: Port used for sending audio data, defaults to 5558.
     :vartype audio_sender_port: int
-    :ivar face_detection_port: Port used for sending face detection events, defaults to 5559.
-    :vartype face_detection_port: int
-    :ivar face_detection_interval: Time between face detection events, defaults to 1000 ms.
-    :vartype face_detection_interval: int
     """
     def __init__(
         self,
@@ -29,16 +32,12 @@ class AgentSettings(object):
         main_receiver_port=None,
         video_sender_port=None,
         audio_sender_port=None,
-        face_detection_port=None,
-        face_detection_interval=None,
     ):
         self.control_backend_host = get_config(control_backend_host, "AGENT__CONTROL_BACKEND_HOST", "localhost")
         self.actuation_receiver_port = get_config(actuation_receiver_port, "AGENT__ACTUATION_RECEIVER_PORT", 5557, int)
         self.main_receiver_port = get_config(main_receiver_port, "AGENT__MAIN_RECEIVER_PORT", 5555, int)
         self.video_sender_port = get_config(video_sender_port, "AGENT__VIDEO_SENDER_PORT", 5556, int)
         self.audio_sender_port = get_config(audio_sender_port, "AGENT__AUDIO_SENDER_PORT", 5558, int)
-        self.face_detection_port = get_config(face_detection_port, "AGENT__FACE_DETECTION_PORT", 5559, int)
-        self.face_detection_interval = get_config(face_detection_interval, "AGENT__FACE_DETECTION_INTERVAL", 1000, int)
 
 
 class VideoConfig(object):
@@ -69,7 +68,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", 11, int)
+        self.color_space = get_config(color_space, "VIDEO__COLOR_SPACE", 13, 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

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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

src/robot_interface/endpoints/audio_sender.py

@@ -1,3 +1,10 @@
+# -*- 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

src/robot_interface/endpoints/face_detector.py

@@ -1,93 +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 __future__ import unicode_literals
-
-import json
-import logging
-import threading
-import time
-import zmq
-
-from robot_interface.endpoints.socket_base import SocketBase
-from robot_interface.state import state
-from robot_interface.core.config import settings
-
-
-class FaceDetectionSender(SocketBase):
-    """
-    Face detection endpoint.
-
-    Subscribes to and polls ALMemory["FaceDetected"], sends events to CB.
-    """
-
-    def __init__(self, zmq_context, port=settings.agent_settings.face_detection_port):
-        super(FaceDetectionSender, self).__init__("face")
-
-        self.create_socket(zmq_context, zmq.PUB, port)
-
-        self._face_service = None
-        self._memory_service = None
-
-        self._face_thread = None
-
-    def start_face_detection(self):
-        if not state.qi_session:
-            logging.warning("No Qi session available. Face detection not started.")
-            return
-
-        self._face_service = state.qi_session.service("ALFaceDetection")
-        self._memory_service = state.qi_session.service("ALMemory")
-
-        self._face_service.setTrackingEnabled(False)
-        self._face_service.setRecognitionEnabled(False)
-
-        self._face_service.subscribe(
-            "FaceDetectionSender",
-            settings.agent_settings.face_detection_interval,
-            0.0,
-        )
-
-        self._face_thread = threading.Thread(target=self._face_loop)
-        self._face_thread.start()
-
-        logging.info("Face detection started.")
-
-    def _face_loop(self):
-        """
-        Continuously send face detected to the CB, at the interval set in the
-        ``start_face_detection`` method.
-        """
-        while not state.exit_event.is_set():
-            try:
-                value = self._memory_service.getData("FaceDetected", 0)
-
-                face_present = (
-                    value
-                    and len(value) > 1
-                    and value[1]
-                    and value[1][0]
-                    and len(value[1][0]) > 0
-                )
-
-                self.socket.send(json.dumps({"face_detected": face_present}).encode("utf-8"))
-            except Exception:
-                logging.exception("Error reading FaceDetected")
-
-            time.sleep(settings.agent_settings.face_detection_interval / 1000.0)
-
-    def stop_face_detection(self):
-        try:
-            if self._face_service:
-                self._face_service.unsubscribe("FaceDetectionSender")
-                self._face_service.setTrackingEnabled(False)
-            logging.info("Face detection stopped.")
-        except Exception:
-            logging.warning("Error during face detection cleanup.")
-
-    def close(self):
-        super(FaceDetectionSender, self).close()
-        self.stop_face_detection()

src/robot_interface/endpoints/gesture_settings.py

@@ -1,3 +1,10 @@
+# -*- 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", 

src/robot_interface/endpoints/main_receiver.py

@@ -1,10 +1,16 @@
+# -*- 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
 from robot_interface.state import state
 
 from robot_interface.core.config import settings
-from robot_interface.endpoints.face_detector import FaceDetectionSender
 
 
 class MainReceiver(ReceiverBase):
@@ -38,7 +44,6 @@ class MainReceiver(ReceiverBase):
         """
         return {"endpoint": "ping", "data": message.get("data")}
 
-
     @staticmethod
     def _handle_port_negotiation(message):
         """

src/robot_interface/endpoints/receiver_base.py

@@ -1,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,7 +1,16 @@
+# -*- 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
@@ -19,7 +28,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.CONFLATE,1)])
+        self.create_socket(zmq_context, zmq.PUB, port, [(zmq.SNDHWM,3)])
     
     def start_video_rcv(self):
         """
@@ -52,10 +61,23 @@ 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)
-                #Possibly limit images sent if queuing issues arise
+                    if img is not None:
-                self.socket.send(img[settings.video_config.image_buffer])
+                        raw_data = img[6]
+                        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,3 +1,10 @@
+# -*- 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
@@ -12,8 +19,6 @@ from robot_interface.endpoints.video_sender import VideoSender
 from robot_interface.state import state
 from robot_interface.core.config import settings
 from robot_interface.utils.timeblock import TimeBlock
-from robot_interface.endpoints.face_detector import FaceDetectionSender
-
 
 
 def main_loop(context):
@@ -37,12 +42,6 @@ def main_loop(context):
     video_sender.start_video_rcv()
     audio_sender.start()
 
-    # --- Face detection sender ---
-    face_sender = FaceDetectionSender(context)
-    state.sockets.append(face_sender)
-    face_sender.start_face_detection()
-
-
     # 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]
 

src/robot_interface/state.py

@@ -1,3 +1,10 @@
+# -*- 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

src/robot_interface/utils/__init__.py

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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

test/integration/__init__.py

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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

@@ -0,0 +1,6 @@
+# -*- 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,3 +1,10 @@
+# -*- 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

test/unit/test_audio_sender.py

@@ -1,4 +1,10 @@
-# 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

test/unit/test_face_detection_sender.py

@@ -1,175 +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 __future__ import unicode_literals
-
-import json
-import mock
-import pytest
-
-from robot_interface.endpoints.face_detector import FaceDetectionSender
-from robot_interface.state import state
-
-
-@pytest.fixture(autouse=True)
-def initialized_state(monkeypatch):
-    """
-    Fully initialize global state so __getattribute__ allows access.
-    """
-    # Bypass the initialization guard
-    monkeypatch.setattr(state, "is_initialized", True, raising=False)
-
-    # Install a controllable exit_event
-    exit_event = mock.Mock()
-    exit_event.is_set = mock.Mock(return_value=True)
-    monkeypatch.setattr(state, "exit_event", exit_event, raising=False)
-
-    # Default qi_session is None unless overridden
-    monkeypatch.setattr(state, "qi_session", None, raising=False)
-
-    yield
-
-
-def test_start_face_detection_no_qi_session():
-    """
-    Returns early when qi_session is None.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-    sender.start_face_detection()
-
-    assert sender._face_thread is None
-    assert sender._face_service is None
-    assert sender._memory_service is None
-
-
-def test_start_face_detection_happy_path(mocker):
-    """
-    Initializes services and starts background thread.
-    """
-    mock_face = mock.Mock()
-    mock_memory = mock.Mock()
-
-    mock_qi = mock.Mock()
-    mock_qi.service.side_effect = lambda name: {
-        "ALFaceDetection": mock_face,
-        "ALMemory": mock_memory,
-    }[name]
-
-    state.qi_session = mock_qi
-
-    fake_thread = mock.Mock()
-    mocker.patch("threading.Thread", return_value=fake_thread)
-
-    sender = FaceDetectionSender(mock.Mock())
-    sender.start_face_detection()
-
-    mock_face.setTrackingEnabled.assert_called_with(False)
-    mock_face.setRecognitionEnabled.assert_called_with(False)
-    mock_face.subscribe.assert_called_once()
-    fake_thread.start.assert_called_once()
-
-
-def test_face_loop_face_detected_true(mocker):
-    """
-    Sends face_detected=True when face data exists.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    sender._memory_service = mock.Mock()
-    sender._memory_service.getData.return_value = [0, [[1]]]
-    sender.socket = mock.Mock()
-
-    mocker.patch("time.sleep")
-    state.exit_event.is_set.side_effect = [False, True]
-
-    sender._face_loop()
-
-    sent = sender.socket.send.call_args[0][0]
-    payload = json.loads(sent.decode("utf-8"))
-
-    assert payload["face_detected"] is True
-
-
-def test_face_loop_face_detected_false(mocker):
-    """
-    Sends face_detected=False when no face data exists.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    sender._memory_service = mock.Mock()
-    sender._memory_service.getData.return_value = []
-    sender.socket = mock.Mock()
-
-    mocker.patch("time.sleep")
-    state.exit_event.is_set.side_effect = [False, True]
-
-    sender._face_loop()
-
-    sent = sender.socket.send.call_args[0][0]
-    payload = json.loads(sent.decode("utf-8"))
-
-    assert not payload["face_detected"]
-
-
-def test_face_loop_handles_exception(mocker):
-    """
-    Exceptions inside loop are swallowed.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    sender._memory_service = mock.Mock()
-    sender._memory_service.getData.side_effect = Exception("boom")
-    sender.socket = mock.Mock()
-
-    mocker.patch("time.sleep")
-    state.exit_event.is_set.side_effect = [False, True]
-
-    # Must not raise
-    sender._face_loop()
-
-
-def test_stop_face_detection_happy_path():
-    """
-    Unsubscribes and disables tracking.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    mock_face = mock.Mock()
-    sender._face_service = mock_face
-
-    sender.stop_face_detection()
-
-    mock_face.unsubscribe.assert_called_once()
-    mock_face.setTrackingEnabled.assert_called_with(False)
-
-
-def test_stop_face_detection_exception():
-    """
-    stop_face_detection swallows service exceptions.
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    mock_face = mock.Mock()
-    mock_face.unsubscribe.side_effect = Exception("fail")
-    sender._face_service = mock_face
-
-    sender.stop_face_detection()
-
-
-def test_close_calls_stop_face_detection(mocker):
-    """
-    close() calls parent close and stop_face_detection().
-    """
-    sender = FaceDetectionSender(mock.Mock())
-
-    mocker.patch.object(sender, "stop_face_detection")
-    mocker.patch(
-        "robot_interface.endpoints.face_detector.SocketBase.close"
-    )
-
-    sender.close()
-
-    sender.stop_face_detection.assert_called_once()

test/unit/test_get_config.py

@@ -1,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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
@@ -55,9 +62,6 @@ class DummySender:
     def start(self):
         self.called = True
 
-    def start_face_detection(self):
-        self.called = True
-
     def close(self):
         pass
 
@@ -111,13 +115,11 @@ def patched_main_components(monkeypatch, fake_sockets, fake_poll):
     fake_act = FakeReceiver(act_sock)
     video_sender = DummySender()
     audio_sender = DummySender()
-    face_sender = DummySender()
 
     monkeypatch.setattr(main_mod, "MainReceiver", lambda ctx: fake_main)
     monkeypatch.setattr(main_mod, "ActuationReceiver", lambda ctx: fake_act)
     monkeypatch.setattr(main_mod, "VideoSender", lambda ctx: video_sender)
     monkeypatch.setattr(main_mod, "AudioSender", lambda ctx: audio_sender)
-    monkeypatch.setattr(main_mod, "FaceDetectionSender", lambda ctx: face_sender)
 
     # Register sockets for the fake poller
     fake_poll.registered = {main_sock: zmq.POLLIN, act_sock: zmq.POLLIN}

test/unit/test_main_receiver.py

@@ -1,3 +1,10 @@
+# -*- 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,4 +1,10 @@
-# 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,3 +1,10 @@
+# -*- 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,5 +1,11 @@
-# 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
@@ -48,8 +54,10 @@ 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 = [None]*6 + ["fake_img"]
+    mock_video_service.getImageRemote.return_value = [test_width, test_height, None, None, None, None, b"fake_img"]
 
     fake_session = mock.Mock()
     fake_session.service.return_value = mock_video_service
@@ -63,12 +71,16 @@ def test_video_streaming(zmq_context, mocker):
 
     sender = VideoSender(zmq_context)
     send_socket = mock.Mock()
-    sender.socket.send = send_socket
+    sender.socket.send_multipart = send_socket
 
     sender.start_video_rcv()
     sender.video_rcv_loop(mock_video_service, "stream_name")
 
-    send_socket.assert_called_with("fake_img")
+    send_socket.assert_called_with([
+    struct.pack('<I', 320),
+    struct.pack('<I', 240),
+    b"fake_img"
+])
 
 
 def test_video_receive_error(zmq_context, mocker):
@@ -91,9 +103,30 @@ def test_video_receive_error(zmq_context, mocker):
 
     sender = VideoSender(zmq_context)
     send_socket = mock.Mock()
-    sender.socket.send = send_socket
+    sender.socket.send_multipart = 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.")