Tetszik 0
Fork 0

Utoljára aktív 1772128893

Use this script to send vban text requests over a network

onyx_online's Avatar onyx_online gist felülvizsgálása 1772128892. Revízióhoz ugrás

1 file changed, 41 insertions, 40 deletions

sendtext.py

@@ -111,29 +111,29 @@ class RTHeader:
111 111 try:
112 112 self.bps_index = self.BPS_OPTS.index(self.bps)
113 113 except ValueError as e:
114 - ERR_MSG = f"Invalid bps: {self.bps}, must be one of {self.BPS_OPTS}"
114 + ERR_MSG = f'Invalid bps: {self.bps}, must be one of {self.BPS_OPTS}'
115 115 e.add_note(ERR_MSG)
116 116 raise
117 117
118 118 @property
119 119 def vban(self) -> bytes:
120 - return b"VBAN"
120 + return b'VBAN'
121 121
122 122 @property
123 123 def sr(self) -> bytes:
124 - return (0x40 + self.bps_index).to_bytes(1, "little")
124 + return (0x40 + self.bps_index).to_bytes(1, 'little')
125 125
126 126 @property
127 127 def nbs(self) -> bytes:
128 - return (0).to_bytes(1, "little")
128 + return (0).to_bytes(1, 'little')
129 129
130 130 @property
131 131 def nbc(self) -> bytes:
132 - return (self.channel).to_bytes(1, "little")
132 + return (self.channel).to_bytes(1, 'little')
133 133
134 134 @property
135 135 def bit(self) -> bytes:
136 - return (0x10).to_bytes(1, "little")
136 + return (0x10).to_bytes(1, 'little')
137 137
138 138 @property
139 139 def streamname(self) -> bytes:
@@ -150,7 +150,7 @@ class RTHeader:
150 150 data.extend(header.nbc)
151 151 data.extend(header.bit)
152 152 data.extend(header.streamname)
153 - data.extend(header.framecounter.to_bytes(4, "little"))
153 + data.extend(header.framecounter.to_bytes(4, 'little'))
154 154 return bytes(data)
155 155
156 156
@@ -181,7 +181,7 @@ def ratelimit(func):
181 181
182 182 if time_since_last_call < ratelimit:
183 183 sleep_time = ratelimit - time_since_last_call
184 - logger.debug(f"Rate limiting: sleeping for {sleep_time:.3f} seconds")
184 + logger.debug(f'Rate limiting: sleeping for {sleep_time:.3f} seconds')
185 185 time.sleep(sleep_time)
186 186
187 187 last_call_time[0] = time.time()
@@ -212,8 +212,8 @@ def send(
212 212 bps=args.bps,
213 213 channel=args.channel,
214 214 framecounter=framecounter,
215 - ) + cmd.encode("utf-8")
216 - logger.debug("Sending packet: %s", raw_packet)
215 + ) + cmd.encode('utf-8')
216 + logger.debug('Sending packet: %s', raw_packet)
217 217 sock.sendto(raw_packet, (args.host, args.port))
218 218
219 219
@@ -231,59 +231,59 @@ def parse_args() -> argparse.Namespace:
231 231 text: Text to send (positional argument)
232 232 """
233 233
234 - parser = argparse.ArgumentParser(description="Voicemeeter VBAN Send Text CLI")
234 + parser = argparse.ArgumentParser(description='Voicemeeter VBAN Send Text CLI')
235 235 parser.add_argument(
236 - "--host",
237 - "-H",
236 + '--host',
237 + '-H',
238 238 type=str,
239 - default="localhost",
240 - help="VBAN host to send to (default: localhost)",
239 + default='localhost',
240 + help='VBAN host to send to (default: localhost)',
241 241 )
242 242 parser.add_argument(
243 - "--port",
244 - "-P",
243 + '--port',
244 + '-P',
245 245 type=int,
246 246 default=6980,
247 - help="VBAN port to send to (default: 6980)",
247 + help='VBAN port to send to (default: 6980)',
248 248 )
249 249 parser.add_argument(
250 - "--streamname",
251 - "-s",
250 + '--streamname',
251 + '-s',
252 252 type=str,
253 - default="Command1",
254 - help="VBAN stream name (default: Command1)",
253 + default='Command1',
254 + help='VBAN stream name (default: Command1)',
255 255 )
256 256 parser.add_argument(
257 - "--bps",
258 - "-b",
257 + '--bps',
258 + '-b',
259 259 type=int,
260 260 default=256000,
261 - help="Bits per second for VBAN stream (default: 256000)",
261 + help='Bits per second for VBAN stream (default: 256000)',
262 262 )
263 263 parser.add_argument(
264 - "--channel",
265 - "-c",
264 + '--channel',
265 + '-c',
266 266 type=int,
267 267 default=1,
268 - help="Channel number for VBAN stream (default: 1)",
268 + help='Channel number for VBAN stream (default: 1)',
269 269 )
270 270 parser.add_argument(
271 - "--ratelimit",
272 - "-r",
271 + '--ratelimit',
272 + '-r',
273 273 type=float,
274 274 default=0.02,
275 - help="Minimum time in seconds between sending messages (default: 0.02)",
275 + help='Minimum time in seconds between sending messages (default: 0.02)',
276 276 )
277 277 parser.add_argument(
278 - "--loglevel",
279 - "-l",
278 + '--loglevel',
279 + '-l',
280 280 type=str,
281 - default="info",
282 - choices=["debug", "info", "warning", "error", "critical"],
283 - help="Set the logging level (default: info)",
281 + default='info',
282 + choices=['debug', 'info', 'warning', 'error', 'critical'],
283 + help='Set the logging level (default: info)',
284 284 )
285 285 parser.add_argument(
286 - "cmds", nargs="+", type=str, help="Text to send (positional argument)"
286 + 'cmds', nargs='+', type=str, help='Text to send (positional argument)'
287 287 )
288 288 return parser.parse_args()
289 289
@@ -292,18 +292,19 @@ def main(args: argparse.Namespace):
292 292 """
293 293 Main function to send text using VBAN.
294 294 Args:
295 - args (argparse.Namespace): Parsed command-line arguments.
295 + args (argparse.Namespace): The command-line arguments containing
296 + host, port, stream name, bits per second, channel, and text commands to send.
296 297 Behavior:
297 298 Creates a UDP socket and sends each command in 'args.cmds' to the specified VBAN host and port using the 'send' function, with rate limiting applied.
298 299 """
299 300
300 301 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
301 302 for n, cmd in enumerate(args.cmds):
302 - logger.debug(f"Sending: {cmd}")
303 + logger.debug(f'Sending: {cmd}')
303 304 send(sock, args, cmd, n)
304 305
305 306
306 - if __name__ == "__main__":
307 + if __name__ == '__main__':
307 308 args = parse_args()
308 309
309 310 logging.basicConfig(level=getattr(logging, args.loglevel.upper()))

onyx_online's Avatar onyx_online gist felülvizsgálása 1771937379. Revízióhoz ugrás

1 file changed, 2 insertions, 2 deletions

sendtext.py

@@ -271,8 +271,8 @@ def parse_args() -> argparse.Namespace:
271 271 "--ratelimit",
272 272 "-r",
273 273 type=float,
274 - default=0.2,
275 - help="Minimum time in seconds between sending messages (default: 0.2)",
274 + default=0.02,
275 + help="Minimum time in seconds between sending messages (default: 0.02)",
276 276 )
277 277 parser.add_argument(
278 278 "--loglevel",

onyx_online's Avatar onyx_online gist felülvizsgálása 1771936920. Revízióhoz ugrás

1 file changed, 1 insertion, 3 deletions

sendtext.py

@@ -173,8 +173,6 @@ def ratelimit(func):
173 173
174 174 @functools.wraps(func)
175 175 def wrapper(*args, **kwargs):
176 - resp = func(*args, **kwargs)
177 -
178 176 _, args_namespace, _, _ = args
179 177 ratelimit = args_namespace.ratelimit
180 178
@@ -187,7 +185,7 @@ def ratelimit(func):
187 185 time.sleep(sleep_time)
188 186
189 187 last_call_time[0] = time.time()
190 - return resp
188 + return func(*args, **kwargs)
191 189
192 190 return wrapper
193 191

onyx_online's Avatar onyx_online gist felülvizsgálása 1771936185. Revízióhoz ugrás

1 file changed, 1 insertion, 1 deletion

sendtext.py

@@ -49,7 +49,7 @@ Usage:
49 49 bash:
50 50 xargs ./sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 < commands.txt
51 51 powershell:
52 - ./sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 < (Get-Content commands.txt)
52 + ./sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 (Get-Content commands.txt)
53 53 Messages to Matrix are also possible:
54 54 python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "Point(ASIO128.IN[2],ASIO128.OUT[1]).dBGain = -8"
55 55 """

