pipeflowControl.py

import json
import subprocess

from pipeflowData import *

class PipeflowControl:
	def __init__(self, data):
		self.data = data
		self.clog = data.command_log

	def run(self, cmd, nullout=False):
		if nullout:
			subprocess.run(cmd, stdout=subprocess.DEVNULL)
		else:
			subprocess.Popen(cmd)
		self.clog.append(" ".join(cmd))
		self.data.log_refresh()

		if self.data.config.command_log_file:
			with open(self.data.config.command_log_file, "a") as f:
				f.write(" ".join(cmd) + '\n')
		#print(" ".join(cmd))

	# Command helpers
	def set_metadata(self, args):
		self.run(["pw-metadata"] + args)

	def set_param(self, args):
		self.run(["pw-cli", "set-param"] + args, True)

	def pw_link(self, args):
		self.run(["pw-link"] + args)

	def custom_command(self, cmd):
		print(" ".join(cmd))
		self.run(cmd)

	# Control functions
	def update_meta(self, meta, is_sink):
		if is_sink:
			data = json.dumps({
				"name": meta.default_sink
			})
			self.set_metadata(["0", "default.audio.sink", data])
			self.set_metadata(["0", "default.configured.audio.sink", data])
		else:
			data = json.dumps({
				"name": meta.default_source
			})
			self.set_metadata(["0", "default.audio.source", data])
			self.set_metadata(["0", "default.configured.audio.sink", data])

	def update_mute(self, node):
		data = json.dumps({
			"mute": node.mute
		})
		self.set_param([str(node.node_id), "Props", data])

	def update_volume(self, node):
		vols = [ c.chn_volume for c in node.channels if not c.placeholder ]
		#print("Set volumes ", vols)
		data = json.dumps({
			"channelVolumes": vols
		})
		self.set_param([str(node.node_id), "Props", data])

	def add_link(self, inport, outport):
		self.pw_link([str(outport.port_id), str(inport.port_id)])
		print("ADDED link", inport.port_label, "->", outport.port_label)

	def remove_link(self, link):
		self.pw_link(["-d", str(link.link_id)])
		print("REMOVED link", link.link_id)