# This file is part of the OpenMV project. # # Copyright (c) 2013-2020 Ibrahim Abdelkader # Copyright (c) 2013-2020 Kwabena W. Agyeman # # This work is licensed under the MIT license, see the file LICENSE for details. import gc import machine import omv import select import struct import time class rpc: _COMMAND_HEADER_PACKET_MAGIC = 0x1209 _COMMAND_DATA_PACKET_MAGIC = 0xABD1 _RESULT_HEADER_PACKET_MAGIC = 0x9021 _RESULT_DATA_PACKET_MAGIC = 0x1DBA @micropython.viper def __def_crc_16(self, data, size: int) -> int: # private d = ptr8(data) crc = 0xFFFF for i in range(size): crc ^= d[i] << 8 for j in range(8): crc = (crc << 1) ^ (0x1021 if crc & 0x8000 else 0) return crc & 0xFFFF @micropython.viper def __stm_crc_16(self, data, size: int) -> int: # private import stm ptr32(stm.CRC + stm.CRC_CR)[0] = (1 << 3) | 1 crc8 = ptr8(stm.CRC + stm.CRC_DR) d = ptr8(data) for i in range(size): crc8[0] = d[i] return ptr32(stm.CRC + stm.CRC_DR)[0] @micropython.viper def _zero(self, buff, size: int): # private d = ptr8(buff) for i in range(size): d[i] = 0 @micropython.viper def _same(self, data, size: int) -> bool: # private if not size: return False d = ptr8(data) old = d[0] for i in range(1, size): new = d[i] if new != old: return False old = new return True # djb2 algorithm; see http://www.cse.yorku.ca/~oz/hash.html @micropython.viper def _hash(self, data, size: int) -> uint: # private h = 5381 d = ptr8(data) for i in range(size): h = ((h << 5) + h) ^ d[i] return uint(h) def __init__(self): # private self.__crc_16 = self.__def_crc_16 if omv.board_type() == "H7": import stm stm.mem32[stm.RCC + stm.RCC_AHB4ENR] = stm.mem32[stm.RCC + stm.RCC_AHB4ENR] | (1 << 19) stm.mem32[stm.CRC + stm.CRC_POL] = 0x1021 self.__crc_16 = self.__stm_crc_16 elif omv.board_type() == "F7": import stm stm.mem32[stm.RCC + stm.RCC_AHB1ENR] = stm.mem32[stm.RCC + stm.RCC_AHB1ENR] | (1 << 12) stm.mem32[stm.CRC + stm.CRC_POL] = 0x1021 self.__crc_16 = self.__stm_crc_16 self._stream_writer_queue_depth_max = 255 def _get_packet_pre_alloc(self, payload_len=0): buff = bytearray(payload_len + 4) return (buff, memoryview(buff)[2:-2]) def _get_packet(self, magic_value, payload_buf_tuple, timeout): # private packet = self.get_bytes(payload_buf_tuple[0], timeout) if packet is not None: magic = packet[0] | (packet[1] << 8) crc = packet[-2] | (packet[-1] << 8) if magic == magic_value and crc == self.__crc_16(packet, len(packet) - 2): return payload_buf_tuple[1] return None def _set_packet(self, magic_value, payload=bytes()): # private new_payload = bytearray(len(payload) + 4) new_payload[:2] = struct.pack("> 1) ^ (0xB8 if tx_lfsr & 1 else 0x00) def stream_writer(self, call_back, write_timeout_ms=5000): # public packet = self._stream_get_bytes(bytearray(8), 1000) if packet is None: return magic = packet[0] | (packet[1] << 8) crc = packet[-2] | (packet[-1] << 8) if magic != 0xEDF6 and crc != self.__crc_16(packet, len(packet) - 2): return queue_depth = max( min(struct.unpack("> 1) ^ (0xB8 if rx_lfsr & 1 else 0x00) credits += 1 if credits > 0: data = call_back() try: self._stream_put_bytes( self._set_packet(0x542E, struct.pack("= self._get_short_timeout: return None self.__spi.init(pyb.SPI.SLAVE, polarity=self.__polarity, phase=self.__clk_phase) try: self.__spi.send_recv(buff, buff, timeout=timeout_ms) # SPI.recv() is broken. except OSError: buff = None self.__spi.deinit() return buff def put_bytes(self, data, timeout_ms): # protected import pyb start = pyb.millis() while self.__pin.value(): if pyb.elapsed_millis(start) >= self._put_short_timeout: return self.__spi.init(pyb.SPI.SLAVE, polarity=self.__polarity, phase=self.__clk_phase) try: self.__spi.send(data, timeout=timeout_ms) except OSError: pass self.__spi.deinit() class rpc_uart_master(rpc_master): def __init__(self, baudrate=9600, uart_port=3): # private self.__uart = machine.UART(uart_port, baudrate, timeout=2, timeout_char=2) rpc_master.__init__(self) def _flush(self): # protected self.__uart.read(self.__uart.any()) def get_bytes(self, buff, timeout_ms): # protected if self.__uart.readinto(buff) is None: time.sleep_ms(self._get_short_timeout) return None return buff def put_bytes(self, data, timeout_ms): # protected self.__uart.write(data) def _stream_get_bytes(self, buff, timeout_ms): # protected p = select.poll() p.register(self.__uart, select.POLLIN) p.poll(1000) return self.get_bytes(buff, timeout_ms) class rpc_uart_slave(rpc_slave): def __init__(self, baudrate=9600, uart_port=3): # private self.__uart = machine.UART(uart_port, baudrate, timeout=2, timeout_char=2) rpc_slave.__init__(self) def _flush(self): # protected self.__uart.read(self.__uart.any()) def get_bytes(self, buff, timeout_ms): # protected if self.__uart.readinto(buff) is None: return None return buff def put_bytes(self, data, timeout_ms): # protected self.__uart.write(data) def _stream_get_bytes(self, buff, timeout_ms): # protected p = select.poll() p.register(self.__uart, select.POLLIN) p.poll(1000) return self.get_bytes(buff, timeout_ms) class rpc_usb_vcp_master(rpc_master): def __init__(self): # private import pyb self.__usb_vcp = pyb.USB_VCP() if self.__usb_vcp.debug_mode_enabled(): raise OSError("You cannot use the USB VCP while the IDE is connected!") self.__usb_vcp.setinterrupt(-1) rpc_master.__init__(self) def _flush(self): # protected self.__usb_vcp.read() def get_bytes(self, buff, timeout_ms): # protected if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): return None return buff def put_bytes(self, data, timeout_ms): # protected self.__usb_vcp.send(data, timeout=timeout_ms) class rpc_usb_vcp_slave(rpc_slave): def __init__(self): # private import pyb self.__usb_vcp = pyb.USB_VCP() if self.__usb_vcp.debug_mode_enabled(): raise OSError("You cannot use the USB VCP while the IDE is connected!") self.__usb_vcp.setinterrupt(-1) rpc_slave.__init__(self) def _flush(self): # protected self.__usb_vcp.read() def get_bytes(self, buff, timeout_ms): # protected if self.__usb_vcp.recv(buff, timeout=timeout_ms) != len(buff): return None return buff def put_bytes(self, data, timeout_ms): # protected self.__usb_vcp.send(data, timeout=timeout_ms)