refactor: added config file and moved constants

- Moved hardcoded configuration constants to a dedicated config.py file.
- Created VideoConfig, AudioConfig, MainConfig, and Settings classes in config.py

ref: N25B-236
This commit is contained in:
Pim Hutting
2025-11-09 15:43:22 +01:00
parent c037eb7ec2
commit 4402b21a73
8 changed files with 74 additions and 21 deletions

View File

@@ -6,9 +6,11 @@ import zmq
from robot_interface.endpoints.receiver_base import ReceiverBase
from robot_interface.state import state
from robot_interface.core.config import settings
class ActuationReceiver(ReceiverBase):
def __init__(self, zmq_context, port=5557):
def __init__(self, zmq_context, port= settings.agent_settings.actuating_receiver_port):
"""
The actuation receiver endpoint, responsible for handling speech and gesture requests.

View File

@@ -8,13 +8,13 @@ import zmq
from robot_interface.endpoints.socket_base import SocketBase
from robot_interface.state import state
from robot_interface.utils.microphone import choose_mic
from robot_interface.core.config import settings
logger = logging.getLogger(__name__)
class AudioSender(SocketBase):
def __init__(self, zmq_context, port=5558):
def __init__(self, zmq_context, port= settings.agent_settings.audio_sender_port):
super(AudioSender, self).__init__(str("audio")) # Convert future's unicode_literal to str
self.create_socket(zmq_context, zmq.PUB, port)
self.thread = None
@@ -49,13 +49,14 @@ class AudioSender(SocketBase):
self.thread = None
def _stream(self):
chunk = 512 # 320 at 16000 Hz is 20ms, 512 is required for Silero-VAD
audio_settings = settings.audio_config
chunk = audio_settings.chunk_size # 320 at 16000 Hz is 20ms, 512 is required for Silero-VAD
# Docs say this only raises an error if neither `input` nor `output` is True
stream = self.audio.open(
format=pyaudio.paFloat32,
channels=1,
rate=16000,
channels=audio_settings.channels,
rate=audio_settings.sample_rate,
input=True,
input_device_index=self.microphone["index"],
frames_per_buffer=chunk,

View File

@@ -3,9 +3,10 @@ import zmq
from robot_interface.endpoints.receiver_base import ReceiverBase
from robot_interface.state import state
from robot_interface.core.config import settings
class MainReceiver(ReceiverBase):
def __init__(self, zmq_context, port=5555):
def __init__(self, zmq_context, port= settings.agent_settings.main_receiver_port):
"""
The main receiver endpoint, responsible for handling ping and negotiation requests.

View File

@@ -4,10 +4,10 @@ import logging
from robot_interface.endpoints.socket_base import SocketBase
from robot_interface.state import state
from robot_interface.core.config import settings
class VideoSender(SocketBase):
def __init__(self, zmq_context, port=5556):
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)])
@@ -20,12 +20,13 @@ class VideoSender(SocketBase):
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)
video_settings = settings.video_config
camera_index = video_settings.camera_index
kQVGA = video_settings.resolution
kRGB = video_settings.color_space
FPS = video_settings.fps
video_name = video_settings.stream_name
vid_stream_name = video.subscribeCamera(video_name, camera_index, kQVGA, kRGB, FPS)
thread = threading.Thread(target=self.video_rcv_loop, args=(video, vid_stream_name))
thread.start()
@@ -43,6 +44,6 @@ class VideoSender(SocketBase):
try:
img = vid_service.getImageRemote(vid_stream_name)
#Possibly limit images sent if queuing issues arise
self.socket.send(img[6])
self.socket.send(img[settings.video_config.image_buffer])
except:
logging.warn("Failed to retrieve video image from robot.")