uli/intercom
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

feat: configurable name. Readme revamped. Code formatted.

Browse Source
This commit is contained in:
Eneko Nieto
2026-04-14 12:48:41 +02:00
parent 066535f5c1
commit 61017757fa
7 changed files with 126 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
call_manager.py
View File

@@ -4,22 +4,30 @@ import socket
NO_CALL = 0
INCOMING_CALL = 1
OUTGOING_CALL = 2
OP_CALL = 'CALL'.encode()
OP_HANG = 'HANG'.encode()
OP_PING = 'PING'.encode()
OP_OK = 'OK'.encode()
OP_ERROR = 'ERROR'.encode()
OP_CALL = "CALL".encode()
OP_HANG = "HANG".encode()
OP_PING = "PING".encode()
OP_OK = "OK".encode()
OP_ERROR = "ERROR".encode()
def exec_mumble(deaf: bool):
subprocess.Popen(['mumble', 'rpc', 'deaf' if deaf else 'undeaf'])
subprocess.Popen(['mumble', 'rpc', 'mute' if deaf else 'unmute'])
subprocess.run(
["xvfb-run", "-n", "1", "mumble", "rpc", "deaf" if deaf else "undeaf"]
)
subprocess.run(
["xvfb-run", "-n", "2", "mumble", "rpc", "mute" if deaf else "unmute"]
)
def deaf_mumble():
exec_mumble(True)
def undeaf_mumble():
exec_mumble(False)
class CallManager:
def __init__(self, config):
self.config = config
@@ -29,31 +37,31 @@ class CallManager:
def call(self, station):
try:
s = socket.socket()
print('host=', station['host'])
print('port=', station['port'])
s.connect((station['host'], station['port']))
print("host=", station["host"])
print("port=", station["port"])
s.connect((station["host"], station["port"]))
s.send(OP_CALL)
response = s.recv(1024)
print('Call notice sent', 'response=', response.decode())
print("Call notice sent", "response=", response.decode())
s.close()
undeaf_mumble()
self.state = OUTGOING_CALL
self.station = station
station['red'].on()
station["red"].on()
except Exception as e:
print('Error al realizar la llamada: ', e)
station['green'].off()
print("Error al realizar la llamada: ", e)
station["green"].off()
def incoming_call(self, station):
undeaf_mumble()
self.state = INCOMING_CALL
self.station = station
station['red'].on()
station["red"].on()
def hang(self):
print('Hang! self.station', self.station)
print("Hang! self.station", self.station)
if self.station is not None:
self.station['red'].off()
self.station["red"].off()
deaf_mumble()
self.state = NO_CALL
self.station = None
@@ -61,8 +69,8 @@ class CallManager:
def notify_hang(self):
if self.station is not None:
s = socket.socket()
s.connect((self.station['host'], self.station['port']))
s.connect((self.station["host"], self.station["port"]))
s.send(OP_HANG)
response = s.recv(1024)
print('Hang notify sent', 'response=', response.decode())
print("Hang notify sent", "response=", response.decode())
s.close()