# 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