maikandClaude Opus 4.8
a7c93fa266
Läuft die Voice Acoustic VA-Remotecontrol (AllDSP AllControl) nativ unter Linux auf echtem wxGTK 3.0 — ohne Wine und ohne dekompilierten App-Code. Ausgeführt wird das unveränderte Python-2.7-Bytecode aus dem macOS-.app; C-Extensions (wxGTK/numpy/ scipy/PortAudio) kommen nativ aus Debian stretch. Enthält: va/ (Original-.pyc + pure-Python-Libs + Assets + launch.py mit den drei Linux-Fixes), Dockerfile/run.sh/entrypoint.sh, reference/ (dekompilierter Quellcode zum Debuggen) und README/STATUS. img.cache (378 MB, regenerierbar) ist per .gitignore ausgeschlossen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
420 lines
16 KiB
Python
420 lines
16 KiB
Python
# uncompyle6 version 3.9.3
|
|
# Python bytecode version base 2.6 (62161)
|
|
# Decompiled from: Python 3.9.25 (main, Oct 31 2025, 23:16:49)
|
|
# [GCC 14.2.0]
|
|
# Embedded file name: d_usb.pyc
|
|
# Compiled at: 2023-05-16 13:01:50
|
|
import myserial, mytime, threading, copy, struct, types, d_protocol, os.path, traceback, glob, sys, main, one_unit, string
|
|
from subprocess import Popen, PIPE
|
|
import subprocess, os, platform
|
|
mac_names = 'posix'
|
|
|
|
def ListUSBDevices(find='find'):
|
|
import win32com.client, pythoncom
|
|
try:
|
|
pythoncom.CoInitialize()
|
|
except:
|
|
pass
|
|
|
|
devices = {}
|
|
try:
|
|
strComputer = '.'
|
|
objWMIService = win32com.client.Dispatch('WbemScripting.SWbemLocator')
|
|
objSWbemServices = objWMIService.ConnectServer(strComputer, 'root\\cimv2')
|
|
colItems = objSWbemServices.ExecQuery('Select * from Win32_USBControllerDevice')
|
|
for objItem in colItems:
|
|
DeviceID = objItem.Dependent[objItem.Dependent.find('=') + 2:-1]
|
|
if DeviceID.find('VID_0684') < 0:
|
|
continue
|
|
searchstring = "Select * from Win32_PnPEntity Where DeviceID = '" + DeviceID + "'"
|
|
colUSBDevice = objSWbemServices.ExecQuery(searchstring)
|
|
for objUSBDevice in colUSBDevice:
|
|
if False:
|
|
continue
|
|
try:
|
|
devices[int(objUSBDevice.Name[objUSBDevice.Name.find('COM') + 3:objUSBDevice.Name.rfind(')')])] = os.path.normpath(DeviceID)
|
|
except:
|
|
print mytime.displayTime() + ' Failed to connect USB device', objUSBDevice.Name, 'ID:', DeviceID
|
|
|
|
except:
|
|
traceback.print_exc(file=sys.stdout)
|
|
|
|
try:
|
|
pythoncom.CoUninitialize()
|
|
except:
|
|
pass
|
|
|
|
return devices
|
|
|
|
|
|
def RestartDevice(device):
|
|
if os.name in mac_names or platform.version()[0] == '6':
|
|
return
|
|
print mytime.displayTime() + ' Restarting', device
|
|
try:
|
|
startupinfo = subprocess.STARTUPINFO()
|
|
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
|
args = ['srvdz_rt.exe', 'restart', '@' + device]
|
|
Popen(args, stdout=PIPE, startupinfo=startupinfo)
|
|
except:
|
|
print mytime.displayTime() + ' Restart Device', device, 'Failed'
|
|
|
|
return
|
|
|
|
|
|
class USB_updater:
|
|
|
|
def __init__(self, parent, ports=range(100)):
|
|
self.ports = ports
|
|
self.Connected = False
|
|
self.ResetValues = False
|
|
self.last_peq_key = None
|
|
self.last_xover_key = None
|
|
self.values = {}
|
|
self.last_requested = -1
|
|
self.parent = parent
|
|
self.lock = threading.Lock()
|
|
self.stopping = False
|
|
self.broadcasted_skin = ''
|
|
self.last_hw_packet_id = 0
|
|
self.invalid_packet_counter = 0
|
|
self.DisconnectTime = mytime.clock()
|
|
self.thread = threading.Thread(name='AllDSP Server USB Detection', target=self.process)
|
|
self.thread.daemon = True
|
|
self.active_port = None
|
|
self.active_usb_devices = {}
|
|
self.all_usb_devices = {}
|
|
self.loops = 0
|
|
return
|
|
|
|
def start(self):
|
|
"""Start the updating thread"""
|
|
self.stopping = False
|
|
self.thread.start()
|
|
return
|
|
|
|
def stop(self):
|
|
"""Stop the updating thread"""
|
|
self.stopping = True
|
|
self.thread.join()
|
|
return
|
|
|
|
def process(self):
|
|
one_unit.lowPriority()
|
|
self.last_list_time = 0
|
|
if os.name in mac_names:
|
|
while not self.stopping and one_unit.app.application_stopping == False:
|
|
ser = None
|
|
self.active_port = None
|
|
while mytime.clock() - self.last_list_time < 3:
|
|
mytime.sleep(0.5)
|
|
|
|
self.last_list_time = mytime.clock()
|
|
try:
|
|
mac_ports = glob.glob('/dev/tty.usb*')
|
|
except:
|
|
mac_ports = []
|
|
|
|
for port in mac_ports:
|
|
try:
|
|
self.active_port = port
|
|
s = port
|
|
IP = 'USB port ' + ('').join([_[1] for letter in s if letter.isdigit()])
|
|
ser = myserial.Serial(port, 115200, timeout=1)
|
|
pkt = read_packet(ser)
|
|
if isinstance(pkt, Announcement):
|
|
unit = d_protocol.RemoteUnit()
|
|
try:
|
|
new_mac = ('0000000005A' + unicode(hex(int(pkt.broadcasted_skin.replace('_', '') + unicode(pkt.Serial)))).upper()[2:])[-12:]
|
|
except:
|
|
new_mac = '0000005A0000'
|
|
traceback.print_exc(file=sys.stdout)
|
|
else:
|
|
f_mac = ''
|
|
for i in range(5):
|
|
f_mac = f_mac + new_mac[i * 2:i * 2 + 2] + ':'
|
|
|
|
f_mac += new_mac[10:12]
|
|
unit.MAC = f_mac
|
|
unit.IP = IP
|
|
unit.type = pkt.broadcasted_skin
|
|
unit.name = 'No Name'
|
|
unit.pkt = pkt
|
|
unit.link_status = 'ready'
|
|
unit.link_type = 'USB'
|
|
unit.HWVersion = pkt.HWVersion
|
|
unit.PICVersion = pkt.PICVersion
|
|
unit.Serial = pkt.Serial
|
|
unit.ProductionDate = pkt.ProductionDate
|
|
unit.originalSwID = pkt.originalSwID
|
|
unit.InstanceId = port
|
|
self.parent.peers.add(unit)
|
|
pkt = None
|
|
ser.close()
|
|
ser = None
|
|
except:
|
|
try:
|
|
self.lock.release()
|
|
except:
|
|
pass
|
|
else:
|
|
if ser and ser != None:
|
|
try:
|
|
ser.close()
|
|
except:
|
|
pass
|
|
else:
|
|
try:
|
|
ser = None
|
|
except:
|
|
pass
|
|
|
|
else:
|
|
try:
|
|
mytime.sleep(0.01)
|
|
except:
|
|
pass
|
|
else:
|
|
try:
|
|
ser.close()
|
|
except:
|
|
pass
|
|
|
|
self.active_port = None
|
|
try:
|
|
mytime.sleep(1)
|
|
except:
|
|
pass
|
|
|
|
else:
|
|
port_index = 0
|
|
ser = None
|
|
while not self.stopping and one_unit.app.application_stopping == False:
|
|
while mytime.clock() - self.last_list_time < 3:
|
|
mytime.sleep(0.5)
|
|
|
|
self.last_list_time = mytime.clock()
|
|
self.active_usb_devices = ListUSBDevices()
|
|
if self.loops == 3:
|
|
self.all_usb_devices = ListUSBDevices('findall')
|
|
for device in self.all_usb_devices.values():
|
|
for unit in self.parent.peers.units.values():
|
|
if unit.InstanceId == device and unit.link_status != 'disconnected':
|
|
print device, 'already connected, skipping'
|
|
continue
|
|
else:
|
|
try:
|
|
RestartDevice(device)
|
|
except:
|
|
pass
|
|
|
|
self.loops += 1
|
|
if self.active_usb_devices == None or len(self.active_usb_devices) == 0:
|
|
try:
|
|
mytime.sleep(1)
|
|
continue
|
|
except:
|
|
return
|
|
|
|
for port in self.active_usb_devices.keys():
|
|
try:
|
|
self.active_port = int(port) - 1
|
|
IP = ('').join(('USB port ', str(self.active_port + 1)))
|
|
if ser is None:
|
|
ser = myserial.Serial(self.active_port, 115200, timeout=1)
|
|
start_time = mytime.clock()
|
|
while mytime.clock() - start_time < 1:
|
|
waiting = ser.inWaiting()
|
|
if waiting >= 42:
|
|
break
|
|
|
|
print mytime.displayTime() + ' opened serial port', ser.port + 1
|
|
pkt = read_packet(ser)
|
|
ser.close()
|
|
self.active_port = None
|
|
ser = None
|
|
if isinstance(pkt, Announcement):
|
|
print mytime.displayTime(), 'Got USB announcement from', IP
|
|
unit = d_protocol.RemoteUnit()
|
|
try:
|
|
new_mac = ('0000000005A' + unicode(hex(int(pkt.broadcasted_skin.replace('_', '') + unicode(pkt.Serial)))).upper()[2:])[-12:]
|
|
except:
|
|
new_mac = '0000005A0000'
|
|
traceback.print_exc(file=sys.stdout)
|
|
else:
|
|
f_mac = ''
|
|
for i in range(5):
|
|
f_mac = f_mac + new_mac[i * 2:i * 2 + 2] + ':'
|
|
|
|
f_mac += new_mac[10:12]
|
|
unit.MAC = f_mac
|
|
unit.IP = IP
|
|
unit.type = pkt.broadcasted_skin
|
|
unit.name = 'No Name'
|
|
unit.pkt = pkt
|
|
unit.link_status = 'ready'
|
|
unit.link_type = 'USB'
|
|
unit.HWVersion = pkt.HWVersion
|
|
unit.PICVersion = pkt.PICVersion
|
|
unit.Serial = pkt.Serial
|
|
unit.ProductionDate = pkt.ProductionDate
|
|
unit.originalSwID = pkt.originalSwID
|
|
unit.InstanceId = self.active_usb_devices[port]
|
|
self.parent.peers.add(unit)
|
|
else:
|
|
print mytime.displayTime() + ' No announcement'
|
|
except:
|
|
try:
|
|
self.lock.release()
|
|
except:
|
|
pass
|
|
else:
|
|
if ser and ser != None:
|
|
try:
|
|
ser.close()
|
|
except:
|
|
pass
|
|
else:
|
|
try:
|
|
ser = None
|
|
except:
|
|
pass
|
|
|
|
else:
|
|
try:
|
|
mytime.sleep(0.01)
|
|
except:
|
|
pass
|
|
else:
|
|
try:
|
|
ser.close()
|
|
except:
|
|
pass
|
|
|
|
self.active_port = None
|
|
ser = None
|
|
try:
|
|
mytime.sleep(1)
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
|
|
class Announcement:
|
|
"""The announcement broadcast by hardware units over USB"""
|
|
|
|
def __init__(self):
|
|
"""Constructor"""
|
|
self.UnitType2 = None
|
|
self.UnitType1 = None
|
|
self.MAC = None
|
|
self.PC_PacketId = None
|
|
self.HW_PacketId = None
|
|
self.skin_high = None
|
|
self.skin_mid = None
|
|
self.skin_low = None
|
|
self.broadcasted_skin = None
|
|
self.HWVersion = None
|
|
self.PICVersion = None
|
|
self.Serial = None
|
|
self.ProductionDate = None
|
|
self.originalSwID = ''
|
|
self.MK2 = False
|
|
return
|
|
|
|
def __str__(self):
|
|
return 'Announcement(%s, Type1=%s, Type2=%s)' % (self.MAC,
|
|
self.UnitType1, self.UnitType2)
|
|
|
|
|
|
def parse_announce(data):
|
|
"""Parse the data sent by MakeConfigData(void) in DDEthernet.c,
|
|
line 537. This data can be found in UDP broadcasts and in TCP
|
|
responces to the 'J' command from the user.
|
|
"""
|
|
if len(data) < 42 or data[41] != 'Q' or data[:3] != 'DDB':
|
|
print mytime.displayTime() + " unexpected data: len=%d, last='%s'" % (len(data), data[41])
|
|
return None
|
|
else:
|
|
res = Announcement()
|
|
res.UnitType2 = ord(data[4])
|
|
res.UnitType1 = ord(data[5])
|
|
res.MAC = data[5:22]
|
|
res.PC_PacketID = ord(data[39])
|
|
res.HW_PacketID = ord(data[40])
|
|
res.skin_high = unicode(ord(data[33]))
|
|
res.skin_mid = unicode(ord(data[34]))
|
|
res.skin_low = unicode(ord(data[35]))
|
|
res.broadcasted_skin = res.skin_high + '_' + res.skin_mid + '_' + res.skin_low
|
|
skinBeforeReplacement = ':' + res.skin_high + '.' + res.skin_mid + '.' + res.skin_low
|
|
try:
|
|
res.broadcasted_skin = one_unit.app.type_replacements[res.broadcasted_skin]
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
high = unicode(ord(data[22]))
|
|
mid = unicode(ord(data[23]))
|
|
low = unicode(ord(data[24]))
|
|
res.HWVersion = high + '.' + mid + '.' + low
|
|
high = unicode(ord(data[25]))
|
|
mid = unicode(ord(data[26]))
|
|
low = unicode(ord(data[27]))
|
|
res.PICVersion = high + '.' + mid + '.' + low
|
|
high = int(ord(data[28]))
|
|
mid = int(ord(data[29]))
|
|
low = int(ord(data[30]))
|
|
res.Serial = high * 65536 + mid * 256 + low
|
|
high = unicode(2000 + ord(data[31]))
|
|
mid = unicode(ord(data[32]))
|
|
low = unicode(ord(data[36]))
|
|
res.ProductionDate = low + '-' + mid + '-' + high
|
|
except:
|
|
pass
|
|
|
|
res.originalSwID = res.skin_high + '.' + res.skin_mid + '.' + res.skin_low + skinBeforeReplacement
|
|
return res
|
|
|
|
|
|
def read_packet(ser):
|
|
buf = ''
|
|
while True:
|
|
try:
|
|
chunk = ser.read(42)
|
|
except:
|
|
return
|
|
else:
|
|
if len(chunk) == 0:
|
|
return
|
|
buf += chunk
|
|
pos = buf.find('DD')
|
|
if pos < 0:
|
|
buf = ''
|
|
return
|
|
buf = buf[pos:]
|
|
if len(buf) < 42:
|
|
buf = ''
|
|
return
|
|
if buf[41] != 'Q':
|
|
buf = ''
|
|
return
|
|
while len(buf) < 42:
|
|
chunk = ser.read(42 - len(buf))
|
|
buf += chunk
|
|
|
|
pkt = None
|
|
if buf[:3] == 'DDB':
|
|
pkt = parse_announce(buf)
|
|
else:
|
|
return
|
|
|
|
buf = buf[42:]
|
|
return pkt
|
|
|
|
return
|
|
|
|
|
|
|
|
# okay decompiling pycode/d_usb.pyc
|