Add node volume

include/warppipe/warppipe.hpp

@@ -137,6 +137,11 @@ struct RouteRule {
   std::string target_node;
 };
 
+struct VolumeState {
+  float volume = 1.0f;
+  bool mute = false;
+};
+
 struct MetadataInfo {
   std::string default_sink_name;
   std::string default_source_name;
@@ -166,6 +171,9 @@ class Client {
                                             const VirtualNodeOptions& options = VirtualNodeOptions{});
   Status RemoveNode(NodeId node);
 
+  Status SetNodeVolume(NodeId node, float volume, bool mute);
+  Result<VolumeState> GetNodeVolume(NodeId node) const;
+
   Result<Link> CreateLink(PortId output, PortId input, const LinkOptions& options);
   Result<Link> CreateLinkByName(std::string_view output_node,
                                 std::string_view output_port,
@@ -193,6 +201,8 @@ class Client {
   Status Test_ForceDisconnect();
   Status Test_TriggerPolicyCheck();
   size_t Test_GetPendingAutoLinkCount() const;
+  Status Test_SetNodeVolume(NodeId node, float volume, bool mute);
+  Result<VolumeState> Test_GetNodeVolume(NodeId node) const;
 #endif
 
  private:

src/warppipe.cpp

@@ -14,6 +14,8 @@
 #include <pipewire/extensions/metadata.h>
 
 #include <spa/param/audio/format-utils.h>
+#include <spa/param/props.h>
+#include <spa/pod/builder.h>
 #include <spa/utils/defs.h>
 
 #include <spa/utils/result.h>
@@ -277,6 +279,8 @@ struct Client::Impl {
   std::unordered_map<uint32_t, std::unique_ptr<StreamData>> virtual_streams;
   std::unordered_map<uint32_t, std::unique_ptr<LinkProxy>> link_proxies;
 
+  std::unordered_map<uint32_t, VolumeState> volume_states;
+
   uint32_t next_rule_id = 1;
   std::unordered_map<uint32_t, RouteRule> route_rules;
   std::vector<PendingAutoLink> pending_auto_links;
@@ -1216,6 +1220,64 @@ Status Client::RemoveNode(NodeId node) {
   return Status::Ok();
 }
 
+Status Client::SetNodeVolume(NodeId node, float volume, bool mute) {
+  Status status = impl_->EnsureConnected();
+  if (!status.ok()) {
+    return status;
+  }
+  if (node.value == 0) {
+    return Status::Error(StatusCode::kInvalidArgument, "invalid node id");
+  }
+
+  volume = std::clamp(volume, 0.0f, 1.5f);
+
+  pw_thread_loop_lock(impl_->thread_loop);
+
+  {
+    std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+    if (impl_->nodes.find(node.value) == impl_->nodes.end()) {
+      pw_thread_loop_unlock(impl_->thread_loop);
+      return Status::Error(StatusCode::kNotFound, "node not found");
+    }
+  }
+
+  auto* proxy = static_cast<pw_node*>(
+      pw_registry_bind(impl_->registry, node.value,
+                       PW_TYPE_INTERFACE_Node, PW_VERSION_NODE, 0));
+  if (!proxy) {
+    pw_thread_loop_unlock(impl_->thread_loop);
+    return Status::Error(StatusCode::kInternal, "failed to bind node proxy");
+  }
+
+  uint8_t buffer[128];
+  spa_pod_builder builder = SPA_POD_BUILDER_INIT(buffer, sizeof(buffer));
+  auto* param = reinterpret_cast<const spa_pod*>(spa_pod_builder_add_object(
+      &builder,
+      SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
+      SPA_PROP_volume, SPA_POD_Float(volume),
+      SPA_PROP_mute, SPA_POD_Bool(mute)));
+
+  pw_node_set_param(proxy, SPA_PARAM_Props, 0, param);
+  pw_proxy_destroy(reinterpret_cast<pw_proxy*>(proxy));
+
+  {
+    std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+    impl_->volume_states[node.value] = VolumeState{volume, mute};
+  }
+
+  pw_thread_loop_unlock(impl_->thread_loop);
+  return Status::Ok();
+}
+
+Result<VolumeState> Client::GetNodeVolume(NodeId node) const {
+  std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+  auto it = impl_->volume_states.find(node.value);
+  if (it == impl_->volume_states.end()) {
+    return {Status::Ok(), VolumeState{}};
+  }
+  return {Status::Ok(), it->second};
+}
+
 Result<Link> Client::CreateLink(PortId output, PortId input, const LinkOptions& options) {
   Status status = impl_->EnsureConnected();
   if (!status.ok()) {
@@ -1719,6 +1781,30 @@ size_t Client::Test_GetPendingAutoLinkCount() const {
   std::lock_guard<std::mutex> lock(impl_->cache_mutex);
   return impl_->pending_auto_links.size();
 }
+
+Status Client::Test_SetNodeVolume(NodeId node, float volume, bool mute) {
+  if (!impl_) {
+    return Status::Error(StatusCode::kUnavailable, "no impl");
+  }
+  std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+  if (impl_->nodes.find(node.value) == impl_->nodes.end()) {
+    return Status::Error(StatusCode::kNotFound, "node not found");
+  }
+  impl_->volume_states[node.value] = VolumeState{std::clamp(volume, 0.0f, 1.5f), mute};
+  return Status::Ok();
+}
+
+Result<VolumeState> Client::Test_GetNodeVolume(NodeId node) const {
+  if (!impl_) {
+    return {Status::Error(StatusCode::kUnavailable, "no impl"), {}};
+  }
+  std::lock_guard<std::mutex> lock(impl_->cache_mutex);
+  auto it = impl_->volume_states.find(node.value);
+  if (it == impl_->volume_states.end()) {
+    return {Status::Ok(), VolumeState{}};
+  }
+  return {Status::Ok(), it->second};
+}
 #endif
 
 }  // namespace warppipe

tests/warppipe_tests.cpp

@@ -4,6 +4,7 @@
 #include <fstream>
 #include <string>
 
+#include <catch2/catch_approx.hpp>
 #include <catch2/catch_test_macros.hpp>
 
 namespace {
@@ -792,3 +793,63 @@ TEST_CASE("policy mode does not override user defaults") {
   REQUIRE(defaults2.value.configured_sink_name == defaults.value.configured_sink_name);
   REQUIRE(defaults2.value.configured_source_name == defaults.value.configured_source_name);
 }
+
+TEST_CASE("Test_SetNodeVolume sets and retrieves volume state") {
+  auto result = warppipe::Client::Create(DefaultOptions());
+  REQUIRE(result.ok());
+  auto &client = result.value;
+
+  warppipe::NodeInfo node;
+  node.id = warppipe::NodeId{900};
+  node.name = "vol-sink";
+  node.media_class = "Audio/Sink";
+  REQUIRE(client->Test_InsertNode(node).ok());
+
+  auto vol = client->Test_GetNodeVolume(warppipe::NodeId{900});
+  REQUIRE(vol.ok());
+  REQUIRE(vol.value.volume == Catch::Approx(1.0f));
+  REQUIRE(vol.value.mute == false);
+
+  REQUIRE(client->Test_SetNodeVolume(warppipe::NodeId{900}, 0.5f, false).ok());
+  vol = client->Test_GetNodeVolume(warppipe::NodeId{900});
+  REQUIRE(vol.ok());
+  REQUIRE(vol.value.volume == Catch::Approx(0.5f));
+  REQUIRE(vol.value.mute == false);
+
+  REQUIRE(client->Test_SetNodeVolume(warppipe::NodeId{900}, 0.75f, true).ok());
+  vol = client->Test_GetNodeVolume(warppipe::NodeId{900});
+  REQUIRE(vol.ok());
+  REQUIRE(vol.value.volume == Catch::Approx(0.75f));
+  REQUIRE(vol.value.mute == true);
+}
+
+TEST_CASE("Test_SetNodeVolume clamps volume") {
+  auto result = warppipe::Client::Create(DefaultOptions());
+  REQUIRE(result.ok());
+  auto &client = result.value;
+
+  warppipe::NodeInfo node;
+  node.id = warppipe::NodeId{901};
+  node.name = "vol-clamp";
+  node.media_class = "Audio/Sink";
+  REQUIRE(client->Test_InsertNode(node).ok());
+
+  REQUIRE(client->Test_SetNodeVolume(warppipe::NodeId{901}, 2.0f, false).ok());
+  auto vol = client->Test_GetNodeVolume(warppipe::NodeId{901});
+  REQUIRE(vol.ok());
+  REQUIRE(vol.value.volume == Catch::Approx(1.5f));
+
+  REQUIRE(client->Test_SetNodeVolume(warppipe::NodeId{901}, -1.0f, false).ok());
+  vol = client->Test_GetNodeVolume(warppipe::NodeId{901});
+  REQUIRE(vol.ok());
+  REQUIRE(vol.value.volume == Catch::Approx(0.0f));
+}
+
+TEST_CASE("Test_SetNodeVolume fails for nonexistent node") {
+  auto result = warppipe::Client::Create(DefaultOptions());
+  REQUIRE(result.ok());
+
+  auto status = result.value->Test_SetNodeVolume(warppipe::NodeId{999}, 0.5f, false);
+  REQUIRE_FALSE(status.ok());
+  REQUIRE(status.code == warppipe::StatusCode::kNotFound);
+}