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>
589 lines
22 KiB
Python
589 lines
22 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: data_model.pyc
|
|
# Compiled at: 2023-11-20 12:35:49
|
|
import collections, protocol
|
|
Key = collections.namedtuple('Key', 'struct_id, member_id, channel, num')
|
|
import traceback, mytime, sys, one_unit, math, string, inspect
|
|
GET_links = {(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_CHECK_PASSWORD, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_CURRENT_USER, 0, 0), 0.5), (Key(protocol.STRUCT_ID_COMMAND, protocol.MEMBER_ID_COMMAND, 0, protocol.cmdGotoStandby)): (
|
|
Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_HARDWARE_STATUS_FLAGS, 0, 0), 0.5),
|
|
(Key(protocol.STRUCT_ID_COMMAND, protocol.MEMBER_ID_COMMAND, 0, protocol.cmdExitStandby)): (
|
|
Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_HARDWARE_STATUS_FLAGS, 0, 0), 0.5)}
|
|
GET_replace_list = {(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_PIN, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_ACCESS_RIGHTS, 0, 0)),
|
|
(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_BUILD_NUMBER, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_SW_VERSION, 0, 0)),
|
|
(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_MODEL_NUMBER, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_ACCESS_RIGHTS, 0, 0)),
|
|
(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_MODEL_FIRMWARE, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_ACCESS_RIGHTS, 0, 0)),
|
|
(Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_MODEL_NAME, 0, 0)): (Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_SW_VERSION, 0, 0))}
|
|
validFilenameChars = unicode('#$%&*+!?-_.,^@<>={}[]|() ') + string.ascii_letters + string.digits
|
|
forbiddenFilenameChars = unicode('\'"<>*[]?:;,|=')
|
|
|
|
def getValidName(string):
|
|
res = ''
|
|
for c in string:
|
|
byte = ord(c)
|
|
if byte < 128:
|
|
res += c
|
|
elif byte == 246:
|
|
res += 'oe'
|
|
continue
|
|
if byte == 228:
|
|
res += 'ae'
|
|
continue
|
|
if byte == 252:
|
|
res += 'ue'
|
|
continue
|
|
if byte == 223:
|
|
res += 'ss'
|
|
continue
|
|
if byte == 214:
|
|
res += 'Oe'
|
|
continue
|
|
if byte == 196:
|
|
res += 'Ae'
|
|
continue
|
|
if byte == 220:
|
|
res += 'Ue'
|
|
continue
|
|
print mytime.displayTime(), 'unknown char:', byte
|
|
res += '?'
|
|
|
|
res = ('').join(c for c in res if c in validFilenameChars)
|
|
res = res.strip()
|
|
if len(res) == 0:
|
|
res = '????'
|
|
return res
|
|
|
|
|
|
def safeName(val):
|
|
new_val = ('').join(c for c in val if c in validFilenameChars)
|
|
return str(new_val)
|
|
|
|
|
|
class DataModel:
|
|
|
|
def __init__(self, parent):
|
|
self.parent = parent
|
|
self.vars = {}
|
|
self.get_list = []
|
|
self.ignored = []
|
|
self.current_channel = -1
|
|
self.preset_names = {}
|
|
self.changed = False
|
|
self.unit_names = {}
|
|
self.min_build_number = {}
|
|
self.var_names = {}
|
|
self.min_values = {}
|
|
self.max_values = {}
|
|
self.last_forced_update = mytime.clock()
|
|
self.lock = self.parent.lock
|
|
self.lastPrintCompletionTime = 0
|
|
return
|
|
|
|
def reset_get_list(self):
|
|
try:
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.get_list = self.get_list[:]
|
|
print mytime.displayTime(), 'dm.rgl.Reset get list to', len(self.parent.server.peers.units[self.parent.MAC].process.conn.get_list)
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
def fastAdd(self, key, verify=True):
|
|
try:
|
|
if verify == False or self.verify_add_var(key) == True:
|
|
if key not in one_unit.app.member_scale:
|
|
if key.struct_id == protocol.STRUCT_ID_PRESET_GLOBAL and key.member_id == protocol.MEMBER_ID_DELAY:
|
|
one_unit.app.member_scale[key] = 1
|
|
one_unit.app.member_offset[key] = 0.0
|
|
else:
|
|
one_unit.app.member_scale[key] = protocol.get_member_scale(key.member_id)
|
|
one_unit.app.member_offset[key] = 0.0
|
|
if key not in self.get_list:
|
|
self.get_list.append(key)
|
|
val = self.parent.server.peers.get(key, self.parent.MAC)
|
|
if val == None:
|
|
self.vars[key] = None
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.get_list.append(key)
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.config_vars.append(key)
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
def get(self, key, default=None, fast=False):
|
|
if key is None or key.channel is None or key.channel < 0:
|
|
return
|
|
else:
|
|
try:
|
|
val = self.parent.server.peers.get(key, self.parent.MAC)
|
|
if fast:
|
|
return val
|
|
if key not in one_unit.app.member_scale:
|
|
one_unit.app.member_scale[key] = protocol.get_member_scale(key.member_id)
|
|
one_unit.app.member_offset[key] = 0.0
|
|
try:
|
|
val = val - one_unit.app.member_offset[key]
|
|
val = val / one_unit.app.member_scale[key]
|
|
except:
|
|
pass
|
|
|
|
if val == None:
|
|
try:
|
|
if self.parent.MAC.startswith('GRP_'):
|
|
group = self.parent.MAC[4:]
|
|
mac = one_unit.app.groups[group].keys()[0]
|
|
val = self.parent.server.peers.get(key, mac)
|
|
try:
|
|
val = val - one_unit.app.member_offset[key]
|
|
val = val / one_unit.app.member_scale[key]
|
|
except:
|
|
pass
|
|
|
|
except:
|
|
pass
|
|
else:
|
|
if val == None:
|
|
val = self.vars.get(key, default)
|
|
if self.verify_add_var(key) == True:
|
|
if key not in self.get_list:
|
|
self.get_list.append(key)
|
|
if val == None:
|
|
try:
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.get_list.append(key)
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.config_vars.append(key)
|
|
self.vars[key] = None
|
|
except:
|
|
pass
|
|
|
|
return val
|
|
except:
|
|
traceback.print_exc(file=sys.stdout)
|
|
|
|
return val
|
|
|
|
def verify_add_var(self, key):
|
|
try:
|
|
dummy = int(key.struct_id)
|
|
except:
|
|
return False
|
|
else:
|
|
try:
|
|
dummy = int(key.member_id)
|
|
except:
|
|
return False
|
|
else:
|
|
try:
|
|
dummy = int(key.channel)
|
|
except:
|
|
return False
|
|
|
|
try:
|
|
dummy = int(key.num)
|
|
except:
|
|
return False
|
|
|
|
try:
|
|
if self.parent.MAC.startswith('DEMO') == True:
|
|
return False
|
|
if self.parent.MAC.startswith('GRP') == True:
|
|
return False
|
|
if key in self.parent.remote_link.NO_GET_LIST:
|
|
return False
|
|
if key.struct_id == protocol.STRUCT_ID_COMMAND:
|
|
return False
|
|
if key.struct_id < 0 or key.member_id in (protocol.MEMBER_ID_VU_IN, protocol.MEMBER_ID_VU_OUT, protocol.MEMBER_ID_GR_IN, protocol.MEMBER_ID_GR_OUT):
|
|
return False
|
|
if key.struct_id == 255:
|
|
return False
|
|
if key.struct_id == protocol.STRUCT_ID_PRESET_GLOBAL:
|
|
if key.member_id in (protocol.MEMBER_ID_USER_ACCESS_RIGHTS, protocol.MEMBER_ID_SET_PASSWORD) and self.parent.server.peers.units[self.parent.MAC].numericalFWVersion < self.parent.server.peers.units[self.parent.MAC].minPresetLockVersion:
|
|
return False
|
|
try:
|
|
if key.member_id == protocol.MEMBER_ID_SHORT_NAME and key.channel > self.parent.server.peers.units[self.parent.MAC].process.conn.maxNumberOfPresets:
|
|
return False
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
if key in self.parent.server.peers.units[self.parent.MAC].process.conn.get_list:
|
|
return False
|
|
except:
|
|
return True
|
|
|
|
if key in self.min_build_number and (self.min_build_number[key] > int(self.parent.server.peers.units[self.parent.MAC].BuildNumber) or self.min_build_number[key] > 0 and self.parent.server.peers.units[self.parent.MAC].numericalFWVersion < self.parent.server.peers.units[self.parent.MAC].minBuildNumberVersion):
|
|
return False
|
|
except:
|
|
traceback.print_exc(file=sys.stdout)
|
|
self.ignored.append(key)
|
|
print mytime.displayTime() + ' dm.vav.invalid key 12:', key
|
|
return False
|
|
|
|
if key.struct_id == protocol.STRUCT_ID_FIR and key.member_id == protocol.MEMBER_ID_GAIN and self.parent.server.peers.units[self.parent.MAC].numericalFWVersion < self.parent.server.peers.units[self.parent.MAC].minFIRgainVersion:
|
|
return False
|
|
return True
|
|
|
|
def get_by_name(self, name, default=None):
|
|
if name is None:
|
|
return
|
|
else:
|
|
key = None
|
|
for key in self.var_names.keys():
|
|
if self.var_names[key] == name:
|
|
break
|
|
|
|
if key == None:
|
|
return
|
|
if key not in one_unit.app.member_scale:
|
|
one_unit.app.member_scale[key] = protocol.get_member_scale(key.member_id)
|
|
one_unit.app.member_offset[key] = 0.0
|
|
try:
|
|
val = self.parent.server.peers.get(key, self.parent.MAC)
|
|
try:
|
|
val = val - one_unit.app.member_offset[key]
|
|
val = val / one_unit.app.member_scale[key]
|
|
except:
|
|
pass
|
|
|
|
if val == None:
|
|
val = self.vars.get(key, default)
|
|
return val
|
|
except:
|
|
pass
|
|
|
|
return self.vars.get(key, default)
|
|
|
|
def get_scale_by_name(self, name, default=1.0):
|
|
if name is None:
|
|
return 1.0
|
|
else:
|
|
key = None
|
|
for key in self.var_names.keys():
|
|
if self.var_names[key] == name:
|
|
break
|
|
|
|
if key == None:
|
|
return 1.0
|
|
if key not in one_unit.app.member_scale:
|
|
one_unit.app.member_scale[key] = protocol.get_member_scale(key.member_id)
|
|
one_unit.app.member_offset[key] = 0.0
|
|
return one_unit.app.member_scale[key]
|
|
|
|
def get_internal(self, key, default=None):
|
|
if key is None or key.channel is None or key.channel < 0:
|
|
return default
|
|
else:
|
|
return self.vars.get(key, default)
|
|
|
|
def get_unit_name(self, number):
|
|
try:
|
|
return self.unit_names[number]
|
|
except:
|
|
return number
|
|
|
|
return
|
|
|
|
def set_unit_name(self, name, number):
|
|
self.unit_names[number] = name
|
|
return
|
|
|
|
def set_to_unit(self, key, val):
|
|
if key is None or key.channel is None or key.channel < 0:
|
|
print mytime.displayTime() + ' data_model.invalid key!!', key
|
|
return
|
|
else:
|
|
self.vars[key] = val
|
|
if key.struct_id == protocol.STRUCT_ID_INTERNAL:
|
|
return
|
|
self.parent.remote_link.set_by_key(key, val)
|
|
self.parent.remote_link.set_last_key(key)
|
|
if self.parent.MAC[:4] in ('DEMO', 'GRPdeactivated'):
|
|
self.vars[key] = val
|
|
return
|
|
for testkey in GET_links.keys():
|
|
if testkey == key:
|
|
get_key = GET_links[key][0]
|
|
delaytime = GET_links[key][1]
|
|
mytime.sleep(delaytime)
|
|
self.parent.remote_link.set_get_key(get_key)
|
|
|
|
return
|
|
|
|
def update(self):
|
|
print mytime.displayTime(), 'dm.u'
|
|
if mytime.clock() - self.last_forced_update < 0.25:
|
|
return
|
|
self.last_forced_update = mytime.clock()
|
|
try:
|
|
self.parent.current_view.timer.Stop()
|
|
self.parent.current_view.timer.Start(500)
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
def set_direct(self, key, val):
|
|
if key.struct_id < 0:
|
|
try:
|
|
self.parent.current_view.update_gauge(key, val)
|
|
except:
|
|
pass
|
|
|
|
return
|
|
if key not in one_unit.app.member_scale:
|
|
one_unit.app.member_scale[key] = protocol.get_member_scale(key.member_id)
|
|
one_unit.app.member_offset[key] = 0.0
|
|
try:
|
|
val = val - one_unit.app.member_offset[key]
|
|
val = val / one_unit.app.member_scale[key]
|
|
except:
|
|
pass
|
|
|
|
self.set_from_unit(key, val)
|
|
return
|
|
|
|
def set_from_unit(self, key, value):
|
|
if key is None or key.channel is None or key.channel < 0:
|
|
print mytime.displayTime() + ' data_model.invalid key!!', key
|
|
return
|
|
else:
|
|
if key.member_id == protocol.MEMBER_ID_CURRENT_USER:
|
|
if self.parent.current_view.select_skin_to_current_user() == True:
|
|
return
|
|
if key in self.vars.keys():
|
|
self.lock.acquire()
|
|
if value != self.vars[key]:
|
|
self.vars[key] = value
|
|
self.changed = True
|
|
if key.member_id == protocol.MEMBER_ID_UI_CONTROL:
|
|
self.parent.setUIControl(key, value)
|
|
self.lock.release()
|
|
return
|
|
|
|
def set_default(self, key, value):
|
|
if key.struct_id == protocol.STRUCT_ID_COMMAND:
|
|
return
|
|
else:
|
|
if key is None or key.channel is None or key.channel < 0:
|
|
print mytime.displayTime() + ' data_model.sd.invalid key!!', key
|
|
return
|
|
if key in self.vars.keys():
|
|
return
|
|
self.vars[key] = value
|
|
return
|
|
|
|
def key_with_prototype_channel(self, key):
|
|
key = Key(key.struct_id, key.member_id, -1, key.num)
|
|
return key
|
|
|
|
def set_var_name(self, key, name):
|
|
self.var_names[key] = name
|
|
return
|
|
|
|
def get_var_name(self, key):
|
|
try:
|
|
return self.var_names[key]
|
|
except:
|
|
return
|
|
|
|
return
|
|
|
|
def set_min(self, key, val):
|
|
try:
|
|
tmp = float(val)
|
|
self.min_values[key] = val
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
def set_max(self, key, val):
|
|
try:
|
|
tmp = float(val)
|
|
self.max_values[key] = val
|
|
except:
|
|
pass
|
|
|
|
return
|
|
|
|
def set_min_build_number(self, key, val):
|
|
self.min_build_number[key] = val
|
|
return
|
|
|
|
def get_min(self, key):
|
|
try:
|
|
res = self.min_values[key]
|
|
except:
|
|
return
|
|
|
|
if key.member_id == protocol.MEMBER_ID_THRESHOLD and key.channel in self.parent.thresholdCorrection:
|
|
res += self.parent.thresholdCorrection[key.channel]
|
|
return res
|
|
|
|
def get_max(self, key):
|
|
try:
|
|
res = self.max_values[key]
|
|
except:
|
|
return
|
|
|
|
if key.member_id == protocol.MEMBER_ID_THRESHOLD and key.channel in self.parent.thresholdCorrection:
|
|
res += self.parent.thresholdCorrection[key.channel]
|
|
return res
|
|
|
|
def resetFIR(self):
|
|
for testkey in self.get_list[:]:
|
|
if testkey.member_id == protocol.MEMBER_ID_FIR:
|
|
self.get_list.remove(testkey)
|
|
|
|
for testkey in self.vars.keys():
|
|
if testkey.member_id == protocol.MEMBER_ID_FIR:
|
|
del self.vars[testkey]
|
|
|
|
return
|
|
|
|
def reset(self):
|
|
self.vars = {}
|
|
try:
|
|
self.parent.server.peers.units[self.parent.MAC].process.conn.get_list = []
|
|
except:
|
|
pass
|
|
|
|
self.get_list = []
|
|
print mytime.displayTime() + ' dm.reset'
|
|
self.preset_names = {}
|
|
return
|
|
|
|
def access_granted(self, key):
|
|
if self.parent.full_access == True:
|
|
return True
|
|
access_granted = True
|
|
arkey = Key(protocol.STRUCT_ID_GLOBAL, protocol.MEMBER_ID_ACCESS_RIGHTS, 0, 0)
|
|
try:
|
|
accessrights = int(self.get(arkey))
|
|
except:
|
|
accessrights = 0
|
|
|
|
if key.struct_id == protocol.STRUCT_ID_GLOBAL:
|
|
if key.member_id == protocol.MEMBER_ID_CHECK_PASSWORD:
|
|
return True
|
|
if key.member_id == protocol.MEMBER_ID_SHORT_NAME:
|
|
if accessrights & protocol.arUnitName == 0:
|
|
access_granted = False
|
|
elif key.member_id == protocol.MEMBER_ID_STANDBY_DELAY:
|
|
if accessrights & protocol.arUnitConfiguration == 0:
|
|
access_granted = False
|
|
else:
|
|
return True
|
|
elif key.struct_id == protocol.STRUCT_ID_PRESET_GLOBAL:
|
|
return True
|
|
if key.channel < 128 and accessrights & protocol.arInChannel == 0:
|
|
access_granted = False
|
|
if key.channel >= 128 and accessrights & protocol.arOutChannel == 0:
|
|
access_granted = False
|
|
if access_granted == False:
|
|
return False
|
|
return True
|
|
|
|
def completion(self, key=None):
|
|
if one_unit.app.application_stopping == True:
|
|
return 100
|
|
else:
|
|
if key != None:
|
|
try:
|
|
if key in self.parent.server.peers.units[self.parent.MAC].process.conn.get_list:
|
|
return 0
|
|
else:
|
|
return 100
|
|
except:
|
|
if self.parent.MAC[:4] not in ('DEMO', 'GRP_', 'VN::'):
|
|
return 0
|
|
else:
|
|
return 100
|
|
|
|
if len(self.get_list) == 0:
|
|
completion = 0
|
|
else:
|
|
try:
|
|
if len(self.get_list) < len(self.parent.server.peers.units[self.parent.MAC].process.conn.get_list):
|
|
for key in self.parent.server.peers.units[self.parent.MAC].process.conn.get_list:
|
|
if key not in self.get_list:
|
|
self.get_list.append(key)
|
|
|
|
completion = (len(self.get_list) - len(self.parent.server.peers.units[self.parent.MAC].process.conn.get_list)) * 100 / len(self.get_list)
|
|
if completion < 0:
|
|
completion = 0
|
|
except:
|
|
completion = 0
|
|
|
|
if one_unit.app.printCompletion != None and mytime.clock() - self.lastPrintCompletionTime > 3 and completion < 100 and (completion >= one_unit.app.printCompletion or completion < 0):
|
|
self.lastPrintCompletionTime = mytime.clock()
|
|
try:
|
|
print mytime.displayTime() + ' dm.Completion for ' + self.parent.MAC + ':', completion, 'Get list:', self.parent.server.peers.units[self.parent.MAC].process.conn.get_list
|
|
except:
|
|
pass
|
|
|
|
return completion
|
|
|
|
def synced(self, key=None):
|
|
if one_unit.app.application_stopping == True:
|
|
print mytime.displayTime() + ' data_model.synced(): application stopping'
|
|
return True
|
|
else:
|
|
if self.parent.MAC[:4] in ('DEMO', 'GRP_', 'VN::'):
|
|
return True
|
|
if self.parent.MAC[:4] == 'GRP_':
|
|
group = self.parent.MAC[4:]
|
|
maclist = one_unit.app.groups[group]
|
|
for mac in maclist:
|
|
if key == None:
|
|
try:
|
|
if self.parent.server.peers.units[mac].process.conn.full_synced == False or self.parent.server.peers.units[self.parent.MAC].process.conn.synced == False:
|
|
return False
|
|
except:
|
|
pass
|
|
|
|
else:
|
|
try:
|
|
if self.parent.server.peers.units[mac].process.conn.values[key].write_val != None or key in self.parent.server.peers.units[mac].process.conn.get_list:
|
|
return False
|
|
except:
|
|
pass
|
|
|
|
return True
|
|
if key == None:
|
|
try:
|
|
if self.parent.server.peers.units[self.parent.MAC].process.conn.full_synced == True and self.parent.server.peers.units[self.parent.MAC].process.conn.synced == True:
|
|
return True
|
|
else:
|
|
return False
|
|
except:
|
|
return False
|
|
|
|
else:
|
|
try:
|
|
if key in self.parent.server.peers.units[self.parent.MAC].process.conn.get_list:
|
|
return False
|
|
except:
|
|
return False
|
|
|
|
try:
|
|
if self.parent.server.peers.units[self.parent.MAC].process.conn.values[key].write_val != None:
|
|
return False
|
|
except:
|
|
if key.struct_id in (protocol.STRUCT_ID_LPF, protocol.STRUCT_ID_HPF) and key.member_id in (protocol.MEMBER_ID_GAIN, protocol.MEMBER_ID_Q):
|
|
if key in self.parent.server.peers.units[self.parent.MAC].process.conn.get_list:
|
|
return False
|
|
else:
|
|
return True
|
|
print mytime.displayTime() + ' dm.s.Unknown key', key
|
|
return True
|
|
else:
|
|
return True
|
|
|
|
return
|
|
|
|
|
|
|
|
# okay decompiling pycode/data_model.pyc
|