feat: add service for setting announce volume

signage.repos

@@ -11,7 +11,7 @@
     type: git
     url: https://github.com/autowarefoundation/autoware_msgs.git
     version: main
   src/autoware_adapi_msgs:
     type: git
     url: https://github.com/autowarefoundation/autoware_adapi_msgs.git
     version: main
@@ -19,4 +19,3 @@
     type: git
     url: https://github.com/tier4/boot_shutdown_tools.git
     version: main
-

src/signage/package.xml

@@ -4,13 +4,14 @@
   <version>0.1.0</version>
   <description>The signage package</description>
 
   <maintainer email="[email protected]">yabuta</maintainer>
 
   <license>Apache License 2.0</license>
   <exec_depend>ament_index_python</exec_depend>
   <exec_depend>signage_version</exec_depend>
 
   <depend>autoware_auto_system_msgs</depend>
+  <depend>python-pulsectl-pip</depend>
   <depend>rclpy</depend>
   <depend>tier4_api_msgs</depend>
   <depend>tier4_debug_msgs</depend>

src/signage/src/signage/announce_controller.py

@@ -2,9 +2,17 @@
 # -*- coding: utf-8 -*-
 # This Python file uses the following encoding: utf-8
 
+import os
+
 from PyQt5.QtMultimedia import QSound
 from rclpy.duration import Duration
 from ament_index_python.packages import get_package_share_directory
+from pulsectl import Pulse
+
+from std_msgs.msg import Float32
+from tier4_hmi_msgs.srv import SetVolume
+from tier4_external_api_msgs.msg import ResponseStatus
+
 
 # The higher the value, the higher the priority
 PRIORITY_DICT = {
@@ -20,6 +28,8 @@
     "going_to_arrive": 1,
 }
 
+CURRENT_VOLUME_PATH = "/opt/autoware/volume.txt"
+
 
 class AnnounceControllerProperty:
     def __init__(self, node, autoware_interface, parameter_interface):
@@ -34,6 +44,18 @@
         self._package_path = get_package_share_directory("signage") + "/resource/sound/"
         self._check_playing_timer = self._node.create_timer(1, self.check_playing_callback)
 
+        self._pulse = Pulse()
+        if os.path.isfile(CURRENT_VOLUME_PATH):
+            with open(CURRENT_VOLUME_PATH, "r") as f:
+                self._sink = self._pulse.get_sink_by_name(
+                    self._pulse.server_info().default_sink_name
+                )
+                self._pulse.volume_set_all_chans(self._sink, float(f.readline()))
+
+        self._get_volume_pub = self._node.create_publisher(Float32, "~/get/volume", 1)
+        self._node.create_timer(1.0, self.publish_volume_callback)
+        self._node.create_service(SetVolume, "~/set/volume", self.set_volume)
+
     def process_pending_announce(self):
         try:
             for play_sound in self._pending_announce_list:
@@ -100,3 +122,18 @@
             # To stop repeat announcement
             self.send_announce(message)
             self._prev_depart_and_arrive_type = message
+
+    def publish_volume_callback(self):
+        self._sink = self._pulse.get_sink_by_name(self._pulse.server_info().default_sink_name)
+        self._get_volume_pub.publish(Float32(data=self._sink.volume.value_flat))
+
+    def set_volume(self, request, response):
+        try:
+            self._sink = self._pulse.get_sink_by_name(self._pulse.server_info().default_sink_name)
+            self._pulse.volume_set_all_chans(self._sink, request.volume)
+            with open(CURRENT_VOLUME_PATH, "w") as f:
+                f.write(f"{self._sink.volume.value_flat}\n")
+            response.status.code = ResponseStatus.SUCCESS
+        except Exception:
+            response.status.code = ResponseStatus.ERROR
+        return response