From d1c4b0a29fd191dfb216ffe84c22bce226d3fce3 Mon Sep 17 00:00:00 2001 From: Roy Date: Thu, 10 Feb 2022 17:32:36 +0800 Subject: [PATCH] [update] merge Spi & mitiSpi class --- python/biopro/impl/ext_mem.py | 134 +++++++++++++--------------------- python/biopro/server/data.py | 4 +- 2 files changed, 51 insertions(+), 87 deletions(-) diff --git a/python/biopro/impl/ext_mem.py b/python/biopro/impl/ext_mem.py index 374eeaa..9850670 100644 --- a/python/biopro/impl/ext_mem.py +++ b/python/biopro/impl/ext_mem.py @@ -25,20 +25,21 @@ _SLEEP_TIME_ = 0.001 def zero_buffer(size: int) -> List[int]: return [0] * size - -class ExtMemSpiInterface(LowLevelHardwareInterface): +class MultiExtMemSpiInterface(LowLevelHardwareInterface): MEM_INS_MARKED = [MEM_INS_WRITE, 0, 2, 1, 1] MEM_INS_RESET = [MEM_INS_WRITE, 0, 2, 1, 1, 0, 0xFF] - __slots__ = ('_spi', '_tx_buffer', '_tx_buffer_header', '_tx_buffer_data', + __slots__ = ('_selector', '_wait_for_first_data', '_spi', '_tx_buffer', '_tx_buffer_header', '_tx_buffer_data', 'pin_busy', 'pin_request', 'pin_reset', 'pin_sel', '_pin_sel_val', '_read_green_times','_read_red_times', '_elite_data_len', '_mem_header_len', '_mem_tailer_len', '_single_data_len', '_head_wrong_cnt') - def __init__(self, device: Tuple[int, int] = None): - super().__init__() + def __init__(self, + select: Selector, + device: Tuple[int, int] = None): + self._spi = HardwareImplSpiInterface(device, spi_speed=12_000_000 # XXX temp parameter ) @@ -59,46 +60,70 @@ class ExtMemSpiInterface(LowLevelHardwareInterface): self.pin_reset = OutputPin.get_used(P3Pin.MEM_RST, True) self.pin_sel: Optional[InputPin] = InputPin.get_used(P3Pin.MEM_SEL) - self._pin_sel_val = False - 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)] + + @property + def select(self) -> int: + return self._selector.channel + + @select.setter + def select(self, value: int): + self._selector.select(value) + + def changed(self, flip=False) -> bool: + channel = self._selector.channel + old = self._pin_sel_val[channel] + value = bool(self.pin_sel) + + if flip: + self._pin_sel_val[channel] = value + + return value != old + def reset(self): self._spi.reset() def close(self): + self._selector.close() self._spi.close() def flush(self): self.pin_reset.output(False) self.pin_reset.output(True) - @property - def select(self) -> int: - return 0 - - @select.setter - def select(self, value: int): - pass - - def changed(self, flip=False) -> bool: - old = self._pin_sel_val - value = self.pin_sel.input() - - if flip: - self._pin_sel_val = value - - return value != old - 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 + + 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 request_data(self): self.pin_request.output(False) sleep(0.001) @@ -162,7 +187,6 @@ class ExtMemSpiInterface(LowLevelHardwareInterface): address = 0 # first read to get data length - tx = self._tx_buffer tx_h = self._tx_buffer_header tx_h[0] = MEM_INS_READ @@ -359,69 +383,11 @@ class ExtMemSpiInterface(LowLevelHardwareInterface): def flush_all(self): self.flush() - def foreach(self) -> Iterable[int]: - yield 0 -class MultiExtMemSpiInterface(ExtMemSpiInterface): - __slots__ = ('_selector', '_wait_for_first_data') - - def __init__(self, - select: Selector, - device: Tuple[int, int] = None): - super().__init__(device) - - self._selector = select - - self._pin_sel_val = [False for _ in range(Selector.SIZE)] - self._wait_for_first_data = [True for _ in range(Selector.SIZE)] - - @property - def select(self) -> int: - return self._selector.channel - - @select.setter - def select(self, value: int): - self._selector.select(value) - - def changed(self, flip=False) -> bool: - channel = self._selector.channel - old = self._pin_sel_val[channel] - value = bool(self.pin_sel) - - if flip: - self._pin_sel_val[channel] = value - - return value != old - - def reset(self): - self._spi.reset() - - def close(self): - self._selector.close() - super().close() - - 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 - - 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 - class ExtMemManager: - def __init__(self, ext_mem: ExtMemSpiInterface): + 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) diff --git a/python/biopro/server/data.py b/python/biopro/server/data.py index 356f309..dc735a5 100644 --- a/python/biopro/server/data.py +++ b/python/biopro/server/data.py @@ -10,7 +10,7 @@ import base64 from biopro.data import * from biopro.devlib.data import * from biopro.generator import GeneratorOptions -from biopro.impl.ext_mem import ExtMemSpiInterface, MultiExtMemSpiInterface, ExtMemManager +from biopro.impl.ext_mem import MultiExtMemSpiInterface, ExtMemManager from biopro.impl.selector import Selector from biopro.recording import RecordingMetaFile, RecordingFileWriter from biopro.util.address import EMPTY_ADDRESS, address_str @@ -224,8 +224,6 @@ class DataServer(SocketServer, DataAPI): return elif spi_mode == self.MODE_DISABLE: spi = None - elif spi_mode == self.MODE_SINGLE: - spi = ExtMemSpiInterface() elif spi_mode == self.MODE_SELECTOR: spi = MultiExtMemSpiInterface(Selector.get(Selector.MEM_SELECTOR)) else: