pipeflowMonitor.py

import json
import subprocess
import threading
import time

from pipeflowData import *

class PipeflowMonitor:
	def __init__(self, data):
		args = ["pw-dump", "--no-colors", "--monitor"]
		self.p = subprocess.Popen(args, stdout=subprocess.PIPE)

		self.moarjson = ""
		self.running = True
		self.data = data

	def handleStdout(self):
		stdout = self.p.stdout.read1().decode('utf-8')
		if not stdout.endswith("\n]\n"):
			self.moarjson = self.moarjson + stdout
		else:
			lists = (self.moarjson + stdout).split("\n]\n")
			self.moarjson = ""
			for entry in lists:
				if entry.strip() == "":
					continue
				entry = entry + "]"
				data = json.loads(entry)
				for item in data:
					if not "info" in item:
						# Metadata
						if item["type"] == "PipeWire:Interface:Metadata":
								if "props" in item and "metadata.name" in item["props"]:
										if item["props"]["metadata.name"] == "default" and "metadata" in item:
												#print(item["metadata"])
												default_sink = ""
												default_source = ""
												for entry in item["metadata"]:
													if entry["key"] == "default.audio.sink":
														default_sink = entry["value"]["name"]
													elif entry["key"] == "default.audio.source":
														default_source = entry["value"]["name"]
												#print(default_sink)
												#print(default_source)
												self.data.set_meta(default_sink, default_source)
						continue

					id = item["id"]

					# REMOVE
					# if item.info is null, its a removal https://gitlab.freedesktop.org/pipewire/pipewire/-/commit/47e1f38f03adea1e2d00c56b8915ba054e8b73b0
					if item["info"] is None:
						if self.data.remove_by_id(id):
							print("REMOVED", id)
						else:
							#print("TODO remove?", id)
							pass
						continue

					info = item["info"]
					props = info["props"]

					# Node
					if item["type"] == "PipeWire:Interface:Node":
						name = props["node.name"]
						nick = name
						if "node.nick" in props:
							nick = props["node.nick"]
						node = self.data.get_node(id)
						if node:
							if "state" in info["change-mask"]:
								node.node_state = info["state"]
								#print("UPDATED state %d %s", id, info["state"])
							elif "params" in info ["change-mask"] and len(info["params"]["Props"]) > 0:
								if "channelVolumes" in info["params"]["Props"][0]:
									for i, vol in enumerate(info["params"]["Props"][0]["channelVolumes"]):
										if len(node.channels) > i:
											node.channels[i].chn_volume = vol
									#print("UPDATED params channelVolumes %d", id)
								if "channelMap" in info["params"]["Props"][0]:
									_chnMap = ""
									for i, label in enumerate(info["params"]["Props"][0]["channelMap"]):
										if len(node.channels) > i:
											node.channels[i].chn_label = str(label)
								if "mute" in info["params"]["Props"][0]:
									node.mute = info["params"]["Props"][0]["mute"]
									#print("UPDATED params mute %d: %s", id, node.mute)
								self.data.refresh_node(node)
							else:
								pass
								#print("TODO update node %d %s", id, info["change-mask"])
						else:
							_api = "?"
							if "device.api" in props:
								_api = props["device.api"]
							if "client.api" in props:
								_api = props["client.api"]
							_type = "?"
							if "media.class" in props:
								_type = props["media.class"]
							if "media.type" in props:
								_type = props["media.type"]
							_chnVols = []
							_chnMap = []
							_mute = False
							if "Props" in info["params"] and len(info["params"]["Props"]) > 0:
								if "channelVolumes" in info["params"]["Props"][0]:
									_chnVols = info["params"]["Props"][0]["channelVolumes"]
								if "channelMap" in info["params"]["Props"][0]:
									_chnMap = info["params"]["Props"][0]["channelMap"]
								if "mute" in info["params"]["Props"][0]:
									_mute = info["params"]["Props"][0]["mute"]
							self.data.add_node(Node(nick, name, id, _api, _type, info["state"], _mute, _chnVols, _chnMap))
							#print("ADDED node %d '%s'", id, name)

					# Port
					elif item["type"] == "PipeWire:Interface:Port":
						node_id = props["node.id"]
						port_name = props["port.name"]
						port_id = props["object.id"]
						is_inport = info["direction"] == "input"
						self.data.add_port(Port(port_name, port_id, node_id, is_inport))
						#print("ADDED port %d.%d", node_id, port_id)

					# Link
					elif item["type"] == "PipeWire:Interface:Link":
						self.data.add_link(Link(
							id,
							info["output-node-id"],
							info["output-port-id"],
							info["input-node-id"],
							info["input-port-id"]
						))
						#print("ADDED link %d", id)

					# IGNORE: Factory, Module TODO: do we need these for anyhting?
					elif item["type"] == "PipeWire:Interface:Factory" or item["type"] == "PipeWire:Interface:Module":
						pass

					# TODO: should we show clients?
					elif item["type"] == "PipeWire:Interface:Client":
						#print("TODO add? %d %s '%s'", id, item["type"].split(":")[-1], props["application.name"])
						pass
					else:
						#print("TODO add? %d %s", id, item["type"].split(":")[-1])
						pass

	def read_loop(self):
		while self.running:
			self.handleStdout()
			if self.moarjson == "":
				time.sleep(0.1)

	def start_loop(self):
		self.handleStdout()
		while self.moarjson != "":
			self.handleStdout()
		self.thread = threading.Thread(target=self.read_loop)
		self.thread.start()

	def stop_loop(self):
		self.running = False
		self.thread.join()