Rewizja VBAN SendText - single file, zero dependencies

onyx_online zrewidował ten Gist 1737741073. Przejdź do rewizji

1 file changed, 22 insertions, 15 deletions

sendtext.py

			@@ -28,7 +28,7 @@ import logging
28	28		import socket
29	29		import sys
30	30		import time
31		-	from dataclasses import dataclass
	31	+	from dataclasses import dataclass, field
32	32		from pathlib import Path
33	33		from typing import Callable
34	34
			@@ -40,10 +40,29 @@ logger = logging.getLogger(__name__)
40	40		@dataclass
41	41		class RTHeader:
42	42		name: str
43		-	bps_index: int
	43	+	bps: int
44	44		channel: int
45	45		VBAN_PROTOCOL_TXT = 0x40
46	46		framecounter: bytes = (0).to_bytes(4, 'little')
	47	+	# fmt: off
	48	+	BPS_OPTS: list[int] = field(default_factory=lambda: [
	49	+	0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250,
	50	+	38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800, 921600,
	51	+	1000000, 1500000, 2000000, 3000000
	52	+	])
	53	+	# fmt: on
	54	+
	55	+	def __post_init__(self):
	56	+	if len(self.name) > 16:
	57	+	raise ValueError(
	58	+	f"Stream name got: '{self.name}', want: must be 16 characters or fewer"
	59	+	)
	60	+	try:
	61	+	self.bps_index = self.BPS_OPTS.index(self.bps)
	62	+	except ValueError as e:
	63	+	ERR_MSG = f'Invalid bps: {self.bps}, must be one of {self.BPS_OPTS}'
	64	+	e.add_note(ERR_MSG)
	65	+	raise
47	66
48	67		def __sr(self) -> bytes:
49	68		return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
			@@ -108,14 +127,6 @@ def ratelimit(func: Callable) -> Callable:
108	127
109	128
110	129		class VbanSendText:
111		-	# fmt: off
112		-	BPS_OPTS = [
113		-	0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250,
114		-	38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800, 921600,
115		-	1000000, 1500000, 2000000, 3000000
116		-	]
117		-	# fmt: on
118		-
119	130		def __init__(self, **kwargs):
120	131		defaultkwargs = {
121	132		'host': 'localhost',
			@@ -127,11 +138,7 @@ class VbanSendText:
127	138		}
128	139		defaultkwargs.update(kwargs)
129	140		self.__dict__.update(defaultkwargs)
130		-	self._request = RequestPacket(
131		-	RTHeader(
132		-	self.streamname, VbanSendText.BPS_OPTS.index(self.bps), self.channel
133		-	)
134		-	)
	141	+	self._request = RequestPacket(RTHeader(self.streamname, self.bps, self.channel))
135	142		self.lastsent = 0
136	143
137	144		def __enter__(self):

onyx_online zrewidował ten Gist 1737739968. Przejdź do rewizji

1 file changed, 2 insertions

sendtext.py

			@@ -23,6 +23,7 @@
23	23		# SOFTWARE.
24	24
25	25		import argparse
	26	+	import functools
26	27		import logging
27	28		import socket
28	29		import sys
			@@ -95,6 +96,7 @@ def ratelimit(func: Callable) -> Callable:
95	96		pass
96	97		"""
97	98
	99	+	@functools.wraps(func)
98	100		def wrapper(self, args, *kwargs):
99	101		now = time.time()
100	102		if now - self.lastsent < self.delay:

onyx_online zrewidował ten Gist 1737652391. Przejdź do rewizji

1 file changed, 12 insertions, 2 deletions

sendtext.py

			@@ -106,19 +106,29 @@ def ratelimit(func: Callable) -> Callable:
106	106
107	107
108	108		class VbanSendText:
	109	+	# fmt: off
	110	+	BPS_OPTS = [
	111	+	0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250,
	112	+	38400, 57600, 115200, 128000, 230400, 250000, 256000, 460800, 921600,
	113	+	1000000, 1500000, 2000000, 3000000
	114	+	]
	115	+	# fmt: on
	116	+
109	117		def __init__(self, **kwargs):
110	118		defaultkwargs = {
111	119		'host': 'localhost',
112	120		'port': 6980,
113	121		'streamname': 'Command1',
114		-	'bps_index': 0,
	122	+	'bps': 256000,
115	123		'channel': 0,
116	124		'delay': 0.02,
117	125		}
118	126		defaultkwargs.update(kwargs)
119	127		self.__dict__.update(defaultkwargs)
120	128		self._request = RequestPacket(
121		-	RTHeader(self.streamname, self.bps_index, self.channel)
	129	+	RTHeader(
	130	+	self.streamname, VbanSendText.BPS_OPTS.index(self.bps), self.channel
	131	+	)
122	132		)
123	133		self.lastsent = 0
124	134

onyx_online zrewidował ten Gist 1737603186. Przejdź do rewizji

1 file changed, 1 insertion, 1 deletion

sendtext.py

			@@ -182,7 +182,7 @@ def parse_args() -> argparse.Namespace:
182	182		text (str, optional): Text to send.
183	183		"""
184	184
185		-	parser = argparse.ArgumentParser(description='Send text to VBAN')
	185	+	parser = argparse.ArgumentParser(description='Voicemeeter VBAN Send Text CLI')
