diff --git a/__main__.py b/__main__.py index e484f76..db3fd8a 100644 --- a/__main__.py +++ b/__main__.py @@ -1,10 +1,20 @@ import vban_cmd -from time import sleep def main(): - with vban_cmd.connect('potato', ip='ws.local') as vban: - for i in range(100): - print(vban.bus[3].levels.all) + with vban_cmd.connect('potato', ip=ip, port=port, streamname=streamname, bps=bps, channel=channel) as vban: + for param in ['A1', 'A2', 'A3', 'A4', 'A5', 'B1', 'B2', 'B3', 'mono', 'mute']: + setattr(vban.strip[0], param, True) + print(getattr(vban.strip[0], param)) + setattr(vban.strip[0], param, False) + print(getattr(vban.strip[0], param)) + if __name__ == '__main__': + kind_id = 'potato' + ip = 'ws.local' + port=6980 + streamname = 'testing' + bps = 57600 + channel=3 + main() diff --git a/tests/__init__.py b/tests/__init__.py index eac8a58..1b2a0d6 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -3,18 +3,26 @@ from vban_cmd import kinds from vban_cmd.channel import Modes import socket from threading import Thread +from time import sleep -_kind = 'banana' +_kind = 'potato' +opts = { + 'ip': 'ws.local', + 'streamname': 'testing', + 'port': 6980, + 'bps': 0, +} -vbanrs = {kind.id: vban_cmd.connect(_kind, ip='ws.local') for kind in kinds.all} +vbanrs = {kind.id: vban_cmd.connect(_kind, **opts) for kind in kinds.all} tests = vbanrs[_kind] def setup_package(): tests._rt_packet_socket.bind((socket.gethostbyname(socket.gethostname()), tests._port)) tests.worker = Thread(target=tests._send_register_rt, daemon=True) tests.worker.start() - tests._modes = Modes() def teardown_package(): - pass + tests._rt_packet_socket.close() + tests._rt_register_socket.close() + tests._sendrequest_string_socket.close() diff --git a/tests/runmany.ps1 b/tests/runmany.ps1 index 47e2812..c7fe165 100644 --- a/tests/runmany.ps1 +++ b/tests/runmany.ps1 @@ -19,17 +19,6 @@ Function RunTests { if ($cycle_num -eq 20) { $firstrun = $true } - (Get-Content '__init__.py') | ForEach-Object { - $m = [regex]::Match($_, '^(DELAY)\s=\s(\d.\d+)') - if ($m.captures.groups.count -gt 1) { - $delay = $m.captures.groups[2] - } - $m = [regex]::Match($_, '^(MAX_POLLS)\s=\s(\d+)') - if ($m.captures.groups.count -gt 1) { - $maxpolls = $m.captures.groups[2] - } - } - 1..$num | ForEach-Object { ` if ($Log) { "Running test $_ of ${num} runs" | Tee-Object -FilePath $logfile -Append } else { Write-Host "Running test $_ of ${num} runs" } @@ -60,14 +49,12 @@ Function RunTests { "NOTES:" | Tee-Object -FilePath $summary_file -Append } "===========================================================`n" + ` - "${num} test run with ${delay} delay and ${maxpolls} max_polls`n" + ` "Total failures: ${failures}`n" + ` "Logfile for this test run: ${log_backupfile}`n" + ` "===========================================================" | ` Tee-Object -FilePath $summary_file -Append } else { "===========================================================", - "${num} test run with ${delay} delay and ${maxpolls} max_polls", "Total failures: ${failures}", "===========================================================" | Write-Host } diff --git a/tests/tests_higher.py b/tests/tests_higher.py index 6b5468c..3085638 100644 --- a/tests/tests_higher.py +++ b/tests/tests_higher.py @@ -11,7 +11,7 @@ from tests import tests class TestSetAndGetBoolHigher(unittest.TestCase): """ strip tests, physical and virtual """ @parameterized.expand([ - (0, 'mute'), (2, 'mono'), (3, 'A1'), (3, 'B3') + (0, 'mute'), (2, 'mono'), (3, 'A1'), (6, 'B3'), (6, 'mute'), ]) def test_it_sets_and_gets_strip_bool_params(self, index, param): setattr(tests.strip[index], param, self.val) @@ -19,6 +19,16 @@ class TestSetAndGetBoolHigher(unittest.TestCase): self.assertTrue(isinstance(retval, bool)) assert_equal(retval, self.val) + """ bus tests, physical and virtual """ + @parameterized.expand([ + (0, 'mute'), (2, 'mono'), (6, 'mute'), (2, 'eq'), (7, 'eq_ab') + ]) + def test_it_sets_and_gets_bus_bool_params(self, index, param): + setattr(tests.bus[index], param, self.val) + retval = getattr(tests.bus[index], param) + self.assertTrue(isinstance(retval, bool)) + assert_equal(retval, self.val) + #@nottest @parameterized_class([ @@ -27,7 +37,7 @@ class TestSetAndGetBoolHigher(unittest.TestCase): class TestSetAndGetStringHigher(unittest.TestCase): """ strip tests, physical and virtual """ @parameterized.expand([ - (2, 'label'), (3, 'label') + (2, 'label'), (6, 'label') ]) def test_it_sets_and_gets_strip_string_params(self, index, param): setattr(tests.strip[index], param, self.val) @@ -35,8 +45,29 @@ class TestSetAndGetStringHigher(unittest.TestCase): """ bus tests, physical and virtual """ @parameterized.expand([ - (0, 'label'), (4, 'label') + (0, 'label'), (7, '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) + + +#@nottest +class TestSetAndGetFloatHigher(unittest.TestCase): + """ strip tests, physical and virtual """ + @parameterized.expand([ + (0, 1, 'gain', -6.3), (7, 4, 'gain', -12.5), (3, 3, 'gain', 3.3) + ]) + def test_it_sets_and_gets_strip_float_params(self, index, j, param, val): + setattr(tests.strip[index].gainlayer[j], param, val) + retval = getattr(tests.strip[index].gainlayer[j], param) + assert_equal(retval, val) + + """ bus tests, physical and virtual """ + @parameterized.expand([ + (0, 'gain', -6.3), (7, 'gain', -12.5), (3, 'gain', 3.3) + ]) + def test_it_sets_and_gets_bus_float_params(self, index, param, val): + setattr(tests.bus[index], param, val) + retval = getattr(tests.bus[index], param) + assert_equal(retval, val) diff --git a/vban_cmd/bus.py b/vban_cmd/bus.py index a7c1805..31fa02d 100644 --- a/vban_cmd/bus.py +++ b/vban_cmd/bus.py @@ -72,7 +72,7 @@ class OutputBus(Channel): self.setter('Label', val) @property - def gain(self): + def gain(self) -> float: def fget(): val = self.public_packet.busgain[self.index] if val < 10000: diff --git a/vban_cmd/channel.py b/vban_cmd/channel.py index 3dc41e7..9e1937c 100644 --- a/vban_cmd/channel.py +++ b/vban_cmd/channel.py @@ -53,6 +53,7 @@ class Modes: _sel: hex=0x10000000 _monitor: hex=0x20000000 + class Channel(abc.ABC): """ Base class for InputStrip and OutputBus. """ def __init__(self, remote, index): diff --git a/vban_cmd/strip.py b/vban_cmd/strip.py index dcd9e96..860992d 100644 --- a/vban_cmd/strip.py +++ b/vban_cmd/strip.py @@ -158,7 +158,7 @@ class GainLayer(InputStrip): self._i = i @property - def gain(self): + def gain(self) -> float: def fget(): val = getattr(self.public_packet, f'stripgainlayer{self._i+1}')[self.index] if val < 10000: @@ -170,7 +170,7 @@ class GainLayer(InputStrip): return round((fget() * 0.01), 1) @gain.setter - def gain(self, val): + def gain(self, val: float): self.setter(f'GainLayer[{self._i}]', val) diff --git a/vban_cmd/vban_cmd.py b/vban_cmd/vban_cmd.py index d217f96..68ad9ce 100644 --- a/vban_cmd/vban_cmd.py +++ b/vban_cmd/vban_cmd.py @@ -2,7 +2,6 @@ import abc import select import socket from time import sleep -import sys from threading import Thread from typing import NamedTuple, NoReturn @@ -40,7 +39,6 @@ class VbanCmd(abc.ABC): ) self._register_rt_header = RegisterRTHeader() self.expected_packet = VBAN_VMRT_Packet_Header() - self.buff = None self._rt_register_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self._rt_packet_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) @@ -52,12 +50,6 @@ class VbanCmd(abc.ABC): self.ready_to_read, self.ready_to_write, in_error = select.select(is_readable, is_writable, is_error, 60) def __enter__(self): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - res = s.connect_ex((self._ip, self._port)) - s.close() - if res: - raise VMCMDErrors('Could not connect to remote Voicemeeter') - self._rt_packet_socket.bind((socket.gethostbyname(socket.gethostname()), self._port)) worker = Thread(target=self._send_register_rt, daemon=True) worker.start() @@ -107,8 +99,7 @@ class VbanCmd(abc.ABC): @property def public_packet(self): - self.buff = self._get_rt() - return self.buff + return self._get_rt() def _get_rt(self): def fget(): @@ -116,8 +107,7 @@ class VbanCmd(abc.ABC): while not data: data = self._fetch_rt_packet() return data - private_packet = fget() - return private_packet if private_packet == self.buff else fget() + return fget() def set_rt(self, id_, param, val): cmd = f'{id_}.{param}={val}' @@ -150,9 +140,9 @@ class VbanCmd(abc.ABC): self.set_rt('Command', 'Restart', 1) def close(self): - sleep(self._delay) + self._rt_register_socket.close() + self._sendrequest_string_socket.close() self._rt_packet_socket.close() - sys.exit() def __exit__(self, exc_type, exc_value, exc_traceback): self.close()