onyx_online's Avatar onyx_online gist felülvizsgálása 1771936138. Revízióhoz ugrás

1 file changed, 48 insertions, 39 deletions

sendtext.py

@@ -42,9 +42,16 @@ Usage:
42 42 Run the script with appropriate command-line arguments to send text messages.
43 43 Example:
44 44 To send a single message:
45 - python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1"
45 + python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1;strip[1].mute=1;strip[2].mute=1"
46 46 To send multiple messages:
47 - python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1" "strip[1].mute=0"
47 + python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1;strip[1].mute=1;strip[2].mute=1" "bus[0].mute=1;bus[1].mute=1;bus[2].mute=1"
48 + To send commands stored in a text file:
49 + bash:
50 + xargs ./sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 < commands.txt
51 + powershell:
52 + ./sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 < (Get-Content commands.txt)
53 + Messages to Matrix are also possible:
54 + python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "Point(ASIO128.IN[2],ASIO128.OUT[1]).dBGain = -8"
48 55 """
49 56
50 57 import argparse
@@ -57,42 +64,6 @@ from dataclasses import dataclass, field
57 64 logger = logging.getLogger(__name__)
58 65
59 66
60 - def ratelimit(func):
61 - """
62 - Decorator to enforce a rate limit on a function.
63 -
64 - This decorator extracts the rate limit value from the 'args.ratelimit' parameter
65 - of the decorated function and ensures that the function is not called more
66 - frequently than the specified rate limit.
67 -
68 - Args:
69 - func: The function to be rate limited. Must accept 'args' as a parameter
70 - with a 'ratelimit' attribute.
71 -
72 - Returns:
73 - The decorated function with rate limiting applied.
74 - """
75 - last_call_time = [0.0] # Use list to make it mutable in closure
76 -
77 - @functools.wraps(func)
78 - def wrapper(*args, **kwargs):
79 - # Extract args.ratelimit from function parameters
80 - rate_limit = getattr(args[1], "ratelimit", 0.0) if len(args) > 1 else 0.0
81 -
82 - current_time = time.time()
83 - time_since_last_call = current_time - last_call_time[0]
84 -
85 - if time_since_last_call < rate_limit:
86 - sleep_time = rate_limit - time_since_last_call
87 - logger.debug(f"Rate limiting: sleeping for {sleep_time:.3f} seconds")
88 - time.sleep(sleep_time)
89 -
90 - last_call_time[0] = time.time()
91 - return func(*args, **kwargs)
92 -
93 - return wrapper
94 -
95 -
96 67 @dataclass
97 68 class RTHeader:
98 69 """RTHeader represents the header of a VBAN packet for sending text messages.
@@ -183,6 +154,44 @@ class RTHeader:
183 154 return bytes(data)
184 155
185 156
157 + def ratelimit(func):
158 + """
159 + Decorator to enforce a rate limit on a function.
160 +
161 + This decorator extracts the rate limit value from the 'args.ratelimit' parameter
162 + of the decorated function and ensures that the function is not called more
163 + frequently than the specified rate limit.
164 +
165 + Args:
166 + func: The function to be rate limited. Must accept 'args' as a parameter
167 + with a 'ratelimit' attribute.
168 +
169 + Returns:
170 + The decorated function with rate limiting applied.
171 + """
172 + last_call_time = [0.0] # Use list to make it mutable in closure
173 +
174 + @functools.wraps(func)
175 + def wrapper(*args, **kwargs):
176 + resp = func(*args, **kwargs)
177 +
178 + _, args_namespace, _, _ = args
179 + ratelimit = args_namespace.ratelimit
180 +
181 + current_time = time.time()
182 + time_since_last_call = current_time - last_call_time[0]
183 +
184 + if time_since_last_call < ratelimit:
185 + sleep_time = ratelimit - time_since_last_call
186 + logger.debug(f"Rate limiting: sleeping for {sleep_time:.3f} seconds")
187 + time.sleep(sleep_time)
188 +
189 + last_call_time[0] = time.time()
190 + return resp
191 +
192 + return wrapper
193 +
194 +
186 195 @ratelimit
187 196 def send(
188 197 sock: socket.socket, args: argparse.Namespace, cmd: str, framecounter: int
@@ -301,4 +310,4 @@ if __name__ == "__main__":
301 310
302 311 logging.basicConfig(level=getattr(logging, args.loglevel.upper()))
303 312
304 - main(args)
313 + main(args)

onyx_online's Avatar onyx_online gist felülvizsgálása 1771934403. Revízióhoz ugrás

Nincsenek változtatások

onyx_online's Avatar onyx_online gist felülvizsgálása 1771934351. Revízióhoz ugrás

1 file changed, 187 insertions, 165 deletions

sendtext.py

@@ -30,44 +30,100 @@ It includes classes and functions to construct and send VBAN packets with text d
30 30 from a TOML file, and handle command-line arguments for customization.
31 31
32 32 Classes:
33 - RTHeader: Represents the VBAN header for a text stream.
34 - RequestPacket: Encapsulates a VBAN request packet with text data.
35 - VbanSendText: Manages the sending of text messages over VBAN with rate limiting.
33 + RTHeader: A dataclass representing the header of a VBAN packet, with methods to convert it to bytes.
36 34
37 35 Functions:
38 - ratelimit: Decorator to enforce a rate limit on a function.
39 - conn_from_toml: Reads a TOML configuration file and returns its contents as a dictionary.
40 - parse_args: Parses command-line arguments.
41 - main: Main function to send text using VbanSendText.
36 + ratelimit(func): A decorator to enforce a rate limit on a function.
37 + send(sock, args, cmd, framecounter): Sends a text message using the provided socket and packet.
38 + parse_args(): Parses command-line arguments and returns them as an argparse.Namespace object.
39 + main(args): Main function to send text using VbanSendText.
42 40
43 41 Usage:
44 42 Run the script with appropriate command-line arguments to send text messages.
45 43 Example:
46 - python sendtext.py --config /path/to/config.toml --log-level DEBUG "strip[0].mute=0 strip[1].mute=0"
44 + To send a single message:
45 + python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1"
46 + To send multiple messages:
47 + python sendtext.py --host localhost --port 6980 --streamname Command1 --bps 256000 --channel 1 "strip[0].mute=1" "strip[1].mute=0"
47 48 """
48 49
49 50 import argparse
50 51 import functools
51 52 import logging
52 53 import socket
53 - import sys
54 54 import time
55 55 from dataclasses import dataclass, field
56 - from pathlib import Path
57 - from typing import Callable
58 -
59 - import tomllib
60 56
61 57 logger = logging.getLogger(__name__)
62 58
63 59
60 + def ratelimit(func):
61 + """
62 + Decorator to enforce a rate limit on a function.
63 +
64 + This decorator extracts the rate limit value from the 'args.ratelimit' parameter
65 + of the decorated function and ensures that the function is not called more
66 + frequently than the specified rate limit.
67 +
68 + Args:
69 + func: The function to be rate limited. Must accept 'args' as a parameter
70 + with a 'ratelimit' attribute.
71 +
72 + Returns:
73 + The decorated function with rate limiting applied.
74 + """
75 + last_call_time = [0.0] # Use list to make it mutable in closure
76 +
77 + @functools.wraps(func)
78 + def wrapper(*args, **kwargs):
79 + # Extract args.ratelimit from function parameters
80 + rate_limit = getattr(args[1], "ratelimit", 0.0) if len(args) > 1 else 0.0
81 +
82 + current_time = time.time()
83 + time_since_last_call = current_time - last_call_time[0]
84 +
85 + if time_since_last_call < rate_limit:
86 + sleep_time = rate_limit - time_since_last_call
87 + logger.debug(f"Rate limiting: sleeping for {sleep_time:.3f} seconds")
88 + time.sleep(sleep_time)
89 +
90 + last_call_time[0] = time.time()
91 + return func(*args, **kwargs)
92 +
93 + return wrapper
94 +
95 +
64 96 @dataclass
65 97 class RTHeader:
98 + """RTHeader represents the header of a VBAN packet for sending text messages.
99 + It includes fields for the stream name, bits per second, channel, and frame counter,
100 + as well as methods to convert these fields into the appropriate byte format for transmission.
101 +
102 + Attributes:
103 + name (str): The name of the VBAN stream (max 16 characters).
104 + bps (int): The bits per second for the VBAN stream, must be one of the predefined options.
105 + channel (int): The channel number for the VBAN stream.
106 + framecounter (int): A counter for the frames being sent, default is 0.
107 + BPS_OPTS (list[int]): A list of valid bits per second options for VBAN streams.
108 +
109 + Methods:
110 + __post_init__(): Validates the stream name length and bits per second value.
111 + vban(): Returns the VBAN header as bytes.
112 + sr(): Returns the sample rate byte based on the bits per second index.
113 + nbs(): Returns the number of bits per sample byte (currently set to 0).
114 + nbc(): Returns the number of channels byte based on the channel attribute.
115 + bit(): Returns the bit depth byte (currently set to 0x10).
116 + streamname(): Returns the stream name as bytes, padded to 16 bytes.
117 + to_bytes(name, bps, channel, framecounter): Class method to create a byte representation of the RTHeader with the given parameters.
118 +
119 + Raises:
120 + ValueError: If the stream name exceeds 16 characters or if the bits per second value is not in the predefined options.
121 + """
122 +
66 123 name: str
67 124 bps: int
68 125 channel: int
69 - VBAN_PROTOCOL_TXT = 0x40
70 - framecounter: bytes = (0).to_bytes(4, "little")
126 + framecounter: int = 0
71 127 # fmt: off
72 128 BPS_OPTS: list[int] = field(default_factory=lambda: [
73 129 0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250,
@@ -88,131 +144,70 @@ class RTHeader:
88 144 e.add_note(ERR_MSG)
89 145 raise
90 146
91 - def __sr(self) -> bytes:
92 - return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, "little")
93 -
94 - def __nbc(self) -> bytes:
95 - return (self.channel).to_bytes(1, "little")
96 -
97 - def build(self) -> bytes:
98 - header = "VBAN".encode("utf-8")
99 - header += self.__sr()
100 - header += (0).to_bytes(1, "little")
101 - header += self.__nbc()
102 - header += (0x10).to_bytes(1, "little")
103 - header += self.name.encode() + bytes(16 - len(self.name))
104 - header += RTHeader.framecounter
105 - return header
106 -
107 -
108 - class RequestPacket:
109 - def __init__(self, header: RTHeader):
110 - self.header = header
147 + @property
148 + def vban(self) -> bytes:
149 + return b"VBAN"
111 150
112 - def encode(self, text: str) -> bytes:
113 - return self.header.build() + text.encode("utf-8")
151 + @property
152 + def sr(self) -> bytes:
153 + return (0x40 + self.bps_index).to_bytes(1, "little")
114 154
115 - def bump_framecounter(self) -> None:
116 - self.header.framecounter = (
117 - int.from_bytes(self.header.framecounter, "little") + 1
118 - ).to_bytes(4, "little")
119 -
120 - logger.debug(
121 - f"framecounter: {int.from_bytes(self.header.framecounter, 'little')}"
122 - )
155 + @property
156 + def nbs(self) -> bytes:
157 + return (0).to_bytes(1, "little")
123 158
159 + @property
160 + def nbc(self) -> bytes:
161 + return (self.channel).to_bytes(1, "little")
124 162
125 - def ratelimit(func: Callable) -> Callable:
163 + @property
164 + def bit(self) -> bytes:
165 + return (0x10).to_bytes(1, "little")
166 +
167 + @property
168 + def streamname(self) -> bytes:
169 + return self.name.encode() + bytes(16 - len(self.name))
170 +
171 + @classmethod
172 + def to_bytes(cls, name: str, bps: int, channel: int, framecounter: int) -> bytes:
173 + header = cls(name=name, bps=bps, channel=channel, framecounter=framecounter)
174 +
175 + data = bytearray()
176 + data.extend(header.vban)
177 + data.extend(header.sr)
178 + data.extend(header.nbs)
179 + data.extend(header.nbc)
180 + data.extend(header.bit)
181 + data.extend(header.streamname)
182 + data.extend(header.framecounter.to_bytes(4, "little"))
183 + return bytes(data)
184 +
185 +
186 + @ratelimit
187 + def send(
188 + sock: socket.socket, args: argparse.Namespace, cmd: str, framecounter: int
189 + ) -> None:
126 190 """
127 - Decorator to enforce a rate limit on a function.
128 - This decorator ensures that the decorated function is not called more frequently
129 - than the specified delay. If the function is called before the delay has passed
130 - since the last call, it will wait for the remaining time before executing.
131 - Args:
132 - func (callable): The function to be decorated.
133 - Returns:
134 - callable: The wrapped function with rate limiting applied.
135 - Example:
136 - @ratelimit
137 - def send_message(self, message):
138 - # Function implementation
139 - pass
140 - """
141 -
142 - @functools.wraps(func)
143 - def wrapper(self, *args, **kwargs):
144 - now = time.time()
145 - if now - self.lastsent < self.delay:
146 - time.sleep(self.delay - (now - self.lastsent))
147 - self.lastsent = time.time()
148 - return func(self, *args, **kwargs)
149 -
150 - return wrapper
151 -
191 + Send a text message using the provided socket and packet.
152 192
153 - class VbanSendText:
154 - def __init__(self, **kwargs):
155 - defaultkwargs = {
156 - "host": "localhost",
157 - "port": 6980,
158 - "streamname": "Command1",
159 - "bps": 256000,
160 - "channel": 0,
161 - "delay": 0.02,
162 - }
163 - defaultkwargs.update(kwargs)
164 - self.__dict__.update(defaultkwargs)
165 - self._request = RequestPacket(RTHeader(self.streamname, self.bps, self.channel))
166 - self.lastsent = 0
167 -
168 - def __enter__(self):
169 - self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
170 - return self
171 -
172 - def __exit__(self, exc_type, exc_value, traceback):
173 - if self._sock:
174 - self._sock.close()
175 - self._sock = None
176 -
177 - @ratelimit
178 - def sendtext(self, text: str):
179 - """
180 - Sends a text message to the specified host and port.
181 - Args:
182 - text (str): The text message to be sent.
183 - """
184 -
185 - self._sock.sendto(self._request.encode(text), (self.host, self.port))
186 -
187 - self._request.bump_framecounter()
188 -
189 -
190 - def conn_from_toml(filepath: str = "config.toml") -> dict:
191 - """
192 - Reads a TOML configuration file and returns its contents as a dictionary.
193 193 Args:
194 - filepath (str): The path to the TOML file. Defaults to "config.toml".
194 + sock (socket.socket): The socket to use for sending the message.
195 + args (argparse.Namespace): The command-line arguments containing the stream name, bits per second, and channel information.
196 + cmd (str): The text command to send.
197 + framecounter (int): The frame counter to include in the VBAN header.
198 +
195 199 Returns:
196 - dict: The contents of the TOML file as a dictionary.
197 - Example:
198 - # config.toml
199 - host = "localhost"
200 - port = 6980
201 - streamname = "Command1"
200 + None
202 201 """
203 202
204 - pn = Path(filepath)
205 - if not pn.exists():
206 - logger.info(
207 - f"no {pn} found, using defaults: localhost:6980 streamname: Command1"
208 - )
209 - return {}
210 -
211 - try:
212 - with open(pn, "rb") as f:
213 - return tomllib.load(f)
214 - except tomllib.TOMLDecodeError as e:
215 - raise ValueError(f"Error decoding TOML file: {e}") from e
203 + raw_packet = RTHeader.to_bytes(
204 + name=args.streamname,
205 + bps=args.bps,
206 + channel=args.channel,
207 + framecounter=framecounter,
208 + ) + cmd.encode("utf-8")
209 + logger.debug("Sending packet: %s", raw_packet)
210 + sock.sendto(raw_packet, (args.host, args.port))
216 211
217 212
218 213 def parse_args() -> argparse.Namespace:
@@ -221,62 +216,89 @@ def parse_args() -> argparse.Namespace:
221 216 Returns:
222 217 argparse.Namespace: Parsed command-line arguments.
223 218 Command-line arguments:
224 - --log-level (str): Set the logging level. Choices are "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL". Default is "INFO".
225 - --config (str): Path to config file. Default is "config.toml".
226 - -i, --input-file (argparse.FileType): Input file to read from. Default is sys.stdin.
227 - text (str, optional): Text to send.
219 + --host, -H: VBAN host to send to (default: localhost)
220 + --port, -P: VBAN port to send to (default: 6980)
221 + --streamname, -s: VBAN stream name (default: Command1)
222 + --bps, -b: Bits per second for VBAN stream (default: 256000)
223 + --channel, -c: Channel number for VBAN stream (default: 1)
224 + text: Text to send (positional argument)
228 225 """
229 226
230 227 parser = argparse.ArgumentParser(description="Voicemeeter VBAN Send Text CLI")
231 228 parser.add_argument(
232 - "--log-level",
229 + "--host",
230 + "-H",
233 231 type=str,
234 - choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
235 - default="INFO",
236 - help="Set the logging level",
232 + default="localhost",
233 + help="VBAN host to send to (default: localhost)",
237 234 )
238 235 parser.add_argument(
239 - "--config", type=str, default="config.toml", help="Path to config file"
236 + "--port",
237 + "-P",
238 + type=int,
239 + default=6980,
240 + help="VBAN port to send to (default: 6980)",
240 241 )
241 242 parser.add_argument(
242 - "-i",
243 - "--input-file",
244 - type=argparse.FileType("r"),
245 - default=sys.stdin,
243 + "--streamname",
244 + "-s",
245 + type=str,
246 + default="Command1",
247 + help="VBAN stream name (default: Command1)",
248 + )
249 + parser.add_argument(
250 + "--bps",
251 + "-b",
252 + type=int,
253 + default=256000,
254 + help="Bits per second for VBAN stream (default: 256000)",
255 + )
256 + parser.add_argument(
257 + "--channel",
258 + "-c",
259 + type=int,
260 + default=1,
261 + help="Channel number for VBAN stream (default: 1)",
262 + )
263 + parser.add_argument(
264 + "--ratelimit",
265 + "-r",
266 + type=float,
267 + default=0.2,
268 + help="Minimum time in seconds between sending messages (default: 0.2)",
269 + )
270 + parser.add_argument(
271 + "--loglevel",
272 + "-l",
273 + type=str,
274 + default="info",
275 + choices=["debug", "info", "warning", "error", "critical"],
276 + help="Set the logging level (default: info)",
277 + )
278 + parser.add_argument(
279 + "cmds", nargs="+", type=str, help="Text to send (positional argument)"
246 280 )
247 - parser.add_argument("text", nargs="?", type=str, help="Text to send")
248 281 return parser.parse_args()
249 282
250 283
251 - def main(config: dict):
284 + def main(args: argparse.Namespace):
252 285 """
253 - Main function to send text using VbanSendText.
286 + Main function to send text using VBAN.
254 287 Args:
255 - config (dict): Configuration dictionary for VbanSendText.
288 + args (argparse.Namespace): Parsed command-line arguments.
256 289 Behavior:
257 - - If 'args.text' is provided, sends the text and returns.
258 - - Otherwise, reads lines from 'args.input_file', strips whitespace, and sends each line.
259 - - Stops reading and sending if a line equals "Q".
260 - - Logs each line being sent at the debug level.
290 + Creates a UDP socket and sends each command in 'args.cmds' to the specified VBAN host and port using the 'send' function, with rate limiting applied.
261 291 """
262 292
263 - with VbanSendText(**config) as vban:
264 - if args.text:
265 - vban.sendtext(args.text)
266 - return
267 -
268 - for line in args.input_file:
269 - line = line.strip()
270 - if line.upper() == "Q":
271 - break
272 -
273 - logger.debug(f"Sending {line}")
274 - vban.sendtext(line)
293 + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
294 + for n, cmd in enumerate(args.cmds):
295 + logger.debug(f"Sending: {cmd}")
296 + send(sock, args, cmd, n)
275 297
276 298
277 299 if __name__ == "__main__":
278 300 args = parse_args()
279 301
280 - logging.basicConfig(level=args.log_level)
302 + logging.basicConfig(level=getattr(logging, args.loglevel.upper()))
281 303
282 - main(conn_from_toml(args.config))
304 + main(args)

onyx_online's Avatar onyx-and-iris gist felülvizsgálása 1738749123. Revízióhoz ugrás

1 file changed, 4 insertions, 3 deletions

sendtext.py

@@ -164,14 +164,15 @@ class VbanSendText:
164 164 self.__dict__.update(defaultkwargs)
165 165 self._request = RequestPacket(RTHeader(self.streamname, self.bps, self.channel))
166 166 self.lastsent = 0
167 - self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
168 167
169 168 def __enter__(self):
170 - self._sock.__enter__()
169 + self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
171 170 return self
172 171
173 172 def __exit__(self, exc_type, exc_value, traceback):
174 - return self._sock.__exit__(exc_type, exc_value, traceback)
173 + if self._sock:
174 + self._sock.close()
175 + self._sock = None
175 176
176 177 @ratelimit
177 178 def sendtext(self, text: str):

onyx_online's Avatar onyx-and-iris gist felülvizsgálása 1738730392. Revízióhoz ugrás

1 file changed, 3 insertions, 2 deletions

sendtext.py

@@ -164,13 +164,14 @@ class VbanSendText:
164 164 self.__dict__.update(defaultkwargs)
165 165 self._request = RequestPacket(RTHeader(self.streamname, self.bps, self.channel))
166 166 self.lastsent = 0
167 + self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
167 168
168 169 def __enter__(self):
169 - self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
170 + self._sock.__enter__()
170 171 return self
171 172
172 173 def __exit__(self, exc_type, exc_value, traceback):
173 - self._sock.close()
174 + return self._sock.__exit__(exc_type, exc_value, traceback)
174 175
175 176 @ratelimit
176 177 def sendtext(self, text: str):

onyx_online's Avatar onyx-and-iris gist felülvizsgálása 1738730387. Revízióhoz ugrás

1 file changed, 58 insertions, 34 deletions

sendtext.py

@@ -22,6 +22,30 @@
22 22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23 23 # SOFTWARE.
24 24
25 + """
26 + sendtext.py
27 +
28 + This script provides functionality to send text messages over a network using the VBAN protocol.
29 + It includes classes and functions to construct and send VBAN packets with text data, read configuration
30 + from a TOML file, and handle command-line arguments for customization.
31 +
32 + Classes:
33 + RTHeader: Represents the VBAN header for a text stream.
34 + RequestPacket: Encapsulates a VBAN request packet with text data.
35 + VbanSendText: Manages the sending of text messages over VBAN with rate limiting.
36 +
37 + Functions:
38 + ratelimit: Decorator to enforce a rate limit on a function.
39 + conn_from_toml: Reads a TOML configuration file and returns its contents as a dictionary.
40 + parse_args: Parses command-line arguments.
41 + main: Main function to send text using VbanSendText.
42 +
43 + Usage:
44 + Run the script with appropriate command-line arguments to send text messages.
45 + Example:
46 + python sendtext.py --config /path/to/config.toml --log-level DEBUG "strip[0].mute=0 strip[1].mute=0"
47 + """
48 +
25 49 import argparse
26 50 import functools
27 51 import logging
@@ -43,7 +67,7 @@ class RTHeader:
43 67 bps: int
44 68 channel: int
45 69 VBAN_PROTOCOL_TXT = 0x40
46 - framecounter: bytes = (0).to_bytes(4, 'little')
70 + framecounter: bytes = (0).to_bytes(4, "little")
47 71 # fmt: off
48 72 BPS_OPTS: list[int] = field(default_factory=lambda: [
49 73 0, 110, 150, 300, 600, 1200, 2400, 4800, 9600, 14400, 19200, 31250,
@@ -60,22 +84,22 @@ class RTHeader:
60 84 try:
61 85 self.bps_index = self.BPS_OPTS.index(self.bps)
62 86 except ValueError as e:
63 - ERR_MSG = f'Invalid bps: {self.bps}, must be one of {self.BPS_OPTS}'
87 + ERR_MSG = f"Invalid bps: {self.bps}, must be one of {self.BPS_OPTS}"
64 88 e.add_note(ERR_MSG)
65 89 raise
66 90
67 91 def __sr(self) -> bytes:
68 - return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, 'little')
92 + return (RTHeader.VBAN_PROTOCOL_TXT + self.bps_index).to_bytes(1, "little")
69 93
70 94 def __nbc(self) -> bytes:
71 - return (self.channel).to_bytes(1, 'little')
95 + return (self.channel).to_bytes(1, "little")
72 96
73 97 def build(self) -> bytes:
74 - header = 'VBAN'.encode('utf-8')
98 + header = "VBAN".encode("utf-8")
75 99 header += self.__sr()
76 - header += (0).to_bytes(1, 'little')
100 + header += (0).to_bytes(1, "little")
77 101 header += self.__nbc()
78 - header += (0x10).to_bytes(1, 'little')
102 + header += (0x10).to_bytes(1, "little")
79 103 header += self.name.encode() + bytes(16 - len(self.name))
80 104 header += RTHeader.framecounter
81 105 return header
@@ -86,15 +110,15 @@ class RequestPacket:
86 110 self.header = header
87 111
88 112 def encode(self, text: str) -> bytes:
89 - return self.header.build() + text.encode('utf-8')
113 + return self.header.build() + text.encode("utf-8")
90 114
91 115 def bump_framecounter(self) -> None:
92 116 self.header.framecounter = (
93 - int.from_bytes(self.header.framecounter, 'little') + 1
94 - ).to_bytes(4, 'little')
117 + int.from_bytes(self.header.framecounter, "little") + 1
118 + ).to_bytes(4, "little")
95 119
96 120 logger.debug(
97 - f'framecounter: {int.from_bytes(self.header.framecounter, "little")}'
121 + f"framecounter: {int.from_bytes(self.header.framecounter, 'little')}"
98 122 )
99 123
100 124
@@ -129,12 +153,12 @@ def ratelimit(func: Callable) -> Callable:
129 153 class VbanSendText:
130 154 def __init__(self, **kwargs):
131 155 defaultkwargs = {
132 - 'host': 'localhost',
133 - 'port': 6980,
134 - 'streamname': 'Command1',
135 - 'bps': 256000,
136 - 'channel': 0,
137 - 'delay': 0.02,
156 + "host": "localhost",
157 + "port": 6980,
158 + "streamname": "Command1",
159 + "bps": 256000,
160 + "channel": 0,
161 + "delay": 0.02,
138 162 }
139 163 defaultkwargs.update(kwargs)
140 164 self.__dict__.update(defaultkwargs)
@@ -161,7 +185,7 @@ class VbanSendText:
161 185 self._request.bump_framecounter()
162 186
163 187
164 - def conn_from_toml(filepath: str = 'config.toml') -> dict:
188 + def conn_from_toml(filepath: str = "config.toml") -> dict:
165 189 """
166 190 Reads a TOML configuration file and returns its contents as a dictionary.
167 191 Args:
@@ -178,15 +202,15 @@ def conn_from_toml(filepath: str = 'config.toml') -> dict:
178 202 pn = Path(filepath)
179 203 if not pn.exists():
180 204 logger.info(
181 - f'no {pn} found, using defaults: localhost:6980 streamname: Command1'
205 + f"no {pn} found, using defaults: localhost:6980 streamname: Command1"
182 206 )
183 207 return {}
184 208
185 209 try:
186 - with open(pn, 'rb') as f:
210 + with open(pn, "rb") as f:
187 211 return tomllib.load(f)
188 212 except tomllib.TOMLDecodeError as e:
189 - raise ValueError(f'Error decoding TOML file: {e}') from e
213 + raise ValueError(f"Error decoding TOML file: {e}") from e
190 214
191 215
192 216 def parse_args() -> argparse.Namespace:
@@ -201,24 +225,24 @@ def parse_args() -> argparse.Namespace:
201 225 text (str, optional): Text to send.
202 226 """
203 227
204 - parser = argparse.ArgumentParser(description='Voicemeeter VBAN Send Text CLI')
228 + parser = argparse.ArgumentParser(description="Voicemeeter VBAN Send Text CLI")
205 229 parser.add_argument(
206 - '--log-level',
230 + "--log-level",
207 231 type=str,
208 - choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
209 - default='INFO',
210 - help='Set the logging level',
232 + choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
233 + default="INFO",
234 + help="Set the logging level",
211 235 )
212 236 parser.add_argument(
213 - '--config', type=str, default='config.toml', help='Path to config file'
237 + "--config", type=str, default="config.toml", help="Path to config file"
214 238 )
215 239 parser.add_argument(
216 - '-i',
217 - '--input-file',
218 - type=argparse.FileType('r'),
240 + "-i",
241 + "--input-file",
242 + type=argparse.FileType("r"),
219 243 default=sys.stdin,
220 244 )
221 - parser.add_argument('text', nargs='?', type=str, help='Text to send')
245 + parser.add_argument("text", nargs="?", type=str, help="Text to send")
222 246 return parser.parse_args()
223 247
224 248
@@ -241,14 +265,14 @@ def main(config: dict):
241 265
242 266 for line in args.input_file:
243 267 line = line.strip()
244 - if line.upper() == 'Q':
268 + if line.upper() == "Q":
245 269 break
246 270
247 - logger.debug(f'Sending {line}')
271 + logger.debug(f"Sending {line}")
248 272 vban.sendtext(line)
249 273
250 274
251 - if __name__ == '__main__':
275 + if __name__ == "__main__":
252 276 args = parse_args()
253 277
254 278 logging.basicConfig(level=args.log_level)
Újabb Régebbi

Ez az oldal Opengist alapú ⋅ Load: 151ms⋅

Magyar