mirror of
https://github.com/onyx-and-iris/vban-cmd-python.git
synced 2024-11-15 17:10:46 +00:00
6971feb398
misc module added with Event class. base class observable thread now checks for currently registered events.
195 lines
5.5 KiB
Python
195 lines
5.5 KiB
Python
from abc import abstractmethod
|
|
from enum import IntEnum
|
|
from functools import cached_property
|
|
from typing import Iterable, NoReturn, Self
|
|
|
|
from .base import VbanCmd
|
|
from .bus import request_bus_obj as bus
|
|
from .command import Command
|
|
from .config import request_config as configs
|
|
from .kinds import KindMapClass
|
|
from .kinds import request_kind_map as kindmap
|
|
from .strip import request_strip_obj as strip
|
|
|
|
|
|
class FactoryBuilder:
|
|
"""
|
|
Builder class for factories.
|
|
|
|
Separates construction from representation.
|
|
"""
|
|
|
|
BuilderProgress = IntEnum("BuilderProgress", "strip bus command", start=0)
|
|
|
|
def __init__(self, factory, kind: KindMapClass):
|
|
self._factory = factory
|
|
self.kind = kind
|
|
self._info = (
|
|
f"Finished building strips for {self._factory}",
|
|
f"Finished building buses for {self._factory}",
|
|
f"Finished building commands for {self._factory}",
|
|
)
|
|
|
|
def _pinfo(self, name: str) -> NoReturn:
|
|
"""prints progress status for each step"""
|
|
name = name.split("_")[1]
|
|
print(self._info[int(getattr(self.BuilderProgress, name))])
|
|
|
|
def make_strip(self) -> Self:
|
|
self._factory.strip = tuple(
|
|
strip(i < self.kind.phys_in, self._factory, i)
|
|
for i in range(self.kind.num_strip)
|
|
)
|
|
return self
|
|
|
|
def make_bus(self) -> Self:
|
|
self._factory.bus = tuple(
|
|
bus(i < self.kind.phys_out, self._factory, i)
|
|
for i in range(self.kind.num_bus)
|
|
)
|
|
return self
|
|
|
|
def make_command(self) -> Self:
|
|
self._factory.command = Command.make(self._factory)
|
|
return self
|
|
|
|
|
|
class FactoryBase(VbanCmd):
|
|
"""Base class for factories, subclasses VbanCmd."""
|
|
|
|
def __init__(self, kind_id: str, **kwargs):
|
|
defaultsubs = {"pdirty": True, "ldirty": False}
|
|
if "subs" in kwargs:
|
|
defaultsubs = defaultsubs | kwargs.pop("subs")
|
|
defaultkwargs = {
|
|
"ip": None,
|
|
"port": 6980,
|
|
"streamname": "Command1",
|
|
"bps": 0,
|
|
"channel": 0,
|
|
"ratelimit": 0.01,
|
|
"sync": False,
|
|
"subs": defaultsubs,
|
|
}
|
|
kwargs = defaultkwargs | kwargs
|
|
self.kind = kindmap(kind_id)
|
|
super().__init__(**kwargs)
|
|
self.builder = FactoryBuilder(self, self.kind)
|
|
self._steps = (
|
|
self.builder.make_strip,
|
|
self.builder.make_bus,
|
|
self.builder.make_command,
|
|
)
|
|
self._configs = None
|
|
|
|
def __str__(self) -> str:
|
|
return f"Voicemeeter {self.kind}"
|
|
|
|
@property
|
|
@abstractmethod
|
|
def steps(self):
|
|
pass
|
|
|
|
@cached_property
|
|
def configs(self):
|
|
self._configs = configs(self.kind.name)
|
|
return self._configs
|
|
|
|
|
|
class BasicFactory(FactoryBase):
|
|
"""
|
|
Represents a Basic VbanCmd subclass
|
|
|
|
Responsible for directing the builder class
|
|
"""
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls is BasicFactory:
|
|
raise TypeError(f"'{cls.__name__}' does not support direct instantiation")
|
|
return object.__new__(cls)
|
|
|
|
def __init__(self, kind_id, **kwargs):
|
|
super().__init__(kind_id, **kwargs)
|
|
[step()._pinfo(step.__name__) for step in self.steps]
|
|
|
|
@property
|
|
def steps(self) -> Iterable:
|
|
"""steps required to build the interface for a kind"""
|
|
return self._steps
|
|
|
|
|
|
class BananaFactory(FactoryBase):
|
|
"""
|
|
Represents a Banana VbanCmd subclass
|
|
|
|
Responsible for directing the builder class
|
|
"""
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls is BananaFactory:
|
|
raise TypeError(f"'{cls.__name__}' does not support direct instantiation")
|
|
return object.__new__(cls)
|
|
|
|
def __init__(self, kind_id, **kwargs):
|
|
super().__init__(kind_id, **kwargs)
|
|
[step()._pinfo(step.__name__) for step in self.steps]
|
|
|
|
@property
|
|
def steps(self) -> Iterable:
|
|
"""steps required to build the interface for a kind"""
|
|
return self._steps
|
|
|
|
|
|
class PotatoFactory(FactoryBase):
|
|
"""
|
|
Represents a Potato VbanCmd subclass
|
|
|
|
Responsible for directing the builder class
|
|
"""
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls is PotatoFactory:
|
|
raise TypeError(f"'{cls.__name__}' does not support direct instantiation")
|
|
return object.__new__(cls)
|
|
|
|
def __init__(self, kind_id: str, **kwargs):
|
|
super().__init__(kind_id, **kwargs)
|
|
[step()._pinfo(step.__name__) for step in self.steps]
|
|
|
|
@property
|
|
def steps(self) -> Iterable:
|
|
"""steps required to build the interface for a kind"""
|
|
return self._steps
|
|
|
|
|
|
def vbancmd_factory(kind_id: str, **kwargs) -> VbanCmd:
|
|
"""
|
|
Factory method, invokes a factory creation class of a kind
|
|
|
|
Returns a VbanCmd class of a kind
|
|
"""
|
|
match kind_id:
|
|
case "basic":
|
|
_factory = BasicFactory
|
|
case "banana":
|
|
_factory = BananaFactory
|
|
case "potato":
|
|
_factory = PotatoFactory
|
|
case _:
|
|
raise ValueError(f"Unknown Voicemeeter kind '{kind_id}'")
|
|
return type(f"VbanCmd{kind_id.capitalize()}", (_factory,), {})(kind_id, **kwargs)
|
|
|
|
|
|
def request_vbancmd_obj(kind_id: str, **kwargs) -> VbanCmd:
|
|
"""
|
|
Interface entry point. Wraps factory method and handles errors
|
|
|
|
Returns a reference to a VbanCmd class of a kind
|
|
"""
|
|
VBANCMD_obj = None
|
|
try:
|
|
VBANCMD_obj = vbancmd_factory(kind_id, **kwargs)
|
|
except (ValueError, TypeError) as e:
|
|
raise SystemExit(e)
|
|
return VBANCMD_obj
|