import array class ArrayUintRef: """对 array 某个元素的引用代理""" def __init__(self, arr, index, t = 0): self.arr = arr self.index = index self.t = t @property def value(self): val = 0 if self.t == 0: val = self.arr[self.index] elif self.t == 1: val = (self.arr[self.index] | (self.arr[self.index + 1] << 8)) elif self.t == 2: val = (self.arr[self.index] | (self.arr[self.index + 1] << 8) | (self.arr[self.index + 2] << 16) | (self.arr[self.index + 3] << 24)) #elif self.t == 3: else: val = (self.arr[self.index] | (self.arr[self.index + 1] << 8) | (self.arr[self.index + 2] << 16) | (self.arr[self.index + 3] << 24) | (self.arr[self.index + 4] << 32) | (self.arr[self.index + 5] << 40) | (self.arr[self.index + 6] << 48) | (self.arr[self.index + 7] << 56)) return val @value.setter def value(self, val): self.arr[self.index] = (val & 0xff) if self.t >= 1: self.arr[self.index + 1] = ((val >> 8) & 0xff) if self.t >= 2: self.arr[self.index + 2] = ((val >> 16) & 0xff) self.arr[self.index + 3] = ((val >> 24) & 0xff) if self.t >= 3: self.arr[self.index + 4] = ((val >> 32) & 0xff) self.arr[self.index + 5] = ((val >> 40) & 0xff) self.arr[self.index + 6] = ((val >> 48) & 0xff) self.arr[self.index + 7] = ((val >> 56) & 0xff) #print('setter:', self.t) def __repr__(self): return f"ArrayUintRef(value={self.value})" class MemroyQueueMgr: class MemoryQueue: def __init__(self, mq): self.mq = mq self.data = list() def __getitem__(self, index): if len(self.data) == 0: if isinstance(index, slice): return list()[index] else: return 0 val = self.data[0] if isinstance(index, slice): return val[index] else: if index < 2048: return val[index] else: return 0 def __setitem__(self, index, value): if len(self.data) != 0: pass if index < 2048: self.data[0][index] = value def __len__(self): return len(self.data) def deq(self): if len(self.data) != 0: v = self.data.pop(0) return v else: return None def enq(self, item): if type(item) != array.array: print('Error, item type:', type(item)) return else: self.data.append(item) def reset(self): self.data = list() def __init__(self, item_num): self.memory = list() self.item_num = item_num self.din = MemroyQueueMgr.MemoryQueue(self) self.dout = MemroyQueueMgr.MemoryQueue(self) self.reset() def reset(self): self.din.reset() self.dout.reset() self.memory = list() for i in range(self.item_num): self.memory.append(array.array('B', [0] * 2048)) for i in range(self.item_num): self.din.enq(self.memory[i]) def __getitem__(self, index): return self.dout[index] def __setitem__(self, index, value): self.din[index] = value def __len__(self): return len(self.dout) def enq_data(self): self.dout.enq(self.din.deq()) def free_data(self): #这里 self.din.enq(self.dout.deq()) @staticmethod def clear(arr): if type(arr) != array.array: return for i in range(len(arr)): arr[i] = 0 class device(object): def __init__(self): super().__init__() self.reset() def reset(self): self.reg = array.array('B', [0, ] * 2048) # 配置各SM的item数量（sm0~sm7） self.sm = dict() syncMemNum = [4, 4, 4, 4, 2, 2, 2, 2] for i in range(8): self.sm[f'mem{i}'] = MemroyQueueMgr(syncMemNum[i]) self.sm[f'mem{i}_ctrl'] = ArrayUintRef(self.reg, 0x300 + i * 4) self.sm[f'mem{i}_item_num'] = ArrayUintRef(self.reg, 0x301 + i * 4) self.sm[f'mem{i}_len'] = ArrayUintRef(self.reg, 0x302 + i * 4, 1) #print(self.sm) self.ebmmu = dict() for i in range(16): self.ebmmu[f'va{i}'] = ArrayUintRef(self.reg, 0x200 + i * 0xa, 2) self.ebmmu[f'pa{i}'] = ArrayUintRef(self.reg, 0x204 + i * 0xa, 1) self.ebmmu[f'op{i}'] = ArrayUintRef(self.reg, 0x208 + i * 0xa, 0) self.ebmmu[f'en{i}'] = ArrayUintRef(self.reg, 0x209 + i * 0xa, 0) #看门狗 self.watchdog = dict() self.watchdog["status"] = ArrayUintRef(self.reg, 0x400, 2) self.watchdog["enable"] = ArrayUintRef(self.reg, 0x404, 2) for i in range(18): self.watchdog[f"{i}"] = ArrayUintRef(self.reg, 0x408 + i * 4, 2) #事件 self.event = dict() self.event['ctrl0'] = ArrayUintRef(self.reg, 0x500, 1) self.event['ctrl1'] = ArrayUintRef(self.reg, 0x502, 1) self.event['status0'] = ArrayUintRef(self.reg, 0x504, 1) self.event['status1'] = ArrayUintRef(self.reg, 0x506, 1) self.event['rise0'] = ArrayUintRef(self.reg, 0x508, 3) self.event['fall0'] = ArrayUintRef(self.reg, 0x510, 3) self.event['rise1'] = ArrayUintRef(self.reg, 0x518, 3) self.event['fall1'] = ArrayUintRef(self.reg, 0x520, 3) self.event['host_sm_w'] = ArrayUintRef(self.reg, 0x528, 3) self.event['host_sm_r'] = ArrayUintRef(self.reg, 0x530, 3) self.event['lvds_sm_w'] = ArrayUintRef(self.reg, 0x538, 3) self.event['lvds_sm_r'] = ArrayUintRef(self.reg, 0x540, 3) #主/从设备 self.role = ArrayUintRef(self.reg, 0) #硬件版本号 self.reg[1] = 0x1 #硬件编码 self.reg[2] = 0x4c self.reg[3] = 0x58 self.reg[4] = 0x50 self.reg[5] = 0x30 self.reg[6] = 0x30 self.reg[7] = 0x31 #硬件接口 self.reg[9] = 0x01 #中断状态 self.reg[0x73] = 0x80 self.reg[0x77] = 0x80 self.int_en = ArrayUintRef(self.reg, 0x70, 2) self.int_status = ArrayUintRef(self.reg, 0x74, 2) self.int_clear = ArrayUintRef(self.reg, 0x78, 2) self.int_mode = ArrayUintRef(self.reg, 0x7c, 1) #SYNC同步点 self.sync_point = ArrayUintRef(self.reg, 0x130, 3) self.sync_point.value = 0xffffffffffffffff #print('sync_point', self.reg[0x130:0x138], hex(self.sync_point.value)) ''' self.reg[0x130] = 0xff self.reg[0x131] = 0xff self.reg[0x132] = 0xff self.reg[0x133] = 0xff self.reg[0x134] = 0xff self.reg[0x135] = 0xff self.reg[0x136] = 0xff self.reg[0x137] = 0xff ''' self.__init_temp_regs() self.next_dev = None self.prev_dev = None def connect_next(self, dev): if isinstance(dev, device): self.next_dev = dev dev.prev_dev = self return True return False def __init_temp_regs(self): #临时寄存器 #LVDS侧写入锁 self.lvds_0x18 = 0x0 self.lvds_0x19 = 0x0 #RESET self.reset_0x40 = 0x0 self.reset_0x41 = 0x0 self.reset_0x42 = 0x0 self.reset_0x43 = 0x0 #交换上下行管脚 self.swap_dir_0x4a = 0x0 self.swap_dir_0x4b = 0x0 #code_sel, 4b5b, 8b10b self.code_sel_0x4c = 0x0 self.code_sel_0x4d = 0x0 #cdr_type self.cdr_type_0x4e = 0x0 self.cdr_type_0x4f = 0x0 #int_status temp reg self.int_status = [0x0, 0x0, 0x0, 0x80] def hostwrite(self, addr, data): l = len(data) ret = False for k, v in enumerate(range(addr, addr+l)): ret = (self.__ctrl_regs_hw(v, data[k]) or self.__reset_ctrl_regs_hw(v, data[k]) or self.__int_regs_hw(v, data[k]) or self.__stat_regs_hw(v, data[k]) or self.__link_regs_hw(v, data[k]) or self.__sync_regs_hw(v, data[k]) or self.__custom_def_regs_hw(v, data[k]) or self.__ebmmu_regs_hw(v, data[k]) or self.__sm_regs_hw(v, data[k]) or self.__watch_dog_regs_hw(v, data[k]) or self.__event_regs_hw(v, data[k]) ) # self.__init_temp_regs() return ret def hostread(self, addr, l): data = [0, ] * l for k, v in enumerate(range(addr, addr+l)): if v < 2048: data[k] = self.reg[v] if v >= 2048: pass return data def trans_2_next_dev(self, packet): pass def trans_2_prev_dev(self, packet): pass def __ctrl_regs_hw(self, regaddr, val): if regaddr == 0x00: self.reg[regaddr] = val & 0xff elif regaddr == 0x01: # 硬件版本号，写入无效 pass elif (regaddr >= 0x02) and (regaddr <= 0x07): # 硬件编码，写入无效 pass elif regaddr == 0x09: #硬件接口(0,1,2,3,4,5) self.reg[regaddr] = val & 0xff elif (regaddr >= 0x10) and (regaddr <= 0x17): #硬件唯一编码，写入无效 pass elif (regaddr >= 0x18) and (regaddr <= 0x19): #lvds写入锁 if self.role.value == 1: pass elif self.role.value == 2: if regaddr == 0x18: self.lvds_0x18 = val & 0x1 elif regaddr == 0x19: if val == 0x58: self.reg[regaddr] = self.lvds_0x18 else: pass else: return False return True def __reset_ctrl_regs_hw(self, regaddr, val): #EasyBus复位 if (regaddr >= 0x40) and (regaddr <= 0x43): if regaddr == 0x40: self.reset_0x40 = val & 0x1 elif regaddr == 0x41: self.reset_0x41 = 0 elif regaddr == 0x42: self.reset_0x42 = val & 0xff elif regaddr == 0x43: self.reset_0x43 = val & 0xff else: pass if (self.reset_0x42 == 0x66) and (self.reset_0x43 == 0x79): if self.reset_0x40 == 0x1: #重启芯片 self.reset_0x42 = 0 self.reset_0x43 = 0 #self.reset_0x40 = 0 #self.reset_0x41 = 0 else: pass #交换上下行管脚 elif (regaddr >= 0x4a) and (regaddr <= 0x4b): if regaddr == 0x4a: self.swap_dir_0x4a = val & 0xff elif regaddr == 0x4b: self.swap_dir_0x4b = val & 0xff if self.swap_dir_0x4b == 0x58: self.reg[0x4a] = self.swap_dir_0x4a self.swap_dir_0x4b = 0x0 #进行管脚交换 else: pass else: pass #4b5b,8b10b选择 #BIT0:0,选择8b10b编码；1选择4b5b编码 #BIT15-BIT8:key，当KEY = 0x59，且BIT0为1时，4b5b；BIT0为0时，8b10b。 #读取时KEY始终为0。KEY不正确，不响应修改 elif (regaddr >= 0x4c) and (regaddr <= 0x4d): if regaddr == 0x4c: self.code_sel_0x4c = val & 0xff elif regaddr == 0x4d: self.code_sel_0x4d = val & 0xff if self.code_sel_0x4d == 0x59: self.reg[0x4c] = self.code_sel_0x4c self.code_sel_0x4d = 0 #切换编码 else: pass else: pass #CDR参数 elif (regaddr >= 0x4e) and (regaddr <= 0x4f): if regaddr == 0x4e: self.cdr_type_0x4e = val & 0xff elif regaddr == 0x4f: self.cdr_type_0x4f = val & 0xff if self.cdr_type_0x4f == 0x60: self.reg[0x4e] = self.cdr_type_0x4e self.cdr_type_0x4f = 0x0 #执行cdr参数 else: pass else: pass #4倍频分频系数 elif regaddr == 0x55: pass else: return False return True def __int_regs_hw(self, regaddr, val): #中断使能 if (regaddr >= 0x70) and (regaddr <= 0x73): self.reg[regaddr] = val & 0xff self.reg[0x73] = self.reg[0x73] | 0x80 #中断状态 elif (regaddr >= 0x74) and (regaddr <= 0x77): #只读 #warning pass #中断状态清除 elif (regaddr >= 0x78) and (regaddr <= 0x7b): self.reg[regaddr - 0x4] = (~val) & self.reg[regaddr - 0x4] self.int_status[regaddr - 0x78] = (~val) & self.int_status[regaddr - 0x78] #中断模式 elif (regaddr >= 0x7c) and (regaddr <= 0x7d): self.reg[regaddr] = val & 0xff else: return False return True def __stat_regs_hw(self, regaddr, val): #统计 #下行接收错误计数器（包括包不完全、CDR、CRC、超时） if (regaddr >= 0x80) and (regaddr <= 0x81): regaddr[0x80] = 0 regaddr[0x81] = 0 elif (regaddr >= 0x82) and (regaddr <= 0x83): regaddr[0x82] = 0 regaddr[0x83] = 0 elif (regaddr >= 0x84) and (regaddr <= 0x85): regaddr[0x84] = 0 regaddr[0x85] = 0 elif (regaddr >= 0x86) and (regaddr <= 0x87): regaddr[0x86] = 0 regaddr[0x87] = 0 elif (regaddr >= 0x88) and (regaddr <= 0x89): regaddr[0x88] = 0 regaddr[0x89] = 0 elif (regaddr >= 0x8a) and (regaddr <= 0x8b): regaddr[0x8a] = 0 regaddr[0x8b] = 0 elif (regaddr >= 0x8c) and (regaddr <= 0x8f): regaddr[0x8c] = 0 regaddr[0x8f] = 0 elif (regaddr >= 0x90) and (regaddr <= 0x97): regaddr[0x90] = 0 regaddr[0x91] = 0 regaddr[0x92] = 0 regaddr[0x93] = 0 regaddr[0x94] = 0 regaddr[0x95] = 0 regaddr[0x96] = 0 regaddr[0x97] = 0 #上行 elif (regaddr >= 0xa0) and (regaddr <= 0xa1): regaddr[0xa0] = 0 regaddr[0xa1] = 0 elif (regaddr >= 0xa2) and (regaddr <= 0xa3): regaddr[0xa2] = 0 regaddr[0xa3] = 0 elif (regaddr >= 0xa4) and (regaddr <= 0xa5): regaddr[0xa4] = 0 regaddr[0xa5] = 0 elif (regaddr >= 0xa6) and (regaddr <= 0xa7): regaddr[0xa6] = 0 regaddr[0xa7] = 0 elif (regaddr >= 0xa8) and (regaddr <= 0xa9): regaddr[0xa8] = 0 regaddr[0xa9] = 0 elif (regaddr >= 0xaa) and (regaddr <= 0xab): regaddr[0xaa] = 0 regaddr[0xab] = 0 elif (regaddr >= 0xac) and (regaddr <= 0xaf): regaddr[0xac] = 0 regaddr[0xad] = 0 elif (regaddr >= 0xb0) and (regaddr <= 0xb7): regaddr[0xb0] = 0 regaddr[0xb1] = 0 regaddr[0xb2] = 0 regaddr[0xb3] = 0 regaddr[0xb4] = 0 regaddr[0xb5] = 0 regaddr[0xb6] = 0 regaddr[0xb7] = 0 else: return False return True def __link_regs_hw(self, regaddr, val): #两个寄存器都是只读 #链路状态 if (regaddr >= 0xc0) and (regaddr <= 0xc1): pass #右侧设备地址 elif (regaddr >= 0xc2) and (regaddr <= 0xc3): pass else: return False return True def __sync_regs_hw(self, regaddr, val): #同步 #这里注意一个问题，主站是否要考虑lvds侧锁不锁的问题 if (regaddr >= 0x0100) and (regaddr <= 0x107): pass elif (regaddr >= 0x0108) and (regaddr <= 0x10f): pass elif (regaddr >= 0x0110) and (regaddr <= 0x113): pass elif (regaddr >= 0x0114) and (regaddr <= 0x117): pass elif (regaddr >= 0x0118) and (regaddr <= 0x11f): pass elif (regaddr >= 0x0120) and (regaddr <= 0x121): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0122) and (regaddr <= 0x123): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0124) and (regaddr <= 0x125): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0126) and (regaddr <= 0x127): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0128) and (regaddr <= 0x129): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0130) and (regaddr <= 0x137): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0138) and (regaddr <= 0x13b): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x013c) and (regaddr <= 0x13f): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0140) and (regaddr <= 0x143): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0144) and (regaddr <= 0x147): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif (regaddr >= 0x0148) and (regaddr <= 0x14f): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff else: return False return True def __custom_def_regs_hw(self, regaddr, val): if (regaddr >= 0x180) and (regaddr <= 0x1FF): self.reg[regaddr] = val & 0xff else: return False return True def __ebmmu_regs_hw(self, regaddr, val): ebmmu_base = 0x200 ebmmu_no = 16 for i in range(ebmmu_no): if (ebmmu_base + i * 0xa) <= regaddr <= ((ebmmu_base + i * 0xa) + 0x3): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif ((ebmmu_base + i * 0xa) + 0x4) <= regaddr <= ((ebmmu_base + i * 0xa) + 0x5): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif ((ebmmu_base + i * 0xa) + 0x6) <= regaddr <= ((ebmmu_base + i * 0xa) + 0x7): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif regaddr == ((ebmmu_base + i * 0xa) + 0x8): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff elif regaddr == ((ebmmu_base + i * 0xa) + 0x9): if (self.reg[0x18] & 0x1) == 0x1: self.reg[regaddr] = val & 0xff else: continue return True return False def __sm_regs_hw(self, regaddr, val): #数据缓冲区描述寄存器组 ''' self.sm[f'mem{i}'] = MemroyQueueMgr(syncMemNum[i]) self.sm[f'mem{i}_ctrl'] = ArrayUintRef(self.reg, 0x300 + i * 4) self.sm[f'mem{i}_item_num'] = ArrayUintRef(self.reg, 0x301 + i * 4) self.sm[f'mem{i}_len'] = ArrayUintRef(self.reg, 0x302 + i * 4, 1) ''' for i in range(8): if regaddr == (0x300 + i * 0x4): self.sm[f'mem{i}_ctrl'].value = 0 #复位 self.sm[f'mem{i}'].reset() self.sm[f'mem{i}_item_num'].value = 0 elif regaddr == (0x301 + i * 0x4): pass elif (0x302 + i * 0x4) <= regaddr <= (0x303 + i * 0x4): if self.sm[f'mem{i}_item_num'].value == 0: self.reg[regaddr] = val & 0xff else: continue return True for i in range(8): if (0x800 + i * 0x800) <= regaddr < (0x1000 + i * 0x800): self.sm[f'mem{i}'][regaddr - 0x800 - i * 0x800] = val if (regaddr - 0x800 - i * 0x800 + 1) == self.sm[f'mem{i}_len'].value: self.sm[f'mem{i}'].enq_data() self.sm[f'mem{i}_item_num'].value = len(self.sm[f'mem{i}']) return True return False def __watch_dog_regs_hw(self, regaddr, val): #看门狗 if (regaddr >= 0x400) and (regaddr <= 0x403): pass elif (regaddr >= 0x404) and (regaddr <= 0x407): pass elif (regaddr >= 0x408) and (regaddr <= 0x40B): pass elif (regaddr >= 0x40C) and (regaddr <= 0x40F): pass elif (regaddr >= 0x410) and (regaddr <= 0x413): pass elif (regaddr >= 0x414) and (regaddr <= 0x417): pass elif (regaddr >= 0x418) and (regaddr <= 0x40B): pass elif (regaddr >= 0x41C) and (regaddr <= 0x41F): pass elif (regaddr >= 0x420) and (regaddr <= 0x423): pass elif (regaddr >= 0x424) and (regaddr <= 0x427): pass elif (regaddr >= 0x428) and (regaddr <= 0x42B): pass elif (regaddr >= 0x42C) and (regaddr <= 0x42F): pass elif (regaddr >= 0x430) and (regaddr <= 0x433): pass elif (regaddr >= 0x434) and (regaddr <= 0x437): pass elif (regaddr >= 0x438) and (regaddr <= 0x43B): pass elif (regaddr >= 0x43C) and (regaddr <= 0x43F): pass elif (regaddr >= 0x440) and (regaddr <= 0x443): pass elif (regaddr >= 0x444) and (regaddr <= 0x447): pass elif (regaddr >= 0x448) and (regaddr <= 0x44B): pass elif (regaddr >= 0x44C) and (regaddr <= 0x44F): pass else: return False return True def __event_regs_hw(self, regaddr, val): if (regaddr >= 0x500) and (regaddr <= 0x501): pass elif (regaddr >= 0x502) and (regaddr <= 0x503): pass elif (regaddr >= 0x504) and (regaddr <= 0x505): pass elif (regaddr >= 0x506) and (regaddr <= 0x507): pass elif (regaddr >= 0x508) and (regaddr <= 0x50F): pass elif (regaddr >= 0x510) and (regaddr <= 0x517): pass elif (regaddr >= 0x518) and (regaddr <= 0x51F): pass elif (regaddr >= 0x520) and (regaddr <= 0x527): pass elif (regaddr >= 0x528) and (regaddr <= 0x52F): pass elif (regaddr >= 0x530) and (regaddr <= 0x537): pass elif (regaddr >= 0x538) and (regaddr <= 0x53F): pass elif (regaddr >= 0x540) and (regaddr <= 0x547): pass else: return False return True if __name__ == '__main__': m = device() d0 = device() d1 = device() d2 = device() d3 = device() m.connect_next(d0) d0.connect_next(d1) d1.connect_next(d2) d2.connect_next(d3) m.hostwrite(0, [2,]) data = m.hostread(0, 128) #print(data) data = m.hostread(0x74, 4) #print(data) m.hostwrite(0x78, [0xff, 0xff, 0xff, 0xff]) data = m.hostread(0x74, 4) #print(data) sm0 = MemroyQueueMgr(4) sm0[0] = 1 sm0[1] = 2 sm0[2] = 3 sm0[3] = 4 print(len(sm0)) sm0.enq_data() print(len(sm0)) sm0[0] = 4 sm0[1] = 3 sm0[2] = 2 sm0[3] = 1 sm0.enq_data() print(len(sm0)) #print(len(sm0.din)) #print(len(sm0.dout)) print(sm0[:4], len(sm0)) sm0.free_data() print(sm0[:4], len(sm0)) sm0.free_data() print(sm0[:4], len(sm0)) #sm0.clear(sm0.dout.data[0]) #print(sm0.dout.data[0][:4]) sm0.reset() sm0[0] = 1 sm0[1] = 2 sm0[2] = 3 sm0[3] = 4 print(len(sm0)) sm0.enq_data() print(len(sm0)) sm0[0] = 4 sm0[1] = 3 sm0[2] = 2 sm0[3] = 1 sm0.enq_data() print(len(sm0)) #print(len(sm0.din)) #print(len(sm0.dout)) print(sm0[:4], len(sm0)) sm0.free_data() print(sm0[:4], len(sm0)) sm0.free_data() print(sm0[:4], len(sm0))