Roy
518 lines
19 KiB
Python
518 lines
19 KiB
Python
from functools import lru_cache
|
|
from random import randint
|
|
from time import sleep
|
|
from typing import List, Tuple, Optional, Union, Iterable
|
|
|
|
from biopro.impl.selector import Selector
|
|
from biopro.util.console import pc, GREEN, RED
|
|
from .gpio import P3Pin, OutputPin, InputPin
|
|
from .interface_spi import HardwareImplSpiInterface, LowLevelHardwareInterface
|
|
from datetime import datetime
|
|
|
|
MSM_REG_WRITE = 0x01
|
|
MEM_INS_WRITE = 0x02
|
|
MEM_INS_READ = 0x03
|
|
MEM_REG_READ = 0x05
|
|
DEFAULT_REGISTER_VALUE = 0b0100_0011 # 67
|
|
MEM_SIZE = 0x1000
|
|
|
|
_SLEEP_TIME_ = 0.001
|
|
|
|
|
|
@lru_cache(maxsize=4)
|
|
def zero_buffer(size: int) -> List[int]:
|
|
return [0] * size
|
|
|
|
class MultiExtMemSpiInterface(LowLevelHardwareInterface):
|
|
MEM_INS_MARKED1 = [MEM_INS_WRITE, 0, 2, 0x5A, 0xA5]
|
|
MEM_INS_MARKED2 = [MEM_INS_WRITE, 0, 6, 0x5A, 0xA5]
|
|
MEM_INS_MARKED3 = [MEM_INS_WRITE, 0, 10, 0x5A, 0xA5]
|
|
# MEM_INS_RESET = [MEM_INS_WRITE, 0, 2, 1, 1, 0, 0xFF]
|
|
|
|
__slots__ = ('_selector', '_wait_for_first_data', '_spi', '_tx_buffer_header', '_tx_buffer_data',
|
|
'pin_busy', 'pin_mem_req', 'pin_mem_sel', 'pin_ram_sel',
|
|
'_pin_sel_val',
|
|
'_pin_ram_sel_value', '_pin_mem_sel_value', '_pin_mem_req_value',
|
|
'_read_green_times','_read_red_times',
|
|
'_elite_data_len', '_mem_header_len', '_mem_tailer_len', '_single_data_len',
|
|
'_head_wrong_cnt')
|
|
|
|
def __init__(self,
|
|
select: Selector,
|
|
device: Tuple[int, int] = None):
|
|
|
|
self._spi = HardwareImplSpiInterface(device,
|
|
spi_speed=12_000_000)
|
|
|
|
self._elite_data_len = 40
|
|
self._mem_header_len = 3
|
|
self._mem_tailer_len = 6
|
|
self._single_data_len = self._elite_data_len + self._mem_header_len + self._mem_tailer_len
|
|
|
|
# buffer
|
|
self._tx_buffer_header = [0] * 19
|
|
self._tx_buffer_data = [0] * (self._single_data_len * 10 + 3)
|
|
|
|
# memory control pin
|
|
self.pin_busy = OutputPin.get_used(P3Pin.MEM_BZY, True)
|
|
self.pin_mem_req = OutputPin.get_used(P3Pin.MEM_REQ, False)
|
|
self.pin_mem_sel = OutputPin.get_used(P3Pin.MEM_RST, True) # MEM_RST -> actually which memory board is assign
|
|
self.pin_ram_sel: Optional[InputPin] = InputPin.get_used(P3Pin.MEM_SEL) # MEM_SEL -> actually is RAM_SEL, which RAM is assign
|
|
|
|
self._pin_ram_sel_value = [bool(self.pin_ram_sel) for _ in range(Selector.SIZE)]
|
|
self._pin_mem_sel_value = [bool(self.pin_mem_sel) for _ in range(Selector.SIZE)]
|
|
self._pin_mem_req_value = [bool(self.pin_mem_req) for _ in range(Selector.SIZE)]
|
|
|
|
self._read_green_times = 0
|
|
self._read_red_times = 0
|
|
|
|
self._head_wrong_cnt = [0, 0, 0, 0, 0, 0, 0, 0]
|
|
|
|
|
|
self._selector = select
|
|
|
|
self._pin_sel_val = [False for _ in range(Selector.SIZE)]
|
|
self._wait_for_first_data = [True for _ in range(Selector.SIZE)]
|
|
|
|
def set_pin_mem_req(self, value: bool):
|
|
channel = self.select
|
|
self.pin_mem_req.output(value)
|
|
self._pin_mem_req_value[channel] = value
|
|
|
|
def set_pin_mem_sel(self, value: bool):
|
|
channel = self.select
|
|
self.pin_mem_sel.output(value)
|
|
self._pin_mem_sel_value[channel] = value
|
|
|
|
def get_pin_mem_req(self):
|
|
channel = self.select
|
|
return self._pin_mem_req_value[channel]
|
|
|
|
def get_pin_ram_sel(self):
|
|
channel = self.select
|
|
if self.pin_ram_sel.input() == 0:
|
|
self._pin_ram_sel_value[channel] = False
|
|
else:
|
|
self._pin_ram_sel_value[channel] = True
|
|
return self._pin_ram_sel_value[channel]
|
|
|
|
@property
|
|
def select(self) -> int:
|
|
return self._selector.channel
|
|
|
|
@select.setter
|
|
def select(self, value: int):
|
|
self._selector.select(value)
|
|
|
|
def reset(self):
|
|
self._spi.reset()
|
|
|
|
def flush(self):
|
|
pass
|
|
|
|
def close(self):
|
|
self._selector.close()
|
|
self._spi.close()
|
|
|
|
def send_byte(self, data: bytes):
|
|
raise RuntimeError()
|
|
|
|
def recv_byte(self, size: int) -> Optional[bytes]:
|
|
raise RuntimeError()
|
|
|
|
def foreach(self) -> Iterable[int]:
|
|
for channel in self._selector.foreach():
|
|
yield channel
|
|
|
|
def set_wait_flag(self, spi_idx:int = None, value:bool = None):
|
|
self._wait_for_first_data[spi_idx] = value
|
|
|
|
def get_wait_flag(self, spi_idx:int = None) -> Optional[bool]:
|
|
if spi_idx < Selector.SIZE:
|
|
return self._wait_for_first_data[spi_idx]
|
|
else:
|
|
return None
|
|
|
|
def compare_green_data(self, data_first: Union[bytes, List[int]], data_second: Union[bytes, List[int]], data_third: Union[bytes, List[int]], length: int):
|
|
data = []
|
|
data.append(255)
|
|
data.append(255)
|
|
data.append(255)
|
|
|
|
for i in range(3, length):
|
|
if (data_first[i] == data_second[i] or data_first[i] == data_third[i]):
|
|
data.append(data_first[i])
|
|
elif(data_second[i] == data_third[i]):
|
|
data.append(data_second[i])
|
|
else:
|
|
data.append(data_third[i]) # use last data
|
|
self._read_green_times = self._read_green_times + 1
|
|
print("read green data times", self._read_green_times)
|
|
return data
|
|
|
|
def compare_red_data(self, data_first: Union[bytes, List[int]], data_second: Union[bytes, List[int]], data_third: Union[bytes, List[int]], length: int):
|
|
data = []
|
|
data.append(255)
|
|
data.append(255)
|
|
data.append(255)
|
|
|
|
for i in range(3, length):
|
|
if (data_first[i] == data_second[i] or data_first[i] == data_third[i]):
|
|
data.append(data_first[i])
|
|
elif(data_second[i] == data_third[i]):
|
|
data.append(data_second[i])
|
|
else:
|
|
data.append(data_third[i]) # use last data
|
|
self._read_red_times = self._read_red_times + 1
|
|
print("read red data times", self._read_red_times)
|
|
return data
|
|
|
|
def _compare_green_data_addr_and_flag(self, data: Union[bytes, List[int]], device: int):
|
|
|
|
green_data_section1 = data[3:7]
|
|
green_data_section2 = data[7:11]
|
|
green_data_section3 = data[11:15]
|
|
green_data_section = []
|
|
|
|
if green_data_section1 == green_data_section2:
|
|
green_data_section = green_data_section1
|
|
|
|
elif(green_data_section2 == green_data_section3 or green_data_section1 == green_data_section3):
|
|
print("green data not equal: = ", data[3:15])
|
|
print("green data print:", data, device, datetime.now())
|
|
green_data_section = green_data_section3
|
|
|
|
else:
|
|
print("green data not equal: = ", data[3:15])
|
|
print("green data print:", data, device, datetime.now())
|
|
green_data_section = green_data_section3 # use last data
|
|
|
|
|
|
return green_data_section
|
|
|
|
def _print_ram_all_data(self):
|
|
addr = 0
|
|
red_length = int(7000 / 2)
|
|
tx_temp = [0] * (red_length + 3)
|
|
|
|
while True:
|
|
tx_temp[0] = MEM_INS_READ
|
|
tx_temp[1] = ((addr >> 8) & 0xFF)
|
|
tx_temp[2] = (addr & 0xFF)
|
|
|
|
ram_data = []
|
|
ram_data = self._spi.send_byte(tx_temp)
|
|
ram_data[0:3] = [255, 255, 255]
|
|
print(list(ram_data), len(ram_data), addr)
|
|
addr += len(ram_data) - 3
|
|
|
|
if (7000 < addr + len(ram_data)):
|
|
del ram_data
|
|
break
|
|
print()
|
|
|
|
return
|
|
|
|
def recv_memory(self, device: int) -> Optional[bytes]:
|
|
self.pin_busy.output(False)
|
|
|
|
rx = []
|
|
|
|
try:
|
|
address = 0
|
|
|
|
# first read to get data length
|
|
tx_h = self._tx_buffer_header
|
|
|
|
tx_h[0] = MEM_INS_READ
|
|
tx_h[1] = 0
|
|
tx_h[2] = 0
|
|
|
|
|
|
# -------------------------------------read one times-------------------------------------------
|
|
data = []
|
|
flag_print = False
|
|
|
|
data = self._spi.send_byte(tx_h)
|
|
|
|
if (data[0] != 255 or data[1] != 255 or data[2] != 255):
|
|
self._head_wrong_cnt[device] = self._head_wrong_cnt[device] + 1
|
|
if (self._head_wrong_cnt[device] <= 5): # print 5 times
|
|
print('data_first[0:3] != [255, 255, 255], device:', device, ',', self._head_wrong_cnt[device], 'times')
|
|
print(list(data))
|
|
|
|
data[0:3] = [255, 255, 255]
|
|
# ----------------------------------------------------------------------------------------------
|
|
|
|
|
|
# -------------------------------------read two times-------------------------------------------
|
|
# data = []
|
|
# data_first = []
|
|
# data_second = []
|
|
# flag_print = False
|
|
|
|
# data_first = self._spi.send_byte(tx_h)
|
|
# data_second = self._spi.send_byte(tx_h)
|
|
|
|
# if (data_first[0] != 255 or data_first[1] != 255 or data_first[2] != 255):
|
|
# self._head_wrong_cnt[device] = self._head_wrong_cnt[device] + 1
|
|
# if (self._head_wrong_cnt[device] < 10):
|
|
# print('data_first[0:3] != [255, 255, 255], device:', device, ',', self._head_wrong_cnt[device], 'times')
|
|
# print(list(data_first))
|
|
|
|
# if (data_second[0] != 255 or data_second[1] != 255 or data_second[2] != 255):
|
|
# self._head_wrong_cnt[device] = self._head_wrong_cnt[device] + 1
|
|
# if (self._head_wrong_cnt[device] < 10):
|
|
# print('data_second[0:3] != [255, 255, 255], device:', device, ',', self._head_wrong_cnt[device], 'times')
|
|
# print(list(data_second))
|
|
|
|
# if (data_first[3:] == data_second[3:]):
|
|
# data = data_first
|
|
# data[0:3] = [255, 255, 255]
|
|
|
|
# else:
|
|
# data_third = self._spi.send_byte(tx_h)
|
|
# print("read data_third", data_third)
|
|
# data = self.compare_green_data(data_first, data_second, data_third, len(tx_h))
|
|
# ----------------------------------------------------------------------------------------------
|
|
|
|
green_data_section = self._compare_green_data_addr_and_flag(data, device)
|
|
|
|
length = (green_data_section[0] << 8) | green_data_section[1]
|
|
header = data[15]
|
|
elite_data_len = data[17]
|
|
|
|
if length <= 12:
|
|
if length < 12:
|
|
print("length < 12")
|
|
print("green data print:", data, device, datetime.now())
|
|
|
|
if length == 0:
|
|
self._print_ram_all_data()
|
|
|
|
return None
|
|
|
|
if green_data_section[2] != 0xA5 or green_data_section[3] != 0x5A:
|
|
# give a default length if the data header did not update
|
|
print("green data: is not [0xA5, 0x5A], = ", data[5:7])
|
|
print("green data print:", data, device, datetime.now())
|
|
return None
|
|
|
|
if (length >= 4000):
|
|
flag_print = True
|
|
print("green data: big length:", length)
|
|
|
|
if (header != 255):
|
|
flag_print = True
|
|
print("green data: header is not 255: ", header)
|
|
self._print_ram_all_data()
|
|
|
|
if (elite_data_len != self._elite_data_len):
|
|
flag_print = True
|
|
print("green data: length is not", self._elite_data_len, ": ", elite_data_len)
|
|
|
|
if (flag_print):
|
|
print("green data print:", data, device, datetime.now(), '\n')
|
|
|
|
address = 12
|
|
red_length = int(length / 2)
|
|
tx_d = [0] * (red_length + 3)
|
|
|
|
while True:
|
|
tx_d[0] = MEM_INS_READ
|
|
tx_d[1] = ((address >> 8) & 0xFF)
|
|
tx_d[2] = (address & 0xFF)
|
|
# -------------------------------------read one times-------------------------------------------
|
|
data = []
|
|
data = self._spi.send_byte(tx_d)
|
|
data[0:3] = [255, 255, 255]
|
|
# ----------------------------------------------------------------------------------------------
|
|
|
|
# -------------------------------------read two times-------------------------------------------
|
|
# data = []
|
|
# data_first = []
|
|
# data_second = []
|
|
|
|
# data_first = self._spi.send_byte(tx_d)
|
|
# data_second = self._spi.send_byte(tx_d)
|
|
|
|
# if (data_first[3:] == data_second[3:]):
|
|
# data = data_first
|
|
# data[0:3] = [255, 255, 255]
|
|
# else:
|
|
# data_third = self._spi.send_byte(tx_d)
|
|
# print("read red data_third", data_third)
|
|
# data = self.compare_red_data(data_first, data_second, data_third, len(tx_d))
|
|
# ----------------------------------------------------------------------------------------------
|
|
if length + 3 <= address + len(data):
|
|
rx.extend(data[3:3 + length - address])
|
|
del data
|
|
|
|
break
|
|
|
|
else:
|
|
rx.extend(data[3:])
|
|
address += len(data) - 3
|
|
del data
|
|
|
|
# last_index = 0
|
|
# print("_[Debug] @ spi recv data rx, ram_select:", self._pin_ram_sel_value[device], ",", datetime.now())
|
|
# for i in range(0, len(rx), self._single_data_len):
|
|
# last_index = i
|
|
# # print(rx[i:i+self._single_data_len])
|
|
# if i == 0:
|
|
# print(rx[0:8], 'ram:', rx[47])
|
|
|
|
# if last_index != 0:
|
|
# print(rx[last_index:last_index+8], 'ram:', rx[last_index+47])
|
|
# print()
|
|
|
|
# /*
|
|
# * red data formate:
|
|
# * ramHdr, ramHdr, ramHdr, (3B)
|
|
# * 255, #, data_length, (3B)
|
|
# * data, (40B)
|
|
# * red_wrong, red_retry_cnt, green_wrong, green_retry_cnt, (4B)
|
|
# * check_num, (1B)
|
|
# * */
|
|
|
|
# read again if check num is wrong
|
|
index = 0
|
|
check_number_print = False
|
|
for i in range(0, len(rx), self._single_data_len):
|
|
check_sum = sum(rx[i : i + self._single_data_len - 1]) & 0b11111111
|
|
# print(check_sum, rx[i + self._single_data_len - 1])
|
|
|
|
if (check_sum != rx[i + self._single_data_len - 1]):
|
|
print('check_sum wrong, origin value:', 'check_sum =', check_sum, rx[i : i + self._single_data_len])
|
|
tx_d = [0] * (self._single_data_len + 3)
|
|
address = 12 + self._single_data_len * index
|
|
tx_d[0] = MEM_INS_READ
|
|
tx_d[1] = ((address >> 8) & 0xFF)
|
|
tx_d[2] = (address & 0xFF)
|
|
|
|
data = []
|
|
data = self._spi.send_byte(tx_d)
|
|
data[0:3] = [255, 255, 255]
|
|
rx[i : i + self._single_data_len] = data[3:]
|
|
print('check_sum wrong, read again:', 'check_sum =', check_sum, rx[i : i + self._single_data_len])
|
|
check_number_print = True
|
|
index = index + 1
|
|
|
|
if check_number_print:
|
|
print('check_sum wrong:', device, datetime.now())
|
|
for i in range(0, len(rx), self._single_data_len):
|
|
print(rx[i:i+self._single_data_len])
|
|
print()
|
|
|
|
# mark read
|
|
self._spi.send_byte(self.MEM_INS_MARKED1)
|
|
self._spi.send_byte(self.MEM_INS_MARKED2)
|
|
self._spi.send_byte(self.MEM_INS_MARKED3)
|
|
|
|
except BaseException as e:
|
|
print(e)
|
|
|
|
finally:
|
|
# print("\n")
|
|
self.pin_busy.output(True)
|
|
|
|
return bytes(rx)
|
|
|
|
def read_register(self) -> int:
|
|
return self._spi.send_byte([MEM_REG_READ, 0])[1]
|
|
|
|
def write_register(self, register: int):
|
|
self._spi.send_byte([MSM_REG_WRITE, register])
|
|
|
|
def read_memory(self, address: int, length: int) -> List[int]:
|
|
tx = [MEM_INS_READ, ((address >> 8) & 0xFF), (address & 0xFF)]
|
|
tx.extend(zero_buffer(length))
|
|
|
|
return self._spi.send_byte(tx)[3:]
|
|
|
|
def write_memory(self, address: int, data: Union[bytes, List[int]]):
|
|
tx = [MEM_INS_WRITE, ((address >> 8) & 0xFF), (address & 0xFF)]
|
|
|
|
if isinstance(data, bytes):
|
|
tx.extend(iter(data))
|
|
else:
|
|
tx.extend(data)
|
|
|
|
self._spi.send_byte(tx)
|
|
|
|
class ExtMemManager:
|
|
def __init__(self, ext_mem: MultiExtMemSpiInterface):
|
|
self._ext_mem = ext_mem
|
|
|
|
def get_ext_mem_register(self) -> List[Tuple[Optional[int], Optional[int]]]:
|
|
ret = [(None, None) for _ in range(Selector.SIZE)]
|
|
|
|
for channel in self._ext_mem.foreach():
|
|
r = [None, None]
|
|
|
|
self._ext_mem.set_pin_mem_sel(False)
|
|
pin_value = self._ext_mem.get_pin_mem_req()
|
|
self._ext_mem.set_pin_mem_req(not pin_value)
|
|
|
|
m1 = int(self._ext_mem.get_pin_ram_sel())
|
|
# print("m1 = ", m1)
|
|
self._ext_mem.write_register(DEFAULT_REGISTER_VALUE)
|
|
r[m1] = self._ext_mem.read_register()
|
|
sleep(0.040)
|
|
self._ext_mem.set_pin_mem_sel(True)
|
|
sleep(0.200)
|
|
|
|
self._ext_mem.set_pin_mem_sel(False)
|
|
pin_value = self._ext_mem.get_pin_mem_req()
|
|
self._ext_mem.set_pin_mem_req(not pin_value)
|
|
|
|
m2 = int(self._ext_mem.get_pin_ram_sel())
|
|
# print("m2 = ", m2)
|
|
self._ext_mem.write_register(DEFAULT_REGISTER_VALUE)
|
|
r[m2] = self._ext_mem.read_register()
|
|
sleep(0.040)
|
|
self._ext_mem.set_pin_mem_sel(True)
|
|
sleep(0.200)
|
|
|
|
ret[channel] = tuple(r)
|
|
|
|
return ret
|
|
|
|
@staticmethod
|
|
def is_no_device(result: Tuple[Optional[int], Optional[int]]) -> bool:
|
|
r1, r2 = result
|
|
return (r1 is None or r2 is None) and (r1 is None or r1 == 0) and (r2 is None or r2 == 0)
|
|
|
|
@staticmethod
|
|
def is_memory_test_fail(result: Tuple[Optional[int], Optional[int]]) -> int:
|
|
r1, r2 = result
|
|
|
|
if r1 is None or r2 is None:
|
|
return 1
|
|
|
|
if r1 is not None and r1 > 0 and r1 != DEFAULT_REGISTER_VALUE:
|
|
return 2
|
|
|
|
if r2 is not None and r2 > 0 and r2 != DEFAULT_REGISTER_VALUE:
|
|
return 3
|
|
|
|
return 0
|
|
|
|
def get_available_channel(self, result: List[Tuple[Optional[int], Optional[int]]] = None) -> List[int]:
|
|
if result is None:
|
|
result = self.get_ext_mem_register()
|
|
|
|
ret = []
|
|
|
|
for channel, result in enumerate(result):
|
|
if self.is_no_device(result):
|
|
continue
|
|
|
|
if self.is_memory_test_fail(result) != 0:
|
|
continue
|
|
|
|
ret.append(channel)
|
|
|
|
return ret
|