Like 0
Fork 0

Last active 1738749288

Use this script to send vban text requests over a network

onyx_online's Avatar onyx-and-iris revised this gist 1738749123. Go to revision

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 revised this gist 1738730392. Go to revision

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 revised this gist 1738730387. Go to revision

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)

onyx_online's Avatar onyx_online revised this gist 1737741073. Go to revision

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's Avatar onyx_online revised this gist 1737739968. Go to revision

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's Avatar onyx_online revised this gist 1737652391. Go to revision

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's Avatar onyx_online revised this gist 1737603186. Go to revision

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's Avatar onyx_online revised this gist 1737597025. Go to revision

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's Avatar onyx_online revised this gist 1737596796. Go to revision

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's Avatar onyx_online revised this gist 1737596684. Go to revision

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:
Newer Older

Powered by Opengist ⋅ Load: 42ms⋅

English