186	186		parser.add_argument(
187	187		'--log-level',
188	188		type=str,

onyx_online zrewidował ten Gist 1737597025. Przejdź do rewizji

1 file changed, 33 insertions, 33 deletions

sendtext.py

			@@ -42,20 +42,20 @@ class RTHeader:
42	42		bps_index: int
43	43		channel: int
44	44		VBAN_PROTOCOL_TXT = 0x40
45		-	framecounter: bytes = (0).to_bytes(4, "little")
	45	+	framecounter: bytes = (0).to_bytes(4, 'little')
46	46
47	47		def __sr(self) -> bytes:
48		-	return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, "little")
	48	+	return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
49	49
50	50		def __nbc(self) -> bytes:
51		-	return (self.channel).to_bytes(1, "little")
	51	+	return (self.channel).to_bytes(1, 'little')
52	52
53	53		def build(self) -> bytes:
54		-	header = "VBAN".encode("utf-8")
	54	+	header = 'VBAN'.encode('utf-8')
55	55		header += self.__sr()
56		-	header += (0).to_bytes(1, "little")
	56	+	header += (0).to_bytes(1, 'little')
57	57		header += self.__nbc()
58		-	header += (0x10).to_bytes(1, "little")
	58	+	header += (0x10).to_bytes(1, 'little')
59	59		header += self.name.encode() + bytes(16 - len(self.name))
60	60		header += RTHeader.framecounter
61	61		return header
			@@ -66,15 +66,15 @@ class RequestPacket:
66	66		self.header = header
67	67
68	68		def encode(self, text: str) -> bytes:
69		-	return self.header.build() + text.encode("utf-8")
	69	+	return self.header.build() + text.encode('utf-8')
70	70
71	71		def bump_framecounter(self) -> None:
72	72		self.header.framecounter = (
73		-	int.from_bytes(self.header.framecounter, "little") + 1
74		-	).to_bytes(4, "little")
	73	+	int.from_bytes(self.header.framecounter, 'little') + 1
	74	+	).to_bytes(4, 'little')
75	75
76	76		logger.debug(
77		-	f"framecounter: {int.from_bytes(self.header.framecounter, 'little')}"
	77	+	f'framecounter: {int.from_bytes(self.header.framecounter, "little")}'
78	78		)
79	79
80	80
			@@ -108,12 +108,12 @@ def ratelimit(func: Callable) -> Callable:
108	108		class VbanSendText:
109	109		def __init__(self, **kwargs):
110	110		defaultkwargs = {
111		-	"host": "localhost",
112		-	"port": 6980,
113		-	"streamname": "Command1",
114		-	"bps_index": 0,
115		-	"channel": 0,
116		-	"delay": 0.02,
	111	+	'host': 'localhost',
	112	+	'port': 6980,
	113	+	'streamname': 'Command1',
	114	+	'bps_index': 0,
	115	+	'channel': 0,
	116	+	'delay': 0.02,
117	117		}
118	118		defaultkwargs.update(kwargs)
119	119		self.__dict__.update(defaultkwargs)
			@@ -142,7 +142,7 @@ class VbanSendText:
142	142		self._request.bump_framecounter()
143	143
144	144
145		-	def conn_from_toml(filepath: str = "config.toml") -> dict:
	145	+	def conn_from_toml(filepath: str = 'config.toml') -> dict:
