pipeflowData.py

import json
import math
import os
import time

class Link:
	#Pipewire ids
	link_id: int
	output_node_pw: int
	output_port_pw: int
	input_node_pw: int
	input_port_pw: int

	#Imgui object ids
	im: int
	output_attribute_im: int
	input_attribute_im: int
	deleted: bool

	def __init__(self, link, output_node, output_port, input_node, input_port):
		self.link_id = link
		self.output_node_pw = output_node
		self.output_port_pw = output_port
		self.input_node_pw = input_node
		self.input_port_pw = input_port
		self.im = 0
		self.output_attribute_im = 0
		self.input_attribute_im = 0
		self.deleted = False

	def other_node(self, node):
		return self.output_node_pw if node.node_id != self.output_node_pw else self.input_node_pw

class Port:
	#Pipewire data
	port_label: str
	port_id: int
	node_id: int
	is_inport: bool
	links: [Link]

	im: int
	deleted: bool

	def __init__(self, port_label, port_id, node_id, is_inport):
		self.port_label = port_label
		self.port_id = port_id
		self.node_id = node_id
		self.is_inport = is_inport
		self.links = []
		self.im = 0
		self.deleted = False

def clamp(v, min, max):
		return max(min(v, max), min)

# Volume in human readable percent to volume as float
def p_to_f(v):
	return max(math.pow(v / 100, 3), 0)

# Volume as float to percent
def f_to_p(v):
	return math.cbrt(v) * 100

class Channel:
	#Pipewire data
	chn_label: str
	chn_volume: float
	ports: [Port]

	im: int
	tooltip_im: int

	percent_volume: int
	placeholder: bool

	def __init__(self, chn_label, chn_vol):
		self.chn_label = chn_label
		self.chn_volume = chn_vol
		self.percent_volume = f_to_p(chn_vol)
		self.ports = []
		self.placeholder = chn_label == "N/A"
		self.im = 0

class Node:
	#Pipewire data
	node_label: str
	node_name: str
	node_id: int
	node_api: str
	node_type: str
	node_state: str
	mute: bool
	channels: [Channel]
	channel_lock: bool
	can_default: int

	#Imgui object data
	im: int
	mute_button: int
	default_button: int
	lock_button: int

	#GUI custom data
	pos: [int]
	size: [int]
	deleted: bool
	moved: bool
	current_group: int

	def __init__(self, node_label, node_name, node_id, node_api, node_type, node_state, mute, _chnVols, _chnMap):
		self.node_label = node_label
		self.node_name = node_name
		self.node_id = node_id
		self.node_api = node_api
		self.node_type = node_type
		self.node_state = node_state
		self.mute = mute

		self.channels = [Channel(label, vol) for vol,label in zip(_chnVols,_chnMap)]
		self.channel_lock = False
		if len(self.channels) > 1 and all(x.chn_volume == self.channels[0].chn_volume for x in self.channels):
			self.channel_lock = True

		self.can_default = 0
		if node_type == "Audio/Sink":
			self.can_default = 1
		elif node_type == "Audio/Source":
			self.can_default = 2

		self.im = 0
		self.default_button = 0
		self.mute_button = 0
		self.lock_button = 0
		self.hide_button = 0
		self.pos = [0,0]
		self.deleted = False
		self.moved = False
		self.current_group = -1

	def is_default(self, meta):
		return meta.default_sink == self.node_name or meta.default_source == self.node_name

	def is_hidden(self, config):
		return self.node_label in config.hide_names and not config.show_hidden

	def linked_ids(self):
		links = [ link for channel in self.channels for port in channel.ports for link in port.links ]
		return list(set([ link.other_node(self) for link in links ]))

class Meta:
	default_sink: str
	default_source: str

	sink_button: int
	source_button: int

	def __init__(self, default_sink, default_source):
		self.default_sink = default_sink
		self.default_source = default_source

