The initial use case for this is to use endless rotary knobs on my Nocturn
as a volume control for VLC (using VLC's telnet interface). The issue is that I don't want to create a new
telnet session for every click of the knob. On the other hand, there is the issue of what happens if I close
VLC and reopen it. This is my first stab. (Note, I'm not an expert in network programming, so I'm learning
as I go.) I've not decided on whether to drop or buffer commands when the connection is closed. For now,
if the socket is closed, then the send method will try to reopen. If that fails, the command is dropped.
import socket
from time import sleep
from icecream import ic; ic.configureOutput(includeContext=True)
# Raise an exception if called too many times
# simple pattern to limit iterations
# where the iteration limit can be shared by various (e.g. nested) loops
class Doom:
def __init__(self,m):
self.m = m
self.n = 0
def __call__(self):
self.n += 1
if self.n >= self.m:
raise DoomException()
class DoomException(Exception):
pass
# telnet client for vlc
# purpose is to reuse an existing connection rather than
# creating a new one for each command
# then we can efficiently turn OSC into VLC telnet commands
# with auto-reconnect
# I don't know exactly what exceptions get thrown when,
# though I have determined that BrokenPipeError happens
# if you quit VLC while the connection is open.
class B:
def __init__(self):
self.host = "127.0.0.1"
self.port = 1234
self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.s.settimeout(0.5)
self.up = False
def close(self):
self.s.close()
def connect(self):
s = self.s
try:
s.connect((self.host,self.port))
except Exception as e:
ic("connect",e)
doom = Doom(100)
while True:
doom()
a = s.recv(1024)
if b"Password:" in a:
s.sendall(b"password123\n")
sleep(0.1)
b = s.recv(1024)
if b"aster" in b:
self.up = True
return
def send(self,cmds):
if self.up is False:
try:
self.connect()
except Exception as e:
ic("connect in send",e)
return b""
try:
for cmd in cmds:
cmd = cmd.rstrip()+"\n"
self.s.sendall(cmd.encode())
sleep(0.1)
return self.recv(1024)
except BrokenPipeError as e:
ic("broken pipe in sendall in send",e)
self.up = False
return b""
except Exception as e:
ic("sendall in send",e)
return b""
def send1(self,cmd):
return self.send([cmd])
def recv(self,n):
a = self.s.recv(n)
print("recv",len(a))
try:
b = a.decode()
except Exception:
b = "XX"+"".join( x for x in a if ord(x) < 128 and ord(x) >= 32 )
return b
def __del__(self):
print("del B")
self.s.close()
def main():
# simple test
b = B()
b.connect()
try:
for i in range(100):
b.send1("pause")
sleep(1)
b.send1("pause")
sleep(1)
except Exception as e:
ic(e)
del(b)
if __name__ == "__main__":
main()