wisetop/controller-wisetopdataserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[update] clear not used code in class ExtMemManager

Browse Source
This commit is contained in:
Roy
2022-02-10 17:41:49 +08:00
parent a3c0f1e4a3
commit b6abf0885a
4 changed files with 27 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/biopro/device/device_cc2650.py
+5 -5
View File
@@ -315,12 +315,12 @@ class CC2650MasterDevice(MasterDevice, metaclass=abc.ABCMeta):
return cls.CC2650_RESET_PIN
@classmethod
def get_uart_irq_pin(cls) -> OutputPin:
if cls.CC2650_UART_IRQ is None:
cls.CC2650_UART_IRQ = OutputPin.get_used(P3Pin.MEM_RST, initial=True)
# @classmethod
# def get_uart_irq_pin(cls) -> OutputPin:
# if cls.CC2650_UART_IRQ is None:
# cls.CC2650_UART_IRQ = OutputPin.get_used(P3Pin.MEM_RST, initial=True)
return cls.CC2650_UART_IRQ
# return cls.CC2650_UART_IRQ
class CC2650Device(Device):
python/biopro/impl/ext_mem.py
+11 -234
View File
@@ -16,8 +16,6 @@ MEM_REG_READ = 0x05
DEFAULT_REGISTER_VALUE = 0b0100_0011 # 67
MEM_SIZE = 0x1000
_RUNTIME_COMPILE = False
_SLEEP_TIME_ = 0.001
@@ -30,7 +28,7 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
MEM_INS_RESET = [MEM_INS_WRITE, 0, 2, 1, 1, 0, 0xFF]
__slots__ = ('_selector', '_wait_for_first_data', '_spi', '_tx_buffer', '_tx_buffer_header', '_tx_buffer_data',
'pin_busy', 'pin_request', 'pin_reset', 'pin_sel',
'pin_busy', 'pin_request', 'pin_mem_sel', 'pin_ram_sel',
'_pin_sel_val',
'_read_green_times','_read_red_times',
'_elite_data_len', '_mem_header_len', '_mem_tailer_len', '_single_data_len',
@@ -57,8 +55,8 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
# memory control pin
self.pin_busy = OutputPin.get_used(P3Pin.MEM_BZY, True)
self.pin_request = OutputPin.get_used(P3Pin.MEM_REQ, True)
self.pin_reset = OutputPin.get_used(P3Pin.MEM_RST, True)
self.pin_sel: Optional[InputPin] = InputPin.get_used(P3Pin.MEM_SEL)
self.pin_mem_sel = OutputPin.get_used(P3Pin.MEM_RST, True)
self.pin_ram_sel: Optional[InputPin] = InputPin.get_used(P3Pin.MEM_SEL)
self._read_green_times = 0
self._read_red_times = 0
@@ -82,7 +80,7 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
def changed(self, flip=False) -> bool:
channel = self._selector.channel
old = self._pin_sel_val[channel]
value = bool(self.pin_sel)
value = bool(self.pin_ram_sel)
if flip:
self._pin_sel_val[channel] = value
@@ -96,21 +94,12 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
self._selector.close()
self._spi.close()
def flush(self):
self.pin_reset.output(False)
self.pin_reset.output(True)
def send_byte(self, data: bytes):
raise RuntimeError()
def recv_byte(self, size: int) -> Optional[bytes]:
raise RuntimeError()
# def flush_all(self):
# for _ in self._selector.foreach():
# self.pin_reset.output(False)
# self.pin_reset.output(True)
def foreach(self) -> Iterable[int]:
for channel in self._selector.foreach():
yield channel
@@ -124,7 +113,7 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
else:
return None
def request_data(self):
def mem_request(self):
self.pin_request.output(False)
sleep(0.001)
@@ -380,30 +369,15 @@ class MultiExtMemSpiInterface(LowLevelHardwareInterface):
self._spi.send_byte(tx)
def flush_all(self):
self.flush()
class ExtMemManager:
def __init__(self, ext_mem: MultiExtMemSpiInterface):
self._mem_sel = InputPin.get_used(P3Pin.MEM_SEL, pull_up_down=True)
self._mem_req = OutputPin.get_used(P3Pin.MEM_REQ, initial=True)
self._ext_mem = ext_mem
@property
def mem_sel(self) -> int:
return int(self._mem_sel)
@property
def mem_req(self) -> int:
return int(self._mem_req)
def mem_request(self):
self._mem_req.pulse()
sleep(_SLEEP_TIME_)
self._ext_mem.mem_request()
def get_ram_sel(self):
return int(self._ext_mem.pin_ram_sel)
def get_ext_mem_register(self) -> List[Tuple[Optional[int], Optional[int]]]:
ret = [(None, None) for _ in range(Selector.SIZE)]
@@ -413,14 +387,14 @@ class ExtMemManager:
self.mem_request()
sleep(0.01)
m1 = int(self._mem_sel)
m1 = self.get_ram_sel()
# print("m1 = ", m1)
self._ext_mem.write_register(DEFAULT_REGISTER_VALUE)
r[m1] = self._ext_mem.read_register()
self.mem_request()
sleep(0.01)
m2 = int(self._mem_sel)
m2 = self.get_ram_sel()
# print("m2 = ", m2)
self._ext_mem.write_register(DEFAULT_REGISTER_VALUE)
r[m2] = self._ext_mem.read_register()
@@ -449,33 +423,6 @@ class ExtMemManager:
return 0
def get_no_device_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):
ret.append(channel)
return ret
def get_memory_test_fail(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:
ret.append(channel)
return ret
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()
@@ -492,173 +439,3 @@ class ExtMemManager:
ret.append(channel)
return ret
def test_available_channel(self, result: List[Tuple[Optional[int], Optional[int]]] = None) -> List[int]:
if result is None:
result = self.get_ext_mem_register()
available_channel = []
for channel, result in enumerate(result):
r1, r2 = result
if self.is_no_device(result):
print('channel', channel, 'no device')
continue
test_pass = True
if r1 is None or r2 is None:
test_pass = False
print('channel', channel, 'mem_sel', 'x', 'not change')
if r1 is not None and r1 > 0 and r1 != DEFAULT_REGISTER_VALUE:
test_pass = False
print('channel', channel, 'mem_sel', 0, 'register', r1)
if r2 is not None and r2 > 0 and r2 != DEFAULT_REGISTER_VALUE:
test_pass = False
print('channel', channel, 'mem_sel', 1, 'register', r2)
if test_pass:
print('channel', channel, 'mem_sel', 0, 'register', r1)
print('channel', channel, 'mem_sel', 1, 'register', r2)
available_channel.append(channel)
return available_channel
def test_memory_read_write(self, channel: int, print_result=False) -> bool:
self._ext_mem.select = channel
addr = 4
size = 100
test_pass = True
while addr < MEM_SIZE:
d1 = [randint(10, 99) for _ in range(size)]
d2 = [randint(10, 99) for _ in range(size)]
self.mem_request()
m1 = int(self._mem_sel)
self._ext_mem.write_memory(addr, d1)
self.mem_request()
m2 = int(self._mem_sel)
self._ext_mem.write_memory(addr, d2)
self.mem_request()
m11 = int(self._mem_sel)
r1 = self._ext_mem.read_memory(addr, size)
if m1 != m11:
test_pass = False
if print_result:
print('channel', channel, 'mem_sel not change')
if d1 != r1:
test_pass = False
if print_result:
self.print_bytes_diff(d1, r1)
self.mem_request()
m22 = int(self._mem_sel)
r2 = self._ext_mem.read_memory(addr, size)
if m2 != m22:
test_pass = False
if print_result:
print('channel', channel, 'mem_sel not change')
if d2 != r2:
test_pass = False
if print_result:
self.print_bytes_diff(d2, r2)
addr += size
return test_pass
@staticmethod
def print_bytes_diff(_1: List[int], _2: List[int]):
s1 = ''
s2 = ''
l1 = len(_1)
l2 = len(_2)
for i in range(min(l1, l2)):
v1 = _1[i]
v2 = _2[i]
if v1 == v2:
s1 += '%02X ' % v1
s2 += '%02X ' % v2
else:
s1 += '%02X ' % v1
s2 += pc('%02X' % v2, RED) + ' '
if l1 == l2:
pass
elif l1 < l2:
for i in range(l1, l2):
s2 += pc('%02X' % _2[i], GREEN) + ' '
else:
for i in range(l2, l1):
s1 += pc('%02X' % _1[i], RED) + ' '
print(s1)
print(s2)
def hardware_test():
ext_mem = MultiExtMemSpiInterface(Selector.get(Selector.MEM_SELECTOR))
fake_mode = False
try:
ext_mem.reset()
except FileNotFoundError:
fake_mode = True
ret = {
'channel_count': Selector.SIZE,
'memory_size': MEM_SIZE,
'pin_mem_req': P3Pin.MEM_REQ,
'pin_mem_sel': P3Pin.MEM_SEL,
}
if not fake_mode:
tester = ExtMemManager(ext_mem)
result = tester.get_ext_mem_register()
no_device = tester.get_no_device_channel(result)
reg_fail = tester.get_memory_test_fail(result)
available = tester.get_available_channel(result)
channel = ['unknown' for _ in range(Selector.SIZE)]
for ch in range(len(channel)):
if ch in no_device:
channel[ch] = 'no_device'
elif ch in reg_fail:
channel[ch] = 'reg_fail'
elif not tester.test_memory_read_write(ch):
channel[ch] = 'mem_fail'
else:
channel[ch] = 'available'
ret['register'] = result
ret['available'] = available
ret['channel'] = channel
return ret
if __name__ == '__main__':
from pprint import pprint
pprint(hardware_test())
python/biopro/impl/selector.py
+8 -8
View File
@@ -117,14 +117,14 @@ class Selector:
self._p1.output(p[1])
self._p2.output(p[2])
if (value == 4 and self._last_sel != 6):
print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
elif (value == 5 and self._last_sel != 4):
print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
elif (value == 7 and self._last_sel != 5):
print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
elif (value == 6 and self._last_sel != 7):
print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
# if (value == 4 and self._last_sel != 6):
# print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
# elif (value == 5 and self._last_sel != 4):
# print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
# elif (value == 7 and self._last_sel != 5):
# print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
# elif (value == 6 and self._last_sel != 7):
# print('mem_sel is not use gray code, value:', value, ', last value:', self._last_sel)
self._last_sel = value
python/biopro/server/data.py
+3 -8
View File
@@ -131,9 +131,6 @@ class DataServer(SocketServer, DataAPI):
self._available_channel.clear()
if self._spi is not None:
self.log_verbose('spi flush')
self._spi.flush_all()
self.reset_available_channel()
LEAK.reset()
@@ -788,16 +785,14 @@ class DataServer(SocketServer, DataAPI):
if self._spi is not None:
self._spi.select = device
self._spi.flush()
return
def spi_reset_ram_header(self, device: int):
if self._spi is not None:
self._spi.select = device
self._spi.request_data()
self._spi.mem_request()
sleep(0.01)
self._spi.reset()
self._spi.flush()
return
def spi_send(self, device: int, address: int, data: bytes):
@@ -857,7 +852,7 @@ class DataServer(SocketServer, DataAPI):
# return None
# return sync.recv_memory()
else:
sync.request_data()
sync.mem_request()
return None
@@ -876,7 +871,7 @@ class DataServer(SocketServer, DataAPI):
return sync.recv_memory()
else:
# print('request_data___')
return sync.request_data()
return sync.mem_request()
class DataRuntime(metaclass=abc.ABCMeta):