146	146		"""
147	147		Reads a TOML configuration file and returns its contents as a dictionary.
148	148		Args:
			@@ -159,15 +159,15 @@ def conn_from_toml(filepath: str = "config.toml") -> dict:
159	159		pn = Path(filepath)
160	160		if not pn.exists():
161	161		logger.info(
162		-	f"no {pn} found, using defaults: localhost:6980 streamname: Command1"
	162	+	f'no {pn} found, using defaults: localhost:6980 streamname: Command1'
163	163		)
164	164		return {}
165	165
166	166		try:
167		-	with open(pn, "rb") as f:
	167	+	with open(pn, 'rb') as f:
168	168		return tomllib.load(f)
169	169		except tomllib.TOMLDecodeError as e:
170		-	raise ValueError(f"Error decoding TOML file: {e}") from e
	170	+	raise ValueError(f'Error decoding TOML file: {e}') from e
171	171
172	172
173	173		def parse_args() -> argparse.Namespace:
			@@ -182,24 +182,24 @@ def parse_args() -> argparse.Namespace:
182	182		text (str, optional): Text to send.
183	183		"""
184	184
185		-	parser = argparse.ArgumentParser(description="Send text to VBAN")
	185	+	parser = argparse.ArgumentParser(description='Send text to VBAN')
186	186		parser.add_argument(
187		-	"--log-level",
	187	+	'--log-level',
188	188		type=str,
189		-	choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
190		-	default="INFO",
191		-	help="Set the logging level",
	189	+	choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
	190	+	default='INFO',
	191	+	help='Set the logging level',
192	192		)
193	193		parser.add_argument(
194		-	"--config", type=str, default="config.toml", help="Path to config file"
	194	+	'--config', type=str, default='config.toml', help='Path to config file'
195	195		)
196	196		parser.add_argument(
197		-	"-i",
198		-	"--input-file",
199		-	type=argparse.FileType("r"),
	197	+	'-i',
	198	+	'--input-file',
	199	+	type=argparse.FileType('r'),
200	200		default=sys.stdin,
201	201		)
202		-	parser.add_argument("text", nargs="?", type=str, help="Text to send")
	202	+	parser.add_argument('text', nargs='?', type=str, help='Text to send')
203	203		return parser.parse_args()
204	204
205	205
			@@ -222,14 +222,14 @@ def main(config: dict):
222	222
223	223		for line in args.input_file:
224	224		line = line.strip()
225		-	if line.upper() == "Q":
	225	+	if line.upper() == 'Q':
226	226		break
227	227
228		-	logger.debug(f"Sending {line}")
	228	+	logger.debug(f'Sending {line}')
229	229		vban.sendtext(line)
230	230
231	231
232		-	if __name__ == "__main__":
	232	+	if __name__ == '__main__':
233	233		args = parse_args()
234	234
235	235		logging.basicConfig(level=args.log_level)

onyx_online zrewidował ten Gist 1737596796. Przejdź do rewizji

1 file changed, 3 insertions, 3 deletions

sendtext.py

			@@ -44,10 +44,10 @@ class RTHeader:
44	44		VBAN_PROTOCOL_TXT = 0x40
45	45		framecounter: bytes = (0).to_bytes(4, "little")
46	46
47		-	def __sr(self):
	47	+	def __sr(self) -> bytes:
48	48		return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, "little")
49	49
50		-	def __nbc(self):
	50	+	def __nbc(self) -> bytes:
51	51		return (self.channel).to_bytes(1, "little")
52	52
53	53		def build(self) -> bytes:
			@@ -68,7 +68,7 @@ class RequestPacket:
68	68		def encode(self, text: str) -> bytes:
69	69		return self.header.build() + text.encode("utf-8")
70	70
71		-	def bump_framecounter(self):
	71	+	def bump_framecounter(self) -> None:
72	72		self.header.framecounter = (
73	73		int.from_bytes(self.header.framecounter, "little") + 1
74	74		).to_bytes(4, "little")

onyx_online zrewidował ten Gist 1737596684. Przejdź do rewizji

1 file changed, 41 insertions, 7 deletions

sendtext.py

			@@ -29,6 +29,7 @@ import sys
29	29		import time
30	30		from dataclasses import dataclass
31	31		from pathlib import Path
	32	+	from typing import Callable
32	33
33	34		import tomllib
34	35
			@@ -67,6 +68,42 @@ class RequestPacket:
