diff --git a/tests/__init__.py b/tests/__init__.py index d132a6b..eac8a58 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,5 +1,6 @@ import vban_cmd from vban_cmd import kinds +from vban_cmd.channel import Modes import socket from threading import Thread @@ -13,5 +14,7 @@ def setup_package(): tests.worker = Thread(target=tests._send_register_rt, daemon=True) tests.worker.start() + tests._modes = Modes() + def teardown_package(): pass diff --git a/tests/tests_higher.py b/tests/tests_higher.py new file mode 100644 index 0000000..6b5468c --- /dev/null +++ b/tests/tests_higher.py @@ -0,0 +1,42 @@ +from nose.tools import assert_equal, nottest +from parameterized import parameterized, parameterized_class + +import unittest +from tests import tests + +#@nottest +@parameterized_class([ + { "val": False }, { "val": True } +]) +class TestSetAndGetBoolHigher(unittest.TestCase): + """ strip tests, physical and virtual """ + @parameterized.expand([ + (0, 'mute'), (2, 'mono'), (3, 'A1'), (3, 'B3') + ]) + def test_it_sets_and_gets_strip_bool_params(self, index, param): + setattr(tests.strip[index], param, self.val) + retval = getattr(tests.strip[index], param) + self.assertTrue(isinstance(retval, bool)) + assert_equal(retval, self.val) + + +#@nottest +@parameterized_class([ + { "val": "test0" }, { "val": "test1" } +]) +class TestSetAndGetStringHigher(unittest.TestCase): + """ strip tests, physical and virtual """ + @parameterized.expand([ + (2, 'label'), (3, 'label') + ]) + def test_it_sets_and_gets_strip_string_params(self, index, param): + setattr(tests.strip[index], param, self.val) + assert_equal(getattr(tests.strip[index], param), self.val) + + """ bus tests, physical and virtual """ + @parameterized.expand([ + (0, 'label'), (4, 'label') + ]) + def test_it_sets_and_gets_bus_string_params(self, index, param): + setattr(tests.bus[index], param, self.val) + assert_equal(getattr(tests.bus[index], param), self.val) diff --git a/tests/tests_lower.py b/tests/tests_lower.py new file mode 100644 index 0000000..c71bce1 --- /dev/null +++ b/tests/tests_lower.py @@ -0,0 +1,29 @@ +from nose.tools import assert_equal, nottest +from parameterized import parameterized, parameterized_class + +import unittest +from tests import tests + +#@nottest +@parameterized_class([ + { "val": 0 }, { "val": 1 }, +]) +class TestSetAndGetParamsLower(unittest.TestCase): + """ get_rt, set_rt test """ + @parameterized.expand([ + (0, 'mute'), (4, 'mute'), + ]) + def test_it_sets_and_gets_strip_bool_params(self, index, param): + tests.set_rt(f'Strip[{index}]', param, self.val) + retval = tests.get_rt() + retval = not int.from_bytes(retval.stripstate[index], 'little') & tests._modes._mute == 0 + assert_equal(retval, self.val) + + @parameterized.expand([ + (0, 'mono'), (5, 'mono'), + ]) + def test_it_sets_and_gets_strip_bool_params(self, index, param): + tests.set_rt(f'Strip[{index}]', param, self.val) + retval = tests.get_rt() + retval = not int.from_bytes(retval.stripstate[index], 'little') & tests._modes._mono == 0 + assert_equal(retval, self.val) diff --git a/vban_cmd/bus.py b/vban_cmd/bus.py new file mode 100644 index 0000000..916fa7f --- /dev/null +++ b/vban_cmd/bus.py @@ -0,0 +1,93 @@ +from .errors import VMCMDErrors +from . import channel +from .channel import Channel +from . import kinds +from .meta import bus_output_prop + +class OutputBus(Channel): + """ Base class for output buses. """ + @classmethod + def make(cls, is_physical, remote, index, *args, **kwargs): + """ + Factory function for output busses. + Returns a physical/virtual bus of a kind. + """ + OutputBus = PhysicalOutputBus if is_physical else VirtualOutputBus + OB_cls = type(f'Bus{remote.kind.name}', (OutputBus,), { + }) + return OB_cls(remote, index, *args, **kwargs) + + @property + def identifier(self): + return f'Bus[{self.index}]' + + @property + def mute(self) -> bool: + data = self.getter() + return not int.from_bytes(data.busstate[self.index], 'little') & self._modes._mute == 0 + + @mute.setter + def mute(self, val: bool): + if not isinstance(val, bool) and val not in (0,1): + raise VMCMDErrors('mute is a boolean parameter') + self.setter('mute', 1 if val else 0) + + @property + def mono(self) -> bool: + data = self.getter() + return not int.from_bytes(data.busstate[self.index], 'little') & self._modes._mono == 0 + + @mono.setter + def mono(self, val: bool): + if not isinstance(val, bool) and val not in (0,1): + raise VMCMDErrors('mono is a boolean parameter') + self.setter('mono', 1 if val else 0) + + @property + def eq(self) -> bool: + data = self.getter() + return not int.from_bytes(data.busstate[self.index], 'little') & self._modes._eq == 0 + + @eq.setter + def eq(self, val: bool): + if not isinstance(val, bool) and val not in (0,1): + raise ('eq is a boolean parameter') + self.setter('eq.On', 1 if val else 0) + + @property + def eq_ab(self) -> bool: + data = self.getter() + return not int.from_bytes(data.busstate[self.index], 'little') & self._modes._eqb == 0 + + @eq_ab.setter + def eq_ab(self, val: bool): + if not isinstance(val, bool) and val not in (0,1): + raise VMCMDErrors('eq_ab is a boolean parameter') + self.setter('eq.ab', 1 if val else 0) + + @property + def label(self) -> str: + data = self.getter() + return data.buslabels[self.index] + + @label.setter + def label(self, val: str): + if not isinstance(val, str): + raise VMCMDErrors('label is a string parameter') + self.setter('Label', val) + + +class PhysicalOutputBus(OutputBus): + @property + def device(self) -> str: + data = self.getter() + return + + @property + def sr(self) -> int: + data = self.getter() + return + + +class VirtualOutputBus(OutputBus): + pass diff --git a/vban_cmd/meta.py b/vban_cmd/meta.py index 93c184e..42ff9fe 100644 --- a/vban_cmd/meta.py +++ b/vban_cmd/meta.py @@ -1,7 +1,7 @@ from .errors import VMCMDErrors def strip_output_prop(param): - """ A channel prop. """ + """ A strip output prop. """ def fget(self): data = self.getter() return not int.from_bytes(data.stripstate[self.index], 'little') & getattr(self._modes, f'_bus{param.lower()}') == 0 @@ -9,4 +9,15 @@ def strip_output_prop(param): if not isinstance(val, bool) and val not in (0, 1): raise VMCMDErrors(f'{param} is a boolean parameter') self.setter(param, 1 if val else 0) - return property(fget, fset) \ No newline at end of file + return property(fget, fset) + +def bus_output_prop(param): + """ A bus output prop. """ + def fget(self): + data = self.getter() + return not int.from_bytes(data.busstate[self.index], 'little') & getattr(self._modes, f'_bus{param.lower()}') == 0 + def fset(self, val): + if not isinstance(val, bool) and val not in (0, 1): + raise VMCMDErrors(f'{param} is a boolean parameter') + self.setter(param, 1 if val else 0) + return property(fget, fset) diff --git a/vban_cmd/vban_cmd.py b/vban_cmd/vban_cmd.py index 2ce6e94..72c7e85 100644 --- a/vban_cmd/vban_cmd.py +++ b/vban_cmd/vban_cmd.py @@ -4,19 +4,19 @@ import socket from time import sleep import sys from threading import Thread -from typing import NamedTuple +from typing import NamedTuple, NoReturn from .errors import VMCMDErrors from . import kinds from .dataclass import ( HEADER_SIZE, - MAX_PACKET_SIZE, VBAN_VMRT_Packet_Data, VBAN_VMRT_Packet_Header, RegisterRTHeader, TextRequestHeader ) from .strip import InputStrip +from .bus import OutputBus class VbanCmd(abc.ABC): def __init__(self, *args, **kwargs): @@ -26,6 +26,7 @@ class VbanCmd(abc.ABC): self._bps = kwargs['bps'] self._channel = kwargs['channel'] self._delay = kwargs['delay'] + self._max_polls = kwargs['max_polls'] self._bps_opts = \ [0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400,19200, 31250, 38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800,921600, @@ -99,14 +100,13 @@ class VbanCmd(abc.ABC): while not data: data = self._fetch_rt_packet() return data - for i in range(2): + for i in range(self._max_polls): data = fget() return data def set_rt(self, id_, param, val): cmd = f'{id_}.{param}={val}' if self._sendrequest_string_socket in self.ready_to_write: - print(f'sending {cmd} to {socket.gethostbyname(self._ip)}:{self._port}') self._sendrequest_string_socket.sendto( self._text_header.header + cmd.encode(), (socket.gethostbyname(self._ip), self._port) ) @@ -114,6 +114,29 @@ class VbanCmd(abc.ABC): self._text_header.framecounter = count.to_bytes(4, 'little') sleep(self._delay) + @property + def type(self): + data = self.get_rt() + return data.voicemeetertype + + @property + def version(self): + data = self.get_rt() + return data.voicemeeterversion + + def show(self) -> NoReturn: + """ Shows Voicemeeter if it's hidden. """ + self.set_rt('Command', 'Show', 1) + def hide(self) -> NoReturn: + """ Hides Voicemeeter if it's shown. """ + self.set_rt('Command', 'Show', 0) + def shutdown(self) -> NoReturn: + """ Closes Voicemeeter. """ + self.set_rt('Command', 'Shutdown', 1) + def restart(self) -> NoReturn: + """ Restarts Voicemeeter's audio engine. """ + self.set_rt('Command', 'Restart', 1) + def __exit__(self, exc_type, exc_value, exc_traceback): self._rt_packet_socket.close() sys.exit() @@ -129,7 +152,7 @@ def _make_remote(kind: NamedTuple) -> VbanCmd: def init(self, *args, **kwargs): defaultkwargs = { 'ip': None, 'port': 6990, 'streamname': 'Command1', 'bps': 0, - 'channel': 0, 'delay': 0.001, + 'channel': 0, 'delay': 0.001, 'max_polls': 2 } kwargs = defaultkwargs | kwargs VbanCmd.__init__(self, *args, **kwargs) @@ -139,6 +162,9 @@ def _make_remote(kind: NamedTuple) -> VbanCmd: self.strip = \ tuple(InputStrip.make((i < self.phys_in), self, i) for i in range(self.phys_in + self.virt_in)) + self.bus = \ + tuple(OutputBus.make((i < self.phys_out), self, i) + for i in range(self.phys_out + self.virt_out)) return type(f'VbanCmd{kind.name}', (VbanCmd,), { '__init__': init,