class Config:
	def __init__(self, **entries):
		self.max_volume = 100
		self.hide_names = ["Dummy-Driver","Freewheel-Driver","PulseAudio Volume Control"]
		self.show_hidden = False
		self.hide_button = False

		self.margins = 30
		self.padding = self.margins
		self.group_padding = 15

		self.sort_names = False
		self.sort_moved = False
		self.group_vertical = True
		self.group_mode = 1

		self.__dict__.update(entries)
		self.locked_values()

	def locked_values(self):
		self.group_mode_names = ["None", "Type", "API", "Name", "Connections"]
		self.group_order = ["Unlinked","Stream/Output/Audio","Audio/Source","Stream/Input/Audio","Audio/Sink","Midi/Bridge","?"]

class PipeflowData:
	def __init__(self, log):
		self.metadata = Meta("", "")
		self.config = Config()
		self.nodes = {}
		self.ports = {}
		self.links = {}

		self.log = log
		self.config_file = "pipeflow_config.json"

		self.meta_refresh = None
		self.node_refresh = None
		self.port_refresh = None
		self.link_refresh = None

		#Lock to wait for certain GUI loops to finish
		self.delete_lock = 0

	def set_meta(self, default_sink, default_source):
		while self.delete_lock > 0:
			time.sleep(0.1)
		self.metadata.default_sink = default_sink
		self.metadata.default_source = default_source
		if self.meta_refresh is not None:
			self.meta_refresh()

	def add_node(self, node):
		if node.node_id not in self.nodes:
			while self.delete_lock > 0:
				time.sleep(0.1)
			self.nodes[node.node_id] = node

			if self.node_refresh is not None:
				self.node_refresh(node)
		else:
			print("Added already existing node")

	def get_node(self, id):
		if id in self.nodes:
			return self.nodes[id]
		return None

	def refresh_node(self, node):
		if self.node_refresh is not None:
			self.node_refresh(node)

	def add_port(self, port):
		if port.port_id in self.ports:
			alt_port = self.ports[port.port_id]
			if alt_port.node_id != port.node_id or alt_port.port_label != port.port_label:
				self.log.error("ERROR: mismatched port ", alt_port.port_label, port.port_label)
			return

		node = self.get_node(port.node_id)
		if node:
			while self.delete_lock > 0:
				time.sleep(0.1)
			self.ports[port.port_id] = port
			channel = None
			# Find correct channel
			for c in node.channels:
				if c.chn_label in port.port_label or c.placeholder:
					channel = c
			if channel is None:
				channel = Channel("N/A", 0)
				node.channels.append(channel)

			# Add port to channel
			channel.ports.append(port)

			if self.port_refresh is not None:
				self.port_refresh(port)
		else:
			self.log.error("ERROR: Port added without node")

	def add_link(self, link):
		while self.delete_lock > 0:
			time.sleep(0.1)
		self.links[link.link_id] = link
		self.ports[link.input_port_pw].links.append(link)
		self.ports[link.output_port_pw].links.append(link)
		if self.link_refresh is not None:
			self.link_refresh(link)

	def remove_by_id(self, id):
		if id in self.nodes:
			self.nodes[id].deleted = True
			while self.delete_lock > 0:
				time.sleep(0.1)
			self.node_refresh(self.nodes[id])
			del self.nodes[id]
			return 'node'
		elif id in self.ports:
			self.ports[id].deleted = True
			while self.delete_lock > 0:
				time.sleep(0.1)
			del self.ports[id]
			return 'port'
		elif id in self.links:
			self.links[id].deleted = True
			while self.delete_lock > 0:
				time.sleep(0.1)
			self.link_refresh(self.links[id])
			del self.links[id]
			return 'link'
		return None

	def save_config(self):
		with open(self.config_file, "w") as f:
			f.write(json.dumps(self.config.__dict__, indent=4))

	def load_config(self):
		if os.path.exists(self.config_file):
			with open(self.config_file, "r") as f:
				self.config = Config(**json.load(f))
		self.save_config()