67	68		def encode(self, text: str) -> bytes:
68	69		return self.header.build() + text.encode("utf-8")
69	70
	71	+	def bump_framecounter(self):
	72	+	self.header.framecounter = (
	73	+	int.from_bytes(self.header.framecounter, "little") + 1
	74	+	).to_bytes(4, "little")
	75	+
	76	+	logger.debug(
	77	+	f"framecounter: {int.from_bytes(self.header.framecounter, 'little')}"
	78	+	)
	79	+
	80	+
	81	+	def ratelimit(func: Callable) -> Callable:
	82	+	"""
	83	+	Decorator to enforce a rate limit on a function.
	84	+	This decorator ensures that the decorated function is not called more frequently
	85	+	than the specified delay. If the function is called before the delay has passed
	86	+	since the last call, it will wait for the remaining time before executing.
	87	+	Args:
	88	+	func (callable): The function to be decorated.
	89	+	Returns:
	90	+	callable: The wrapped function with rate limiting applied.
	91	+	Example:
	92	+	@ratelimit
	93	+	def send_message(self, message):
	94	+	# Function implementation
	95	+	pass
	96	+	"""
	97	+
	98	+	def wrapper(self, args, *kwargs):
	99	+	now = time.time()
	100	+	if now - self.lastsent < self.delay:
	101	+	time.sleep(self.delay - (now - self.lastsent))
	102	+	self.lastsent = time.time()
	103	+	return func(self, args, *kwargs)
	104	+
	105	+	return wrapper
	106	+
70	107
71	108		class VbanSendText:
72	109		def __init__(self, **kwargs):
			@@ -83,6 +120,7 @@ class VbanSendText:
83	120		self._request = RequestPacket(
84	121		RTHeader(self.streamname, self.bps_index, self.channel)
85	122		)
	123	+	self.lastsent = 0
86	124
87	125		def __enter__(self):
88	126		self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
			@@ -91,6 +129,7 @@ class VbanSendText:
91	129		def __exit__(self, exc_type, exc_value, traceback):
92	130		self._sock.close()
93	131
	132	+	@ratelimit
94	133		def sendtext(self, text: str):
95	134		"""
96	135		Sends a text message to the specified host and port.
			@@ -99,13 +138,8 @@ class VbanSendText:
99	138		"""
100	139
101	140		self._sock.sendto(self._request.encode(text), (self.host, self.port))
102		-	self._request.header.framecounter = (
103		-	int.from_bytes(self._request.header.framecounter, "little") + 1
104		-	).to_bytes(4, "little")
105		-	logger.debug(
106		-	f"framecounter: {int.from_bytes(self._request.header.framecounter, 'little')}"
107		-	)
108		-	time.sleep(self.delay)
	141	+
	142	+	self._request.bump_framecounter()
109	143
110	144
111	145		def conn_from_toml(filepath: str = "config.toml") -> dict:

onyx_online zrewidował ten Gist 1737589148. Przejdź do rewizji

1 file changed, 1 insertion, 1 deletion

sendtext.py

			@@ -130,7 +130,7 @@ def conn_from_toml(filepath: str = "config.toml") -> dict:
130	130		return {}
131	131
132	132		try:
133		-	with open(filepath, "rb") as f:
	133	+	with open(pn, "rb") as f:
134	134		return tomllib.load(f)
135	135		except tomllib.TOMLDecodeError as e:
136	136		raise ValueError(f"Error decoding TOML file: {e}") from e

onyx_online zrewidował ten Gist 1737589103. Przejdź do rewizji

1 file changed, 1 insertion, 1 deletion

sendtext.py

			@@ -125,7 +125,7 @@ def conn_from_toml(filepath: str = "config.toml") -> dict:
125	125		pn = Path(filepath)
126	126		if not pn.exists():
127	127		logger.info(
128		-	"no config.toml found, using defaults: localhost:6980 streamname: Command1"
	128	+	f"no {pn} found, using defaults: localhost:6980 streamname: Command1"
129	129		)
130	130		return {}
131	131

onyx_online zrewidował ten Gist 1737588973. Przejdź do rewizji

1 file changed, 13 insertions, 2 deletions

sendtext.py

			@@ -28,6 +28,7 @@ import socket
28	28		import sys
29	29		import time
30	30		from dataclasses import dataclass
	31	+	from pathlib import Path
31	32
32	33		import tomllib
33	34
			@@ -121,8 +122,18 @@ def conn_from_toml(filepath: str = "config.toml") -> dict:
121	122		streamname = "Command1"
122	123		"""
123	124
124		-	with open(filepath, "rb") as f:
125		-	return tomllib.load(f)
	125	+	pn = Path(filepath)
	126	+	if not pn.exists():
	127	+	logger.info(
	128	+	"no config.toml found, using defaults: localhost:6980 streamname: Command1"
	129	+	)
	130	+	return {}
	131	+
	132	+	try:
	133	+	with open(filepath, "rb") as f:
	134	+	return tomllib.load(f)
	135	+	except tomllib.TOMLDecodeError as e:
	136	+	raise ValueError(f"Error decoding TOML file: {e}") from e
126	137
127	138
128	139		def parse_args() -> argparse.Namespace:

Nowsze Starsze