Merge pull request #777 from kwagyeman/kwabena/interface_library

New OpenMV Cam RPC Interface Library
This commit is contained in:
Ibrahim Abd Elkader 2020-05-04 21:26:18 +02:00 committed by GitHub
commit c1d31ebcc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 2870 additions and 5 deletions

View File

@ -1,13 +1,35 @@
### OpenMV (Open-Source Machine Vision)
## OpenMV (Open-Source Machine Vision)
<p align="center">
<img src="https://raw.githubusercontent.com/openmv/openmv-media/master/boards/openmv-cam/v3/web-new-cam-v3-angle.jpg" width="320" height="320" border="10">
<img style="border: 10px solid white;" src="https://raw.githubusercontent.com/openmv/openmv-media/master/boards/openmv-cam/v3/web-new-cam-v3-angle.jpg" width="320" height="320">
</p>
The OpenMV project aims at making machine vision more accessible to beginners by developing a user-friendly, open-source, low-cost machine vision platform.
The OpenMV project aims at making machine vision more accessible to beginners by developing a user-friendly, open-source, low-cost machine vision platform.
OpenMV cameras are programmable in Python3 and come with an extensive set of image processing functions such as face detection, keypoints descriptors, color tracking, QR and Bar codes decoding, AprilTags, GIF and MJPEG recording and more. Additionally, OpenMV includes a cross-platform IDE (based on Qt Creator) designed specifically to support programmable cameras. The IDE allows viewing the camera's frame buffer, accessing sensor controls, uploading scripts to the camera via serial over USB (or WiFi/BLE if available) and includes a set of image processing tools to generate tags, thresholds, keypoints etc...
OpenMV cameras are programmable in Python3 and come with an extensive set of image processing functions such as face detection, keypoints descriptors, color tracking, QR and Bar code decoding, AprilTags, GIF and MJPEG recording, and more. Additionally, the OpenMV Cam comes with a cross-platform IDE (based on Qt Creator) designed specifically to support programmable cameras. The IDE allows viewing the camera's frame buffer, accessing sensor controls, uploading scripts to the camera via serial over USB (or WiFi/BLE if available) and includes a set of image processing tools to generate tags, thresholds, keypoints, and etc...
The first generation of OpenMV cameras is based on STM32F ARM Cortex-M Digital Signal Controllers (DSCs) and Omnivision sensors. The board has built-in RGB and IR LEDs, USB FS for programming and video streaming, uSD socket and I/O headers breaking out PWM, UARTs, SPI and I2C. Additionally, OpenMV supports extension modules (shields) using the I/O headers such as WiFi, BLE, Thermal (FIR) and LCD shields.
The first generation of OpenMV cameras is based on STM32 ARM Cortex-M Digital Signal Processors (DSPs) and OmniVision sensors. The boards have built-in RGB and IR LEDs, USB FS support for programming and video streaming, a uSD socket, and I/O headers breaking out PWM, UARTs, SPI, I2C, CAN, and more. Additionally, the OpenMV Cam supports extension modules (shields) using the I/O headers for adding a WiFi adapter, a LCD Display, a Thermal Vision Sensor, a Motor Driver, and more.
The OpenMV project was successfully funded via Kickstarter back in 2015 and has come a long way since then. For more information, please visit [https://openmv.io](https://openmv.io)
## Interface Library
The OpenMV Cam comes built-in with an RPC (Remote Python/Procedure Call) library which makes it easy to connect the OpenMV Cam to your computer, a SBC (single board computer) like the RaspberryPi or Beaglebone, or a microcontroller like the Arduino or ESP8266/32. The RPC Interface Library works over:
* Async Serial (UART) - at up **7.5 Mb/s** on the OpenMV Cam H7.
* I2C Bus - at up to **1 Mb/s** on the OpenMV Cam H7.
* Using 1K pull up resistors.
* SPI Bus - at up to **20 Mb/s** on the OpenMV Cam H7.
* Up to **80 Mb/s** or **40 Mb/s** is achievable with short enough wires.
* CAN Bus - at up to **1 Mb/s** on the OpenMV Cam H7.
* USB Virtual COM Port (VCP) - at up to **12 Mb/s** on the OpenMV Cam M4/M7/H7.
* WiFi using the [WiFi Shield](https://openmv.io/collections/shields/products/wifi-shield-1) - at up to **12 Mb/s** on the OpenMV Cam M4/M7/H7.
With the RPC Library you can easily get image processing results, stream RAW or JPG image data, or have the OpenMV Cam control another Microcontroller for lower-level hardware control like driving motors.
You can find examples that run on the OpenMV Cam under `File->Examples->Remote Control` in OpenMV IDE and online [here](scripts/examples/34-Remote-Control). Finally, OpenMV provides the following libraries for interfacing your OpenMV Cam to other systems below:
* [Generic Python Interface Library for USB and WiFi Comms](tools/rpc/README.md)
* Provides Python code for connecting your OpenMV Cam to a Windows, Mac, or Linux computer (or RaspberryPi/Beaglebone, etc.) with python programmatically over USB VCP or Ethernet/WiFi (i.e. with sockets).
* Arduino Interface Library for I2C, SPI, UART Comms - comming soon!
* RaspberryPi Interface Library for I2C, SPI, UART Comms - comming soon!

View File

@ -0,0 +1,87 @@
# Image Transfer - As The Remote Device
#
# This script is meant to talk to the "image_transfer_jpg_as_the_controller_device.py" on your computer.
#
# This script shows off how to transfer the frame buffer to your computer as a jpeg image.
import image, network, omv, rpc, sensor, struct
sensor.reset()
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
sensor.skip_frames(time = 2000)
# Turn off the frame buffer connection to the IDE from the OpenMV Cam side.
#
# This needs to be done when manually compressing jpeg images at higher quality
# so that the OpenMV Cam does not try to stream them to the IDE using a fall back
# mechanism if the JPEG image is too large to fit in the IDE JPEG frame buffer on the OpenMV Cam.
omv.disable_fb(True)
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to be controlled over USB or WIFI.
################################################################
# Choose the interface you wish to control your OpenMV Cam over.
################################################################
# Uncomment the below line to setup your OpenMV Cam for control over a USB VCP.
#
interface = rpc.rpc_usb_vcp_slave()
# Uncomment the below line to setup your OpenMV Cam for control over WiFi.
#
# * ssid - WiFi network to connect to.
# * ssid_key - WiFi network password.
# * ssid_security - WiFi security.
# * port - Port to route traffic to.
# * mode - Regular or access-point mode.
# * static_ip - If not None then a tuple of the (IP Address, Subnet Mask, Gateway, DNS Address)
#
# interface = rpc.rpc_wifi_slave(ssid="",
# ssid_key="",
# ssid_security=network.WINC.WPA_PSK,
# port=0x1DBA,
# mode=network.WINC.MODE_STA,
# static_ip=None)
################################################################
# Call Backs
################################################################
# When called sets the pixformat and framesize, takes a snapshot
# and then returns the frame buffer jpg size to store the image in.
#
# data is a pixformat string and framesize string.
def jpeg_image_snapshot(data):
pixformat, framesize = bytes(data).decode().split(",")
sensor.set_pixformat(eval(pixformat))
sensor.set_framesize(eval(framesize))
img = sensor.snapshot().compress(quality=90)
return struct.pack("<I", img.size())
def jpeg_image_read_cb():
interface.put_bytes(sensor.get_fb().bytearray(), 5000) # timeout
# Read data from the frame buffer given a offset and size.
# If data is empty then a transfer is scheduled after the RPC call finishes.
#
# data is a 4 byte size and 4 byte offset.
def jpeg_image_read(data):
if not len(data):
interface.schedule_callback(jpeg_image_read_cb)
return bytes()
else:
offset, size = struct.unpack("<II", data)
return memoryview(sensor.get_fb().bytearray())[offset:offset+size]
# Register call backs.
interface.register_callback(jpeg_image_snapshot)
interface.register_callback(jpeg_image_read)
# Once all call backs have been registered we can start
# processing remote events. interface.loop() does not return.
interface.loop()

View File

@ -0,0 +1,79 @@
# Image Transfer - As The Remote Device
#
# This script is meant to talk to the "image_transfer_jpg_streaming_as_the_controller_device.py" on your computer.
#
# This script shows off how to transfer the frame buffer to your computer as a jpeg image.
import network, omv, rpc, sensor
sensor.reset()
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
sensor.skip_frames(time = 2000)
# Turn off the frame buffer connection to the IDE from the OpenMV Cam side.
#
# This needs to be done when manually compressing jpeg images at higher quality
# so that the OpenMV Cam does not try to stream them to the IDE using a fall back
# mechanism if the JPEG image is too large to fit in the IDE JPEG frame buffer on the OpenMV Cam.
omv.disable_fb(True)
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to be controlled over USB or WIFI.
################################################################
# Choose the interface you wish to control your OpenMV Cam over.
################################################################
# Uncomment the below line to setup your OpenMV Cam for control over a USB VCP.
#
interface = rpc.rpc_usb_vcp_slave()
# Uncomment the below line to setup your OpenMV Cam for control over WiFi.
#
# * ssid - WiFi network to connect to.
# * ssid_key - WiFi network password.
# * ssid_security - WiFi security.
# * port - Port to route traffic to.
# * mode - Regular or access-point mode.
# * static_ip - If not None then a tuple of the (IP Address, Subnet Mask, Gateway, DNS Address)
#
# interface = rpc.rpc_wifi_slave(ssid="",
# ssid_key="",
# ssid_security=network.WINC.WPA_PSK,
# port=0x1DBA,
# mode=network.WINC.MODE_STA,
# static_ip=None)
################################################################
# Call Backs
################################################################
# This is called repeatedly by interface.stream_writer().
def stream_generator_cb():
return sensor.snapshot().compress(quality=90).bytearray()
# Transmits a stream of bytes()'s generated by stream_generator_cb to the master device.
def jpeg_image_stream_cb():
interface.stream_writer(stream_generator_cb)
# When called sets the pixformat and framesize, and then schedules
# frame streaming to start after the RPC call finishes.
#
# data is a pixformat string and framesize string.
def jpeg_image_stream(data):
pixformat, framesize = bytes(data).decode().split(",")
sensor.set_pixformat(eval(pixformat))
sensor.set_framesize(eval(framesize))
interface.schedule_callback(jpeg_image_stream_cb)
return bytes()
# Register call backs.
interface.register_callback(jpeg_image_stream)
# Once all call backs have been registered we can start
# processing remote events. interface.loop() does not return.
interface.loop()

View File

@ -0,0 +1,129 @@
# Image Transfer - As The Controller Device
#
# This script is made to pair with another OpenMV Cam running "image_transfer_raw_as_the_remote_device.py"
#
# This script shows off how to transfer the frame buffer from one OpenMV Cam to another.
import image, network, omv, rpc, sensor, struct, time
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to control over CAN, I2C, SPI, UART, or WIFI.
##############################################################
# Choose the interface you wish to control an OpenMV Cam over.
##############################################################
# Uncomment the below line to setup your OpenMV Cam for controlling over CAN.
#
# * message_id - CAN message to use for data transport on the can bus (11-bit).
# * bit_rate - CAN bit rate.
# * sampling_point - Tseg1/Tseg2 ratio. Typically 75%. (50.0, 62.5, 75.0, 87.5, etc.)
#
# NOTE: Master and slave message ids and can bit rates must match. Connect master can high to slave
# can high and master can low to slave can lo. The can bus must be terminated with 120 ohms.
#
# interface = rpc.rpc_can_master(message_id=0x7FF, bit_rate=1000000, sampling_point=75.0)
# Uncomment the below line to setup your OpenMV Cam for controlling over I2C.
#
# * slave_addr - I2C address.
# * rate - I2C Bus Clock Frequency.
#
# NOTE: Master and slave addresses must match. Connect master scl to slave scl and master sda
# to slave sda. You must use external pull ups. Finally, both devices must share a ground.
#
# interface = rpc.rpc_i2c_master(slave_addr=0x12, rate=1000000)
# Uncomment the below line to setup your OpenMV Cam for controlling over SPI.
#
# * cs_pin - Slave Select Pin.
# * freq - SPI Bus Clock Frequency
# * clk_polarity - Idle clock level (0 or 1).
# * clk_phase - Sample data on the first (0) or second edge (1) of the clock.
#
# NOTE: Master and slave settings much match. Connect CS, SCLK, MOSI, MISO to CS, SCLK, MOSI, MISO.
# Finally, both devices must share a common ground.
#
interface = rpc.rpc_spi_master(cs_pin="P3", freq=20000000, clk_polarity=1, clk_phase=0)
# Uncomment the below line to setup your OpenMV Cam for controlling over UART.
#
# * baudrate - Serial Baudrate.
#
# NOTE: Master and slave baud rates must match. Connect master tx to slave rx and master rx to
# slave tx. Finally, both devices must share a common ground.
#
# interface = rpc.rpc_uart_master(baudrate=7500000)
##############################################################
# Call Back Handlers
##############################################################
def get_frame_buffer_call_back(pixformat, framesize, cutthrough, silent):
if not silent: print("Getting Remote Frame...")
result = interface.call("raw_image_snapshot", struct.pack("<II", pixformat, framesize))
if result is not None:
w, h, pixformat, size = struct.unpack("<IIII", result)
img = image.Image(w, h, pixformat, copy_to_fb=True) # Alloc cleared frame buffer.
if cutthrough:
# Fast cutthrough data transfer with no error checking.
# Before starting the cut through data transfer we need to sync both the master and the
# slave device. On return both devices are in sync.
result = interface.call("raw_image_read")
if result is not None:
# GET BYTES NEEDS TO EXECUTE NEXT IMMEDIATELY WITH LITTLE DELAY NEXT.
# Read all the image data in one very large transfer.
interface.get_bytes(img.bytearray(), 5000) # timeout
else:
# Slower data transfer with error checking.
# Transfer 32/8 KB chunks.
chunk_size = (1 << 15) if omv.board_type() == "H7" else (1 << 13)
if not silent: print("Reading %d bytes..." % size)
for i in range(0, size, chunk_size):
ok = False
for j in range(3): # Try up to 3 times.
result = interface.call("raw_image_read", struct.pack("<II", i, chunk_size))
if result is not None:
img.bytearray()[i:i+chunk_size] = result # Write the image data.
if not silent: print("%.2f%%" % ((i * 100) / size))
ok = True
break
if not silent: print("Retrying... %d/2" % (j + 1))
if not ok:
if not silent: print("Error!")
return None
return img
else:
if not silent: print("Failed to get Remote Frame!")
return None
clock = time.clock()
while(True):
clock.tick()
# You may change the pixformat and the framesize of the image transfered from the remote device
# by modifying the below arguments.
#
# When cutthrough is False the image will be transferred through the RPC library with CRC and
# retry protection on all data moved. For faster data transfer set cutthrough to True so that
# get_bytes() and put_bytes() are called after an RPC call completes to transfer data
# more quicly from one image buffer to another. Note: This works because once an RPC call
# completes successfully both the master and slave devices are synchronized completely.
#
img = get_frame_buffer_call_back(sensor.RGB565, sensor.QQVGA, cutthrough=True, silent=True)
if img is not None:
pass # You can process the image here.
print(clock.fps())

View File

@ -0,0 +1,99 @@
# Image Transfer - As The Remote Device
#
# This script is made to pair with another OpenMV Cam running "image_transfer_raw_as_the_controller_device.py"
#
# This script shows off how to transfer the frame buffer from one OpenMV Cam to another.
import image, network, rpc, sensor, struct
sensor.reset()
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
sensor.skip_frames(time = 2000)
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to be controlled over CAN, I2C, SPI, UART, or WIFI.
################################################################
# Choose the interface you wish to control your OpenMV Cam over.
################################################################
# Uncomment the below line to setup your OpenMV Cam for control over CAN.
#
# * message_id - CAN message to use for data transport on the can bus (11-bit).
# * bit_rate - CAN bit rate.
# * sampling_point - Tseg1/Tseg2 ratio. Typically 75%. (50.0, 62.5, 75.0, 87.5, etc.)
#
# NOTE: Master and slave message ids and can bit rates must match. Connect master can high to slave
# can high and master can low to slave can lo. The can bus must be terminated with 120 ohms.
#
# interface = rpc.rpc_can_slave(message_id=0x7FF, bit_rate=1000000, sampling_point=75.0)
# Uncomment the below line to setup your OpenMV Cam for control over I2C.
#
# * slave_addr - I2C address.
#
# NOTE: Master and slave addresses must match. Connect master scl to slave scl and master sda
# to slave sda. You must use external pull ups. Finally, both devices must share a ground.
#
# interface = rpc.rpc_i2c_slave(slave_addr=0x12)
# Uncomment the below line to setup your OpenMV Cam for control over SPI.
#
# * cs_pin - Slave Select Pin.
# * clk_polarity - Idle clock level (0 or 1).
# * clk_phase - Sample data on the first (0) or second edge (1) of the clock.
#
# NOTE: Master and slave settings much match. Connect CS, SCLK, MOSI, MISO to CS, SCLK, MOSI, MISO.
# Finally, both devices must share a common ground.
#
interface = rpc.rpc_spi_slave(cs_pin="P3", clk_polarity=1, clk_phase=0)
# Uncomment the below line to setup your OpenMV Cam for control over UART.
#
# * baudrate - Serial Baudrate.
#
# NOTE: Master and slave baud rates must match. Connect master tx to slave rx and master rx to
# slave tx. Finally, both devices must share a common ground.
#
# interface = rpc.rpc_uart_slave(baudrate=7500000)
################################################################
# Call Backs
################################################################
# When called sets the pixformat and framesize, takes a snapshot
# and then returns the frame buffer shape to store the image in.
#
# data is a 4 byte pixformat and 4 byte framesize.
def raw_image_snapshot(data):
pixformat, framesize = struct.unpack("<II", data)
sensor.set_pixformat(pixformat)
sensor.set_framesize(framesize)
img = sensor.snapshot()
return struct.pack("<IIII", sensor.width(), sensor.height(), sensor.get_pixformat(), img.size())
def raw_image_read_cb():
interface.put_bytes(sensor.get_fb().bytearray(), 5000) # timeout
# Read data from the frame buffer given a offset and size.
# If data is empty then a transfer is scheduled after the RPC call finishes.
#
# data is a 4 byte size and 4 byte offset.
def raw_image_read(data):
if not len(data):
interface.schedule_callback(raw_image_read_cb)
return bytes()
else:
offset, size = struct.unpack("<II", data)
return memoryview(sensor.get_fb().bytearray())[offset:offset+size]
# Register call backs.
interface.register_callback(raw_image_snapshot)
interface.register_callback(raw_image_read)
# Once all call backs have been registered we can start
# processing remote events. interface.loop() does not return.
interface.loop()

View File

@ -0,0 +1,157 @@
# Remote Control - As The Controller Device
#
# This script configures your OpenMV Cam to remotely control another OpenMV Cam using the RPC
# library. This script can be run by any micropython board implementing the pyb module to
# remotely control an OpenMV Cam.
#
# This script is designed to pair with "popular_features_as_the_remote_device.py".
import json, rpc, struct
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to control over CAN, I2C, SPI, or UART.
##############################################################
# Choose the interface you wish to control an OpenMV Cam over.
##############################################################
# Uncomment the below line to setup your OpenMV Cam for controlling over CAN.
#
# * message_id - CAN message to use for data transport on the can bus (11-bit).
# * bit_rate - CAN bit rate.
# * sampling_point - Tseg1/Tseg2 ratio. Typically 75%. (50.0, 62.5, 75, 87.5, etc.)
#
# NOTE: Master and slave message ids and can bit rates must match. Connect master can high to slave
# can high and master can low to slave can lo. The can bus must be terminated with 120 ohms.
#
# interface = rpc.rpc_can_master(message_id=0x7FF, bit_rate=250000, sampling_point=75)
# Uncomment the below line to setup your OpenMV Cam for controlling over I2C.
#
# * slave_addr - I2C address.
# * rate - I2C Bus Clock Frequency.
#
# NOTE: Master and slave addresses must match. Connect master scl to slave scl and master sda
# to slave sda. You must use external pull ups. Finally, both devices must share a ground.
#
# interface = rpc.rpc_i2c_master(slave_addr=0x12, rate=100000)
# Uncomment the below line to setup your OpenMV Cam for controlling over SPI.
#
# * cs_pin - Slave Select Pin.
# * freq - SPI Bus Clock Frequency.
# * clk_polarity - Idle clock level (0 or 1).
# * clk_phase - Sample data on the first (0) or second edge (1) of the clock.
#
# NOTE: Master and slave settings much match. Connect CS, SCLK, MOSI, MISO to CS, SCLK, MOSI, MISO.
# Finally, both devices must share a common ground.
#
# interface = rpc.rpc_spi_master(cs_pin="P3", freq=10000000, clk_polarity=1, clk_phase=0)
# Uncomment the below line to setup your OpenMV Cam for controlling over UART.
#
# * baudrate - Serial Baudrate.
#
# NOTE: Master and slave baud rates must match. Connect master tx to slave rx and master rx to
# slave tx. Finally, both devices must share a common ground.
#
interface = rpc.rpc_uart_master(baudrate=115200)
##############################################################
# Call Back Handlers
##############################################################
def exe_face_detection():
result = interface.call("face_detection")
if result is not None and len(result):
print("Largest Face Detected [x=%d, y=%d, w=%d, h=%d]" % struct.unpack("<HHHH", result))
def exe_person_detection():
result = interface.call("person_detection")
if result is not None:
print(bytes(result).decode())
def exe_qrcode_detection():
result = interface.call("qrcode_detection")
if result is not None and len(result):
print(bytes(result).decode())
def exe_all_qrcode_detection():
result = interface.call("all_qrcode_detection")
if result is not None and len(result):
print("QR Codes Detected:")
for obj in json.loads(result):
print(obj)
def exe_apriltag_detection():
result = interface.call("apriltag_detection")
if result is not None and len(result):
print("Largest Tag Detected [cx=%d, cy=%d, id=%d, rot=%d]" % struct.unpack("<HHHH",result))
def exe_all_apriltag_detection():
result = interface.call("all_apriltag_detection")
if result is not None and len(result):
print("Tags Detected:")
for obj in json.loads(result):
print(obj)
def exe_datamatrix_detection():
result = interface.call("datamatrix_detection")
if result is not None and len(result):
print(bytes(result).decode())
def exe_all_datamatrix_detection():
result = interface.call("all_datamatrix_detection")
if result is not None and len(result):
print("Data Matrices Detected:")
for obj in json.loads(result):
print(obj)
def exe_barcode_detection():
result = interface.call("barcode_detection")
if result is not None and len(result):
print(bytes(result).decode())
def exe_all_barcode_detection():
result = interface.call("all_barcode_detection")
if result is not None and len(result):
print("Bar Codes Detected:")
for obj in json.loads(result):
print(obj)
def exe_color_detection():
thresholds = (30, 100, 15, 127, 15, 127) # generic red thresholds
# thresholds = (30, 100, -64, -8, -32, 32) # generic green thresholds
# thresholds = (0, 30, 0, 64, -128, 0) # generic blue thresholds
result = interface.call("color_detection", struct.pack("<bbbbbb", *thresholds))
if result is not None and len(result):
print("Largest Color Detected [cx=%d, cy=%d]" % struct.unpack("<HH", result))
number = 0
def exe_jpeg_snapshot():
global number
result = interface.call("jpeg_snapshot")
if result is not None:
name = "snapshot-%05d.jpg" % number
print("Writing jpeg %s..." % name)
with open(name, "wb") as snap:
snap.write(result)
number += 1
# Execute remote functions in a loop. Please choose and uncomment one remote function below.
# Executing multiple at a time may run slowly if the camera needs to change camera modes
# per execution.
while(True):
exe_face_detection() # Face should be about 2ft away.
# exe_person_detection()
# exe_qrcode_detection() # Place the QRCode about 2ft away.
# exe_all_qrcode_detection() # Place the QRCode about 2ft away.
# exe_apriltag_detection()
# exe_all_apriltag_detection()
# exe_datamatrix_detection() # Place the Datamatrix about 2ft away.
# exe_all_datamatrix_detection() # Place the Datamatrix about 2ft away.
# exe_barcode_detection() # Place the Barcode about 2ft away.
# exe_all_barcode_detection() # Place the Barcode about 2ft away.
# exe_color_detection()
# exe_jpeg_snapshot()

View File

@ -0,0 +1,269 @@
# Remote Control - As The Remote Device
#
# This script configures your OpenMV Cam as a co-processor that can be remotely controlled by
# another microcontroller or computer such as an Arduino, ESP8266/ESP32, RaspberryPi, and
# even another OpenMV Cam.
#
# This script is designed to pair with "popular_features_as_the_controller_device.py".
import image, network, math, rpc, sensor, struct, tf
sensor.reset()
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
sensor.skip_frames(time = 2000)
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to be controlled over CAN, I2C, SPI, UART, USB VCP, or WiFi.
################################################################
# Choose the interface you wish to control your OpenMV Cam over.
################################################################
# Uncomment the below line to setup your OpenMV Cam for control over CAN.
#
# * message_id - CAN message to use for data transport on the can bus (11-bit).
# * bit_rate - CAN bit rate.
# * sampling_point - Tseg1/Tseg2 ratio. Typically 75%. (50.0, 62.5, 75, 87.5, etc.)
#
# NOTE: Master and slave message ids and can bit rates must match. Connect master can high to slave
# can high and master can low to slave can lo. The can bus must be terminated with 120 ohms.
#
# interface = rpc.rpc_can_slave(message_id=0x7FF, bit_rate=250000, sampling_point=75)
# Uncomment the below line to setup your OpenMV Cam for control over I2C.
#
# * slave_addr - I2C address.
#
# NOTE: Master and slave addresses must match. Connect master scl to slave scl and master sda
# to slave sda. You must use external pull ups. Finally, both devices must share a ground.
#
# interface = rpc.rpc_i2c_slave(slave_addr=0x12)
# Uncomment the below line to setup your OpenMV Cam for control over SPI.
#
# * cs_pin - Slave Select Pin.
# * clk_polarity - Idle clock level (0 or 1).
# * clk_phase - Sample data on the first (0) or second edge (1) of the clock.
#
# NOTE: Master and slave settings much match. Connect CS, SCLK, MOSI, MISO to CS, SCLK, MOSI, MISO.
# Finally, both devices must share a common ground.
#
# interface = rpc.rpc_spi_slave(cs_pin="P3", clk_polarity=1, clk_phase=0)
# Uncomment the below line to setup your OpenMV Cam for control over UART.
#
# * baudrate - Serial Baudrate.
#
# NOTE: Master and slave baud rates must match. Connect master tx to slave rx and master rx to
# slave tx. Finally, both devices must share a common ground.
#
interface = rpc.rpc_uart_slave(baudrate=115200)
# Uncomment the below line to setup your OpenMV Cam for control over a USB VCP.
#
# interface = rpc.rpc_usb_vcp_slave()
# Uncomment the below line to setup your OpenMV Cam for control over WiFi.
#
# * ssid - WiFi network to connect to.
# * ssid_key - WiFi network password.
# * ssid_security - WiFi security.
# * port - Port to route traffic to.
# * mode - Regular or access-point mode.
# * static_ip - If not None then a tuple of the (IP Address, Subnet Mask, Gateway, DNS Address)
#
# interface = rpc.rpc_wifi_slave(ssid="",
# ssid_key="",
# ssid_security=network.WINC.WPA_PSK,
# port=0x1DBA,
# mode=network.WINC.MODE_STA,
# static_ip=None)
################################################################
# Call Backs
################################################################
# Helper methods used by the call backs below.
def draw_detections(img, dects):
for d in dects:
c = d.corners()
l = len(c)
for i in range(l): img.draw_line(c[(i+0)%l] + c[(i+1)%l], color = (0, 255, 0))
img.draw_rectangle(d.rect(), color = (255, 0, 0))
# Remote control works via call back methods that the controller
# device calls via the rpc module on this device. Call backs
# are functions which take a bytes() object as their argument
# and return a bytes() object as their result. The rpc module
# takes care of moving the bytes() objects across the link.
# bytes() may be the micropython int max in size.
# When called returns x, y, w, and h of the largest face within view.
#
# data is unused
def face_detection(data):
sensor.set_pixformat(sensor.GRAYSCALE)
sensor.set_framesize(sensor.QVGA)
faces = sensor.snapshot().gamma_corr(contrast=1.5).find_features(image.HaarCascade("frontalface"))
if not faces: return bytes() # No detections.
for f in faces: sensor.get_fb().draw_rectangle(f, color = (255, 255, 255))
out_face = max(faces, key = lambda f: f[2] * f[3])
return struct.pack("<HHHH", out_face[0], out_face[1], out_face[2], out_face[3])
# When called returns if there's a "person" or "no_person" within view.
#
# data is unused
def person_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
scores = tf.classify("person_detection", sensor.snapshot())[0].output()
return ['unsure', 'person', 'no_person'][scores.index(max(scores))].encode()
# When called returns the payload string for the largest qrcode
# within the OpenMV Cam's field-of-view.
#
# data is unused
def qrcode_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((320, 240))
codes = sensor.snapshot().find_qrcodes()
if not codes: return bytes() # No detections.
draw_detections(sensor.get_fb(), codes)
return max(codes, key = lambda c: c.w() * c.h()).payload().encode()
# When called returns a json list of json qrcode objects for all qrcodes in view.
#
# data is unused
def all_qrcode_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((320, 240))
codes = sensor.snapshot().find_qrcodes()
if not codes: return bytes() # No detections.
draw_detections(sensor.get_fb(), codes)
return str(codes).encode()
# When called returns the x/y centroid, id number, and rotation of the largest
# AprilTag within the OpenMV Cam's field-of-view.
#
# data is unused
def apriltag_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QQVGA)
tags = sensor.snapshot().find_apriltags()
if not tags: return bytes() # No detections.
draw_detections(sensor.get_fb(), tags)
output_tag = max(tags, key = lambda t: t.w() * t.h())
return struct.pack("<HHHH", output_tag.cx(), output_tag.cy(), output_tag.id(),
int(math.degrees(output_tag.rotation())))
# When called returns a json list of json apriltag objects for all apriltags in view.
#
# data is unused
def all_apriltag_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QQVGA)
tags = sensor.snapshot().find_apriltags()
if not tags: return bytes() # No detections.
draw_detections(sensor.get_fb(), tags)
return str(tags).encode()
# When called returns the payload string for the largest datamatrix
# within the OpenMV Cam's field-of-view.
#
# data is unused
def datamatrix_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((320, 240))
codes = sensor.snapshot().find_datamatrices()
if not codes: return bytes() # No detections.
draw_detections(sensor.get_fb(), codes)
return max(codes, key = lambda c: c.w() * c.h()).payload().encode()
# When called returns a json list of json datamatrix objects for all datamatrices in view.
#
# data is unused
def all_datamatrix_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((320, 240))
codes = sensor.snapshot().find_datamatrices()
if not codes: return bytes() # No detections.
draw_detections(sensor.get_fb(), codes)
return str(codes).encode()
# When called returns the payload string for the largest barcode
# within the OpenMV Cam's field-of-view.
#
# data is unused
def barcode_detection(data):
sensor.set_pixformat(sensor.GRAYSCALE)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((sensor.width(), sensor.height()//8))
codes = sensor.snapshot().find_barcodes()
if not codes: return bytes() # No detections.
return max(codes, key = lambda c: c.w() * c.h()).payload().encode()
# When called returns a json list of json barcode objects for all barcodes in view.
#
# data is unused
def all_barcode_detection(data):
sensor.set_pixformat(sensor.GRAYSCALE)
sensor.set_framesize(sensor.VGA)
sensor.set_windowing((sensor.width(), sensor.height()//8))
codes = sensor.snapshot().find_barcodes()
if not codes: return bytes() # No detections.
return str(codes).encode()
# When called returns the x/y centroid of the largest blob
# within the OpenMV Cam's field-of-view.
#
# data is the 6-byte color tracking threshold tuple of L_MIN, L_MAX, A_MIN, A_MAX, B_MIN, B_MAX.
def color_detection(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
thresholds = struct.unpack("<bbbbbb", data)
blobs = sensor.snapshot().find_blobs([thresholds],
pixels_threshold=500,
area_threshold=500,
merge=True,
margin=20)
if not blobs: return bytes() # No detections.
for b in blobs:
sensor.get_fb().draw_rectangle(b.rect(), color = (255, 0, 0))
sensor.get_fb().draw_cross(b.cx(), b.cy(), color = (0, 255, 0))
out_blob = max(blobs, key = lambda b: b.density())
return struct.pack("<HH", out_blob.cx(), out_blob.cy())
# When called returns a jpeg compressed image from the OpenMV
# Cam in one RPC call.
#
# data is unused
def jpeg_snapshot(data):
sensor.set_pixformat(sensor.RGB565)
sensor.set_framesize(sensor.QVGA)
return sensor.snapshot().compress(quality=90).bytearray()
# Register call backs.
interface.register_callback(face_detection)
interface.register_callback(person_detection)
interface.register_callback(qrcode_detection)
interface.register_callback(all_qrcode_detection)
interface.register_callback(apriltag_detection)
interface.register_callback(all_apriltag_detection)
interface.register_callback(datamatrix_detection)
interface.register_callback(all_datamatrix_detection)
interface.register_callback(barcode_detection)
interface.register_callback(all_barcode_detection)
interface.register_callback(color_detection)
interface.register_callback(jpeg_snapshot)
# Once all call backs have been registered we can start
# processing remote events. interface.loop() does not return.
interface.loop()

View File

@ -0,0 +1,55 @@
import pyb, micropython, array, uctypes
micropython.alloc_emergency_exception_buf(100)
class MutexException(OSError):
pass
class Mutex:
@micropython.asm_thumb
def _acquire(r0, r1): # Spinlock: wait on the semaphore. Return on success.
label(LOOP)
ldr(r0, [r1, 0]) # Wait for lock to be zero
cmp(r0, 0)
bne(LOOP) # Another process has the lock: spin on it
cpsid(0) # OK, we have lock at this instant disable interrupts
ldr(r0, [r1, 0]) # and re-check in case an interrupt occurred
cmp(r0, 0)
itt(ne) # if someone got in first re-enable ints
cpsie(0) # and start polling again
b(LOOP)
mov(r0, 1) # We have an exclusive access
str(r0, [r1, 0]) # set the lock
cpsie(0)
@micropython.asm_thumb
def _attempt(r0, r1): # Nonblocking. Try to lock. Return 0 on success, 1 on fail
cpsid(0) # disable interrupts
ldr(r0, [r1, 0])
cmp(r0, 0)
bne(FAIL) # Another process has the lock: fail
mov(r2, 1) # No lock
str(r2, [r1, 0]) # set the lock
label(FAIL)
cpsie(0) # enable interrupts
def __init__(self):
self.lock = array.array('i', (0,)) # 1 if a process has the lock else 0
# POSIX API pthread_mutex_lock() blocks the thread till resource is available.
def __enter__(self):
self._acquire(uctypes.addressof(self.lock))
return self
def __exit__(self, *_):
self.lock[0] = 0
# POSIX pthread_mutex_unlock()
def release(self):
if self.lock[0] == 0:
raise MutexException('Semaphore already released')
self.lock[0] = 0
# POSIX pthread_mutex_trylock() API. When mutex is not available the function returns immediately
def test(self): # Nonblocking: try to acquire, return True if success.
return self._attempt(uctypes.addressof(self.lock)) == 0

857
scripts/libraries/rpc.py Normal file
View File

@ -0,0 +1,857 @@
# This file is part of the OpenMV project.
#
# Copyright (c) 2013-2020 Ibrahim Abdelkader <iabdalkader@openmv.io>
# Copyright (c) 2013-2020 Kwabena W. Agyeman <kwagyeman@openmv.io>
#
# This work is licensed under the MIT license, see the file LICENSE for details.
import gc, network, omv, pyb, select, socket, stm, struct
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
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":
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":
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("<H", magic_value)
new_payload[2:-2] = payload
new_payload[-2:] = struct.pack("<H", self.__crc_16(new_payload, len(payload) + 2))
return new_payload
def _flush(self): # protected
pass
def get_bytes(self, buff, timeout_ms): # protected
return bytes()
def put_bytes(self, data, timeout_ms): # protected
pass
def stream_reader(self, call_back, queue_depth=1, read_timeout_ms=5000): # public
try: self._stream_put_bytes(self._set_packet(0xEDF6, struct.pack("<I", queue_depth)), 1000)
except OSError: return
tx_lfsr = 255
while True:
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 != 0x542E and crc != self.__crc_16(packet, len(packet) - 2): return
data = self._stream_get_bytes(bytearray(struct.unpack("<I", packet[2:-2])[0]), read_timeout_ms)
if data is None: return
call_back(data)
try: self._stream_put_bytes(struct.pack("<B", tx_lfsr), 1000)
except OSError: return
tx_lfsr = (tx_lfsr >> 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("<I", packet[2:-2])[0], self._stream_writer_queue_depth_max), 1)
rx_lfsr = 255
credits = queue_depth
while True:
if credits <= (queue_depth // 2):
data = self._stream_get_bytes(bytearray(1), 1000)
if data is None or data[0] != rx_lfsr: return
rx_lfsr = (rx_lfsr >> 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("<I", len(data))), 1000)
except OSError: return
try: self._stream_put_bytes(data, write_timeout_ms)
except OSError: return
credits -= 1
def _stream_get_bytes(self, buff, timeout_ms): # protected
return self.get_bytes(buff, timeout_ms)
def _stream_put_bytes(self, data, timeout_ms): # protected
self.put_bytes(data, timeout_ms)
class rpc_master(rpc):
def __init__(self): # private
rpc.__init__(self)
self.__in_command_header_buf = self._get_packet_pre_alloc()
self.__in_command_data_buf = self._get_packet_pre_alloc()
self.__out_result_header_ack = self._set_packet(self._RESULT_HEADER_PACKET_MAGIC)
self.__in_result_header_buf = self._get_packet_pre_alloc(4)
self.__out_result_data_ack = self._set_packet(self._RESULT_DATA_PACKET_MAGIC)
self._put_short_timeout_reset = 3
self._get_short_timeout_reset = 3
self._put_long_timeout = 5000
self._get_long_timeout = 5000
def __put_command(self, command, data, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
out_header = self._set_packet(self._COMMAND_HEADER_PACKET_MAGIC, struct.pack("<II", command, len(data)))
out_data = self._set_packet(self._COMMAND_DATA_PACKET_MAGIC, data)
start = pyb.millis()
while pyb.elapsed_millis(start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_command_header_buf[0], len(self.__in_command_header_buf[0]))
self._zero(self.__in_command_data_buf[0], len(self.__in_command_data_buf[0]))
self._flush()
self.put_bytes(out_header, self._put_short_timeout)
if self._get_packet(self._COMMAND_HEADER_PACKET_MAGIC, self.__in_command_header_buf, self._get_short_timeout) is not None:
self.put_bytes(out_data, self._put_long_timeout)
if self._get_packet(self._COMMAND_DATA_PACKET_MAGIC, self.__in_command_data_buf, self._get_short_timeout) is not None:
return True
# Avoid timeout livelocking.
self._put_short_timeout = min((self._put_short_timeout * 4) // 3, timeout)
self._get_short_timeout = min((self._get_short_timeout * 4) // 3, timeout)
return False
def __get_result(self, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
start = pyb.millis()
while pyb.elapsed_millis(start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_result_header_buf[0], len(self.__in_result_header_buf[0]))
self._flush()
self.put_bytes(self.__out_result_header_ack, self._put_short_timeout)
packet = self._get_packet(self._RESULT_HEADER_PACKET_MAGIC, self.__in_result_header_buf, self._get_short_timeout)
if packet is not None:
in_result_data_buf = self._get_packet_pre_alloc(struct.unpack("<I", packet)[0])
self.put_bytes(self.__out_result_data_ack, self._put_short_timeout)
dat_packet = self._get_packet(self._RESULT_DATA_PACKET_MAGIC, in_result_data_buf, self._get_long_timeout)
if dat_packet is not None:
return dat_packet
# Avoid timeout livelocking.
self._put_short_timeout = min((self._put_short_timeout * 4) // 3, timeout)
self._get_short_timeout = min((self._get_short_timeout * 4) // 3, timeout)
return None
def call(self, name, data=bytes(), send_timeout=1000, recv_timeout=1000): # public
return self.__get_result(recv_timeout) if self.__put_command(self._hash(name, len(name)), data, send_timeout) else None
class rpc_slave(rpc):
def __init__(self): # private
self.__dict = {}
self.__schedule_cb = None
self.__loop_cb = None
rpc.__init__(self)
self.__in_command_header_buf = self._get_packet_pre_alloc(8)
self.__out_command_header_ack = self._set_packet(self._COMMAND_HEADER_PACKET_MAGIC)
self.__out_command_data_ack = self._set_packet(self._COMMAND_DATA_PACKET_MAGIC)
self.__in_response_header_buf = self._get_packet_pre_alloc()
self.__in_response_data_buf = self._get_packet_pre_alloc()
self._put_short_timeout_reset = 2
self._get_short_timeout_reset = 2
self._put_long_timeout = 5000
self._get_long_timeout = 5000
def __get_command(self, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
start = pyb.millis()
while pyb.elapsed_millis(start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_command_header_buf[0], len(self.__in_command_header_buf[0]))
self._flush()
packet = self._get_packet(self._COMMAND_HEADER_PACKET_MAGIC, self.__in_command_header_buf, self._get_short_timeout)
if packet is not None:
command, datalen = struct.unpack("<II", packet)
in_command_data_buf = self._get_packet_pre_alloc(datalen)
self.put_bytes(self.__out_command_header_ack, self._put_short_timeout)
dat_packet = self._get_packet(self._COMMAND_DATA_PACKET_MAGIC, in_command_data_buf, self._get_long_timeout)
if dat_packet is not None:
self.put_bytes(self.__out_command_data_ack, self._put_short_timeout)
return (command, dat_packet)
# Avoid timeout livelocking.
self._put_short_timeout = min(self._put_short_timeout + 1, timeout)
self._get_short_timeout = min(self._get_short_timeout + 1, timeout)
return (None, None)
def __put_result(self, data, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
out_header = self._set_packet(self._RESULT_HEADER_PACKET_MAGIC, struct.pack("<I", len(data)))
out_data = self._set_packet(self._RESULT_DATA_PACKET_MAGIC, data)
start = pyb.millis()
while pyb.elapsed_millis(start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_response_header_buf[0], len(self.__in_response_header_buf[0]))
self._zero(self.__in_response_data_buf[0], len(self.__in_response_data_buf[0]))
self._flush()
if self._get_packet(self._RESULT_HEADER_PACKET_MAGIC, self.__in_response_header_buf, self._get_short_timeout) is not None:
self.put_bytes(out_header, self._put_short_timeout)
if self._get_packet(self._RESULT_DATA_PACKET_MAGIC, self.__in_response_data_buf, self._get_short_timeout) is not None:
self.put_bytes(out_data, self._put_long_timeout)
return True
# Avoid timeout livelocking.
self._put_short_timeout = min(self._put_short_timeout + 1, timeout)
self._get_short_timeout = min(self._get_short_timeout + 1, timeout)
return False
def register_callback(self, cb): # public
self.__dict[self._hash(cb.__name__, len(cb.__name__))] = cb
def schedule_callback(self, cb): # public
self.__schedule_cb = cb
def setup_loop_callback(self, cb): # public
self.__loop_cb = cb
def loop(self, recv_timeout=1000, send_timeout=1000): # public
while True:
command, data = self.__get_command(recv_timeout)
if command is not None:
cb = self.__dict.get(command)
if self.__put_result(cb(data) if cb is not None else bytes(), send_timeout) and self.__schedule_cb is not None:
self.__schedule_cb()
self.__schedule_cb = None
if self.__loop_cb is not None: self.__loop_cb()
def __get_can_settings(bit_rate, sampling_point):
clk = 48000000 if omv.board_type() == "H7" else pyb.freq()[2]
for prescaler in range(8):
for bs1 in range(16):
for bs2 in range(8):
if bit_rate == ((clk >> prescaler) // (1 + bs1 + bs2)) and (sampling_point * 10) == (((1 + bs1) * 1000) // (1 + bs1 + bs2)):
return (1 << prescaler, bs1, bs2)
raise ValueError("Invalid bit_rate and/or sampling_point!")
class rpc_can_master(rpc_master):
def __init__(self, message_id=0x7FF, bit_rate=250000, sampling_point=75):
self.__message_id = message_id
can_prescaler, can_bs1, can_bs2 = __get_can_settings(bit_rate, sampling_point)
self.__can = pyb.CAN(2, pyb.CAN.NORMAL, prescaler=can_prescaler, bs1=can_bs1, bs2=can_bs2, auto_restart=True)
self.__can.setfilter(0, pyb.CAN.DUAL if omv.board_type() == "H7" else pyb.CAN.LIST32, 0, [message_id, message_id])
rpc_master.__init__(self)
def _flush(self): # private
while self.__can.any(0): self.__can.recv(0)
def get_bytes(self, buff, timeout_ms): # protected
msg = bytearray(8)
lst = [0, 0, 0, memoryview(msg)]
l = len(buff)
for i in range(0, l, 8):
expected = min(l - i, 8)
try:
id, rtr, fmi, data = self.__can.recv(0, lst, timeout=timeout_ms)
if id == self.__message_id and rtr == 0 and fmi == 0 and len(data) == expected: buff[i:i+8] = data
else:
pyb.delay(self._get_short_timeout)
return None
except OSError:
pyb.delay(self._get_short_timeout)
return None
return buff
def put_bytes(self, data, timeout_ms): # protected
view = memoryview(data)
for i in range(0, len(view), 8):
try: self.__can.send(view[i:i+8], self.__message_id, timeout=timeout_ms)
except OSError: break
class rpc_can_slave(rpc_slave):
def __init__(self, message_id=0x7FF, bit_rate=250000, sampling_point=75):
self.__message_id = message_id
can_prescaler, can_bs1, can_bs2 = __get_can_settings(bit_rate, sampling_point)
self.__can = pyb.CAN(2, pyb.CAN.NORMAL, prescaler=can_prescaler, bs1=can_bs1, bs2=can_bs2, auto_restart=True)
self.__can.setfilter(0, pyb.CAN.DUAL if omv.board_type() == "H7" else pyb.CAN.LIST32, 0, [message_id, message_id])
rpc_slave.__init__(self)
def _flush(self): # private
while self.__can.any(0): self.__can.recv(0)
def get_bytes(self, buff, timeout_ms): # protected
msg = bytearray(8)
lst = [0, 0, 0, memoryview(msg)]
l = len(buff)
for i in range(0, l, 8):
expected = min(l - i, 8)
try:
id, rtr, fmi, data = self.__can.recv(0, lst, timeout=timeout_ms)
if id == self.__message_id and rtr == 0 and fmi == 0 and len(data) == expected: buff[i:i+8] = data
else: return None
except OSError: return None
return buff
def put_bytes(self, data, timeout_ms): # protected
view = memoryview(data)
for i in range(0, len(view), 8):
try: self.__can.send(view[i:i+8], self.__message_id, timeout=timeout_ms)
except OSError: break
class rpc_i2c_master(rpc_master):
def __init__(self, slave_addr=0x12, rate=100000): # private
self.__addr = slave_addr
self.__freq = rate
self.__i2c = pyb.I2C(2)
rpc_master.__init__(self)
self._stream_writer_queue_depth_max = 1
def get_bytes(self, buff, timeout_ms): # protected
view = memoryview(buff)
for i in range(0, len(view), 65535):
pyb.udelay(100) # Give slave time to get ready.
self.__i2c.init(pyb.I2C.MASTER, baudrate=self.__freq, dma=True)
try: self.__i2c.recv(view[i:i+65535], self.__addr, timeout=timeout_ms)
except OSError: view = None
self.__i2c.deinit()
if view is None: break
if view is None or self._same(view, len(view)): pyb.delay(self._get_short_timeout)
return view
def put_bytes(self, data, timeout_ms): # protected
view = memoryview(data)
for i in range(0, len(view), 65535):
pyb.udelay(100) # Give slave time to get ready.
self.__i2c.init(pyb.I2C.MASTER, baudrate=self.__freq, dma=True)
try: self.__i2c.send(view[i:i+65535], self.__addr, timeout=timeout_ms)
except OSError: view = None
self.__i2c.deinit()
if view is None: break
class rpc_i2c_slave(rpc_slave):
def __init__(self, slave_addr=0x12): # private
self.__addr = slave_addr
self.__i2c = pyb.I2C(2)
rpc_slave.__init__(self)
self._stream_writer_queue_depth_max = 1
def get_bytes(self, buff, timeout_ms): # protected
view = memoryview(buff)
for i in range(0, len(view), 65535):
self.__i2c.init(pyb.I2C.SLAVE, addr=self.__addr, dma=True)
try: self.__i2c.recv(view[i:i+65535], timeout=timeout_ms)
except OSError: view = None
self.__i2c.deinit()
if view is None: break
return view
def put_bytes(self, data, timeout_ms): # protected
view = memoryview(data)
for i in range(0, len(view), 65535):
self.__i2c.init(pyb.I2C.SLAVE, addr=self.__addr, dma=True)
try: self.__i2c.send(view[i:i+65535], timeout=timeout_ms)
except OSError: view = None
self.__i2c.deinit()
if view is None: break
class rpc_spi_master(rpc_master):
def __init__(self, cs_pin="P3", freq=1000000, clk_polarity=1, clk_phase=0): # private
self.__pin = pyb.Pin(cs_pin, pyb.Pin.OUT_PP)
self.__freq = freq
self.__polarity = clk_polarity
self.__clk_phase = clk_phase
self.__spi = pyb.SPI(2)
rpc_master.__init__(self)
self._stream_writer_queue_depth_max = 1
def get_bytes(self, buff, timeout_ms): # protected
self.__pin.value(False)
pyb.udelay(100) # Give slave time to get ready.
self.__spi.init(pyb.SPI.MASTER, self.__freq, 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()
self.__pin.value(True)
if buff is None or self._same(buff, len(buff)): pyb.delay(self._get_short_timeout)
return buff
def put_bytes(self, data, timeout_ms): # protected
self.__pin.value(False)
pyb.udelay(100) # Give slave time to get ready.
self.__spi.init(pyb.SPI.MASTER, self.__freq, polarity=self.__polarity, phase=self.__clk_phase)
try: self.__spi.send(data, timeout=timeout_ms)
except OSError: pass
self.__spi.deinit()
self.__pin.value(True)
class rpc_spi_slave(rpc_slave):
def __init__(self, cs_pin="P3", clk_polarity=1, clk_phase=0): # private
self.__pin = pyb.Pin(cs_pin, pyb.Pin.IN)
self.__polarity = clk_polarity
self.__clk_phase = clk_phase
self.__spi = pyb.SPI(2)
rpc_slave.__init__(self)
self._stream_writer_queue_depth_max = 1
def get_bytes(self, buff, timeout_ms): # protected
start = pyb.millis()
while(self.__pin.value()):
if pyb.elapsed_millis(start) >= 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
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): # private
self.__uart = pyb.UART(3, 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:
pyb.delay(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): # private
self.__uart = pyb.UART(3, 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
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
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)
class rpc_wifi_master(rpc_master):
def __valid_tcp_socket(self): # private
if self.__tcp__socket is None:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(self.__myaddr)
s.listen(0)
s.settimeout(1)
self.__tcp__socket, addr = s.accept()
s.close()
except OSError: self.__tcp__socket = None
return self.__tcp__socket is not None
def __close_tcp_socket(self): # private
self.__tcp__socket.close()
self.__tcp__socket = None
def __valid_udp_socket(self): # private
if self.__udp__socket is None:
try:
self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__udp__socket.bind(self.__myaddr)
except OSError: self.__udp__socket = None
return self.__udp__socket is not None
def __close_udp_socket(self): # private
self.__udp__socket.close()
self.__udp__socket = None
def __init__(self, ssid, ssid_key, ssid_security, ip, port=0x1DBA, mode=network.WINC.MODE_STA, static_ip=None): # private
self._udp_limit = 1400
self._timeout_scale = 10
self.__winc = network.WINC(mode=mode)
if mode == network.WINC.MODE_STA:
if static_ip is not None: self.__winc.ifconfig(static_ip)
self.__winc.connect(ssid, key=ssid_key, security=ssid_security)
if not self.__winc.isconnected(): raise OSError("Failed to connect to network!")
elif mode == network.WINC.MODE_AP: self.__winc.start_ap(ssid, key=ssid_key, security=ssid_security)
else: raise ValueError("Invalid mode")
self.__myip = self.__winc.ifconfig()[0]
self.__myaddr = (self.__myip, port)
self.__slave_addr = (ip, port)
self.__tcp__socket = None
self.__udp__socket = None
print("IP Address:Port %s:%d\nRunning..." % self.__myaddr)
rpc_master.__init__(self)
def _flush(self): # protected
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(0.001)
while(True):
data, addr = self.__udp__socket.recvfrom(1400)
if not len(data): break
except OSError: self.__close_udp_socket()
if self.__tcp__socket is not None:
try:
self.__tcp__socket.settimeout(0.001)
while(True):
data = self.__tcp__socket.recv(1400)
if not len(data): break
except OSError: self.__close_tcp_socket()
def get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale)
while l:
data, addr = self.__udp__socket.recvfrom(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
# We don't need to close the socket on error since it's connectionless.
except OSError: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
return buff if not l else None
def put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale)
while l:
data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__slave_addr)
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_udp_socket()
except OSError: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
def _stream_get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
return buff if not l else None
def _stream_put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
if l: raise OSError # Stop Stream.
class rpc_wifi_slave(rpc_slave):
def __valid_tcp_socket(self): # private
if self.__tcp__socket is None:
try:
self.__tcp__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.__tcp__socket.connect(self.__master_addr)
except OSError: self.__tcp__socket = None
return self.__tcp__socket is not None
def __close_tcp_socket(self): # private
self.__tcp__socket.close()
self.__tcp__socket = None
def __valid_udp_socket(self): # private
if self.__udp__socket is None:
try:
self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__udp__socket.bind(self.__myaddr)
except OSError: self.__udp__socket = None
return self.__udp__socket is not None
def __close_udp_socket(self): # private
self.__udp__socket.close()
self.__udp__socket = None
def __init__(self, ssid, ssid_key, ssid_security, port=0x1DBA, mode=network.WINC.MODE_STA, static_ip=None): # private
self._udp_limit = 1400
self._timeout_scale = 10
self.__winc = network.WINC(mode=mode)
if mode == network.WINC.MODE_STA:
if static_ip is not None: self.__winc.ifconfig(static_ip)
self.__winc.connect(ssid, key=ssid_key, security=ssid_security)
if not self.__winc.isconnected(): raise OSError("Failed to connect to network!")
elif mode == network.WINC.MODE_AP: self.__winc.start_ap(ssid, key=ssid_key, security=ssid_security)
else: raise ValueError("Invalid mode")
self.__myip = self.__winc.ifconfig()[0]
self.__myaddr = (self.__myip, port)
self.__master_addr = None
self.__tcp__socket = None
self.__udp__socket = None
print("IP Address:Port %s:%d\nRunning..." % self.__myaddr)
rpc_slave.__init__(self)
def _flush(self): # protected
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(0.001)
while(True):
data, addr = self.__udp__socket.recvfrom(1400)
if not len(data): break
except OSError: self.__close_udp_socket()
if self.__tcp__socket is not None:
try:
self.__tcp__socket.settimeout(0.001)
while(True):
data = self.__tcp__socket.recv(1400)
if not len(data): break
except OSError: self.__close_tcp_socket()
def get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale)
while l:
data, addr = self.__udp__socket.recvfrom(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
self.__master_addr = addr
i += data_len
l -= data_len
# We don't need to close the socket on error since it's connectionless.
except OSError: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
return buff if not l else None
def put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale)
while l:
data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__master_addr)
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_udp_socket()
except OSError: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
def _stream_get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
return buff if not l else None
def _stream_put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except OSError: self.__close_tcp_socket()
if l: raise OSError # Stop Stream.

161
tools/rpc/README.md Normal file
View File

@ -0,0 +1,161 @@
# OpenMV Remote Python Call Library
The `rpc` module on the OpenMV Cam allows you to connect your OpenMV Cam to another microcontroller or computer and execute remote python (or procedure) calls on your OpenMV Cam. The `rpc` module also allows for the reverse too if you want your OpenMV Cam to be able to execute remote procedure (or python) calls on another microcontroller or computer.
For computer control the [rpc](rpc.py) python module in this directory implements the OpenMV `rpc` protocol for control of an OpenMV Cam over USB VCP (i.e. a USB serial port) or over Ethernet/WiFi (i.e. over sockets).
# Library Dependencies
The OpenMV Cam `rpc` library on the computer only depends on [pyserial](https://pythonhosted.org/pyserial/). All other modules used by it come installed with python. To get `pyserial` just do:
pip install pyserial
Because the interface library is implemented in pure python with no external dependencies it works on Windows, Mac, and Linux.
# How to use the Library
Please checkout the following scripts for how to control your OpenMV Cam from the comptuer:
* [Slow but Synchronus JPG Image Transfer](rpc_image_transfer_jpg_as_the_controller_device.py)
* [Fast JPG Image Streaming](rpc_image_transfer_jpg_streaming_as_the_controller_device.py)
* [Face Detection, April Tag Detection, Color Tracking, and more](rpc_popular_features_as_the_controller_device_example.py)
You will need to edit the example code above to choose which interface you want to use (USB versus Ethernet/WiFi) and to play with the settings the scripts use. When you run the scripts make sure to save them first after editing them and then run them with `python -u <script_name>` to make sure that script output to stdio is not buffered.
In general, for the controller device to use the `rpc` library you will create an interface object using the `rpc` library. For example:
interface = rpc.rpc_usb_vcp_master("COM3")
This create a USB VCP interface to talk to your OpenMV Cam over COM3. For Mac and Linux you would pass some type of `/dev/...` device instead.
Once the interface is created you just need to do:
memory_view_object_result = interface.call("remote_function_or_method_name", bytes_object_argument)
And the `rpc` library will try to execute that `"remote_function_or_method_name"` on your OpenMV Cam. The remote function or method will receive the `bytes_object_argument` which can be up to 2^32-1 bytes in size. Once the remote method finishes executing it will return a `memory_view_object_result` which can also be up to 2^32-1 bytes in size. Because the argument and response are both generic byte containers you can pass anything through the `rpc` library and receive any type of response. A simple way to pass arguments is to use `struct.pack()` to create the argument and `struct.unpack()` to receieve the argument on the OpenMV Cam side. For the response, the OpenMV Cam may send a string object or json string as the result which the computer can then interpret. Most objects or lists returned from method calls on the OpenMV Cam generate valid json strings when you call `str()` on them.
As for errors, if you try to execute a non-existant function or method name on the OpenMV Cam the `call` method will return an empty `bytes()` object. If the `rpc` library failed to communicate with the OpenMV Cam the `rpc` library will return `None`.
To keep things simple the `rpc` library doesn't maintain a connection between the master and slave devices. The `call` method encapsulates trying to connect to the OpenMV Cam, starting execution of the remote function or method, and getting the result.
Now, on the OpenMV Cam side of things you have to create an `rpc` interface to communicate with the computer. This looks like:
interface = rpc.rpc_usb_vcp_slave()
This will create the interface layer on the OpenMV Cam (this needs to be done on a script running on the OpenMV Cam - see the [Remote Control](../../scripts/examples/34-Remote-Control) example scripts).
Once you create the slave interface you then need to register call backs that the master can call with the interface object.
def remote_function_or_method_name(memoryview_object_argument):
<lots of code>
return bytes_object_result
interface.register_callback(remote_function_or_method_name)
You may register as many callbacks as you like on the OpenMV Cam that the computer can call. Finally, once you are done registering callbacks you just need to execute:
interface.loop()
On the OpenMV Cam to start the `rpc` library up and begin listening for the computer. Note that the `loop()` method does not return. Also, to make your OpenMV Cam more robust against errors you may want to wrap the `loop()` with `try:` and `except:` for whatever exceptions might be thrown by your callback methods. The `rpc` library will not generate any exceptions itself. Note: passing large data structures around (like jpeg images) can potentially exhaust the heap on the OpenMV Cam and generate `MemoryError` exceptions.
And that is it! The `rpc` library is designed to be simple to use. It was designed to allow remote control of the OpenMV Cam by a computer or microcontroller so there are also interfaces for control of your OpenMV Cam over CAN, I2C, SPI, and UART.
# API
Please see the example scripts above for starting code on how to use the `rpc` library. The below API documents the public interface of the library.
## class rpc():
The `rpc` base class is reimplemented by the `rpc_master` and `rpc_slave` classes to create the master and slave interfaces. It is a pure virtual class and not meant to be used directly.
#### get_bytes(buff, timeout_ms):
This method is meant to be reimplemented by specific interface classes of `rpc_master` and `rpc_slave`. It should fill the `buff` argument which is either a `bytearray` or `memoryview` object of bytes from the interface equal to the length of the `buff` object in `timeout_ms` milliseconds. On timeout this method should return `None`. Note that for master and slave synchronization this method should try to always complete in at least `timeout_ms` milliseconds and not faster as the `rpc_master` and `rpc_slave` objects will automatically increase the `timeout_ms` to synchronize.
#### put_bytes(data, timeout_ms):
This method is meant to be reimplemented by specific interface classes of `rpc_master` and `rpc_slave`. It should send `data` bytes on the interface within `timeout_ms` milliseconds. If it completes faster than the timeout that is okay. No return value is expected.
#### stream_reader(call_back, queue_depth=1, read_timeout_ms=5000):
This method is meant to be called directly. After synchronization of the master and slave on return of a callback `stream_reader` may be called to receive data as fast as possible from the master or slave device. `call_back` will be called repeatedly with a `bytes_or_memory_view` argument that was sent by the `stream_writer`. `call_back` is not expected to return anything. `queue_depth` defines how many frames of data the `stream_writer` may generate before slowing down and waiting on the `stream_reader`. Higher `queue_depth` values lead to higher performance (up to a point) but require the `stream_reader` to be able to handle outstanding packets in its interface layer. Note that computers typically do not buffer much more than 4KB of data in device driver buffers. If you make the `queue_depth` larger than 1 then `call_back` should return very quickly and not block. Otherwise, you should implement a multi-thread architecture to process the received data so that `stream_reader` is always executing and moving data out of device driver buffers into larger user memory buffers. Finally, `read_timeout_ms` defines how many milliseconds to wait to receive the `bytes_or_memory_view` payload per `call_back`.
On any errors `stream_reader` will return. The master and slave devices can try to setup the stream again afterwards to continue as demonstrated in [Fast JPG Image Streaming](rpc_image_transfer_jpg_streaming_as_the_controller_device.py).
If you need to cancel the `stream_reader` just raise an exception in the `call_back` and catch it. The remote side will automatically timeout.
#### stream_writer(call_back, write_timeout_ms=5000):
This method is meant to be called directly. After synchronization of the master and slave on return of a callback `stream_writer` may be called to send data as fast as possible from the master or slave device. `call_back` will be called repeatedly and should return a `bytes_or_memory_view` object that will be sent to the `stream_reader`. `call_back` should not take any arguments. Finally, `write_timeout_ms` defines how many milliseconds to wait to send the `bytes_or_memory_view` object returned by `call_back`.
On any errors `stream_writer` will return. The master and slave devices can try to setup the stream again afterwards to continue as demonstrated in [Fast JPG Image Streaming](rpc_image_transfer_jpg_streaming_as_the_controller_device.py).
If you need to cancel the `stream_writer` just raise an exception in the `call_back` and catch it. The remote side will automatically timeout.
## class rpc_master():
The `rpc_master` master is a pure virtual class and not meant to be used directly. Specific interface classes should reimplement `rpc_master`.
#### call(name, data=bytes(), send_timeout=1000, recv_timeout=1000):
Executes a remote call on the slave device. `name` is a string name of the remote function or method to execute. `data` is the `bytes` like object that will be sent as the argument of the remote function or method to exeucte. `send_timeout` defines how many milliseconds to wait while trying to connect to the slave and get it to execute the remote function or method. Once the master starts sending the argument to the slave deivce `send_timeout` does not apply. The library will allow the argument to take up to 5 seconds to be sent. `recv_timeout` defines how many milliseconds to wait after the slave started executing the remote method to receive the repsonse. Note that once the master starts receiving the repsonse `recv_timeout` does not apply. The library will allow the response to take up to 5 seconds to be received.
Note that a new packet that includes a copy of `data` will be created internally inside the `rpc` library. You may encounter memory issues on the OpenMV Cam if you try to pass very large data arguments.
## class rpc_slave():
The `rpc_slave` master is a pure virtual class and not meant to be used directly. Specific interface classes should reimplement `rpc_slave`.
#### register_callback(cb):
Registers a call back that can be executed by the master device. The call back should take one argument which will be a memoryview object and it should return a `bytes()` like object as the result. The call back should return in less than 1 second if possible.
#### schedule_callback(cb):
After you execute `loop()` it is not possible to execute long running operations outside of the `rpc` library. `schedule_callback` allows you to break out of the `rpc` library temporarily after completion of an call back. You should execute `schedule_callback` during the execution of an `rpc` call back method to register a new non-rpc call back that will be executed immediately after the successful completion of that call back you executed `schedule_callback` in. The function or method should not take any arguments. After the the call back that was registered returns it must be registered again in the next parent call back. On any error of the parent call back the registered call back will not be called and must be registered again. Here's how to use this:
def some_function_or_method_that_takes_a_long_time_to_execute():
<do stuff>
def normal_rpc_call_back(data):
<process data>
interface.schedule_callback(some_function_or_method_that_takes_a_long_time_to_execute)
return bytes(response)
interface.register_callback(normal_rpc_call_back)
interface.loop()
`schedule_callback` in particular allows you to use the `get_bytes` and `put_bytes` methods for cut-through data transfer between one device and another without the cost of packetization which limits the size of the data removed inside the `rpc` library without running out of memory on the OpenMV Cam.
#### setup_loop_callback(cb):
The loop call back is called every loop iteration of `loop()`. Unlike the `schedule_callback()` call back this call back stays registered after being registered once. You can use the loop call back to blink an activity LED or something like that. You should not use the loop call back to execute any blocking code as this will get in the way of polling for communication from the master. Additionally, the loop call back will be called at a variable rate depending on when and what call backs the master is trying to execute. Given this, the loop call back is not suitable for any method that needs to be executed at a fixed frequency.
On the OpenMV Cam, if you need to execute something at a fixed frequency, you should setup a timer before executing `loop()` and use a timer interrupt based callback to execute some function or method at a fixed frequency. Please see how to [Write Interrupt Handlers](http://docs.openmv.io/reference/isr_rules.html) for more information. Note: The `mutex` library is installed on your OpenMV Cam along with the `rpc` library.
#### loop(recv_timeout=1000, send_timeout=1000):
Starts execution of the `rpc` library on the slave to receive data. This method does not return (except via an exception from a call back). You should register all call backs first before executing this method. However, it is possible to register new call backs inside of a call back previously being registered that is executing.
`recv_timeout` defines how long to wait to receive a command from the master device before trying again. `send_timeout` defines how long the slave will wait for the master to receive the call back response before going back to trying to receive. The loop call back will be executed before trying to receive again.
## rpc_usb_vcp_master(port):
Creates a master implementation of the `rpc` library to communicate over a USB VCP (virtual COM port). `port` is the string name of the serial port.
Communication over USB is the most reliable and high speed way to connect an OpenMV Cam to a computer. However, as the OpenMV Cam has only one VCP port we recommend that you fully debug your script using the `rpc_usb_vcp_slave` on your OpenMV using another `rpc` interface if possible as you will not be able to get error messages off the OpenMV Cam easily. For example, connecting the `rpc_usb_vcp_master` to the `rpc_uart_slave` on your OpenMV Cam at 115200 BPS over a USB-to-Serial adapter when debugging will make your life far easier. Alternatively, use the Ethernet or WiFi interface when debugging.
## rpc_usb_vcp_slave(port):
Creates a slave implementation of the `rpc` library to communicate over a USB VCP (virtual COM port). `port` is the string name of the serial port.
Communication over USB is the most reliable and high speed way to connect an OpenMV Cam to a computer. However, as the OpenMV Cam has only one VCP port we recommend that you fully debug your script using the `rpc_usb_vcp_master` on your OpenMV using another `rpc` interface if possible as you will not be able to get error messages off the OpenMV Cam easily. For example, connecting the `rpc_usb_vcp_slave` to the `rpc_uart_master` on your OpenMV Cam at 115200 BPS over a USB-to-Serial adapter when debugging will make your life far easier. Alternatively, use the Ethernet or WiFi interface when debugging.
## rpc_wifi_or_ethernet_master(slave_ip, my_ip="", port=0x1DBA):
Creates a master implementation of the `rpc` library to communicate over WiFi or Ethernet. `slave_ip` is the IPV4 address of the `rpc` slave device. `my_ip` can be `""` which binds the master to any interface adapter to communicate to the slave. If `my_ip` is not `""` then it should be an IP address on the same subnet as `slave_ip`. `port` is a free port to use for UDP and TCP traffic.
## rpc_wifi_or_ethernet_slave(my_ip="", port=0x1DBA):
Creates a slave implementation of the `rpc` library to communicate over WiFi or Ethernet. `my_ip` can be `""` which binds the slave to any interface adapter to communicate to the master. If `my_ip` is not `""` then it should be an IP address on the same subnet as the master. `port` is a free port to use for UDP and TCP traffic.

619
tools/rpc/rpc.py Normal file
View File

@ -0,0 +1,619 @@
# This file is part of the OpenMV project.
#
# Copyright (c) 2013-2020 Ibrahim Abdelkader <iabdalkader@openmv.io>
# Copyright (c) 2013-2020 Kwabena W. Agyeman <kwagyeman@openmv.io>
#
# This work is licensed under the MIT license, see the file LICENSE for details.
import gc, serial, socket, struct, time
class rpc:
_COMMAND_HEADER_PACKET_MAGIC = 0x1209
_COMMAND_DATA_PACKET_MAGIC = 0xABD1
_RESULT_HEADER_PACKET_MAGIC = 0x9021
_RESULT_DATA_PACKET_MAGIC = 0x1DBA
def __def_crc_16(self, data, size): # private
crc = 0xFFFF
for i in range(size):
crc ^= data[i] << 8
for j in range(8): crc = (crc << 1) ^ (0x1021 if crc & 0x8000 else 0)
return crc & 0xFFFF
def _zero(self, buff, size): # private
for i in range(size): buff[i] = 0
def _same(self, data, size): # private
if not size: return False
old = data[0]
for i in range(1, size):
new = data[i]
if new != old: return False
old = new
return True
# djb2 algorithm; see http://www.cse.yorku.ca/~oz/hash.html
def _hash(self, data, size): # private
h = 5381
for i in range(size):
h = ((h << 5) + h) ^ ord(data[i])
return h & 0xFFFFFFFF
def __init__(self): # private
self.__crc_16 = self.__def_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("<H", magic_value)
new_payload[2:-2] = payload
new_payload[-2:] = struct.pack("<H", self.__crc_16(new_payload, len(payload) + 2))
return new_payload
def _flush(self): # protected
pass
def get_bytes(self, buff, timeout_ms): # protected
return bytes()
def put_bytes(self, data, timeout_ms): # protected
pass
def stream_reader(self, call_back, queue_depth=1, read_timeout_ms=5000): # public
try: self._stream_put_bytes(self._set_packet(0xEDF6, struct.pack("<I", queue_depth)), 1000)
except OSError: return
tx_lfsr = 255
while True:
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 != 0x542E and crc != self.__crc_16(packet, len(packet) - 2): return
data = self._stream_get_bytes(bytearray(struct.unpack("<I", packet[2:-2])[0]), read_timeout_ms)
if data is None: return
call_back(data)
try: self._stream_put_bytes(struct.pack("<B", tx_lfsr), 1000)
except OSError: return
tx_lfsr = (tx_lfsr >> 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("<I", packet[2:-2])[0], self._stream_writer_queue_depth_max), 1)
rx_lfsr = 255
credits = queue_depth
while True:
if credits <= (queue_depth // 2):
data = self._stream_get_bytes(bytearray(1), 1000)
if data is None or data[0] != rx_lfsr: return
rx_lfsr = (rx_lfsr >> 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("<I", len(data))), 1000)
except OSError: return
try: self._stream_put_bytes(data, write_timeout_ms)
except OSError: return
credits -= 1
def _stream_get_bytes(self, buff, timeout_ms): # protected
return self.get_bytes(buff, timeout_ms)
def _stream_put_bytes(self, data, timeout_ms): # protected
self.put_bytes(data, timeout_ms)
class rpc_master(rpc):
def __init__(self): # private
rpc.__init__(self)
self.__in_command_header_buf = self._get_packet_pre_alloc()
self.__in_command_data_buf = self._get_packet_pre_alloc()
self.__out_result_header_ack = self._set_packet(self._RESULT_HEADER_PACKET_MAGIC)
self.__in_result_header_buf = self._get_packet_pre_alloc(4)
self.__out_result_data_ack = self._set_packet(self._RESULT_DATA_PACKET_MAGIC)
self._put_short_timeout_reset = 3
self._get_short_timeout_reset = 3
self._put_long_timeout = 5000
self._get_long_timeout = 5000
def __put_command(self, command, data, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
out_header = self._set_packet(self._COMMAND_HEADER_PACKET_MAGIC, struct.pack("<II", command, len(data)))
out_data = self._set_packet(self._COMMAND_DATA_PACKET_MAGIC, data)
start = int(time.time() * 1000)
while (int(time.time() * 1000) - start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_command_header_buf[0], len(self.__in_command_header_buf[0]))
self._zero(self.__in_command_data_buf[0], len(self.__in_command_data_buf[0]))
self._flush()
self.put_bytes(out_header, self._put_short_timeout)
if self._get_packet(self._COMMAND_HEADER_PACKET_MAGIC, self.__in_command_header_buf, self._get_short_timeout) is not None:
self.put_bytes(out_data, self._put_long_timeout)
if self._get_packet(self._COMMAND_DATA_PACKET_MAGIC, self.__in_command_data_buf, self._get_short_timeout) is not None:
return True
# Avoid timeout livelocking.
self._put_short_timeout = min((self._put_short_timeout * 4) // 3, timeout)
self._get_short_timeout = min((self._get_short_timeout * 4) // 3, timeout)
return False
def __get_result(self, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
start = int(time.time() * 1000)
while (int(time.time() * 1000) - start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_result_header_buf[0], len(self.__in_result_header_buf[0]))
self._flush()
self.put_bytes(self.__out_result_header_ack, self._put_short_timeout)
packet = self._get_packet(self._RESULT_HEADER_PACKET_MAGIC, self.__in_result_header_buf, self._get_short_timeout)
if packet is not None:
in_result_data_buf = self._get_packet_pre_alloc(struct.unpack("<I", packet)[0])
self.put_bytes(self.__out_result_data_ack, self._put_short_timeout)
dat_packet = self._get_packet(self._RESULT_DATA_PACKET_MAGIC, in_result_data_buf, self._get_long_timeout)
if dat_packet is not None:
return dat_packet
# Avoid timeout livelocking.
self._put_short_timeout = min((self._put_short_timeout * 4) // 3, timeout)
self._get_short_timeout = min((self._get_short_timeout * 4) // 3, timeout)
return None
def call(self, name, data=bytes(), send_timeout=1000, recv_timeout=1000): # public
return self.__get_result(recv_timeout) if self.__put_command(self._hash(name, len(name)), data, send_timeout) else None
class rpc_slave(rpc):
def __init__(self): # private
self.__dict = {}
self.__schedule_cb = None
self.__loop_cb = None
rpc.__init__(self)
self.__in_command_header_buf = self._get_packet_pre_alloc(8)
self.__out_command_header_ack = self._set_packet(self._COMMAND_HEADER_PACKET_MAGIC)
self.__out_command_data_ack = self._set_packet(self._COMMAND_DATA_PACKET_MAGIC)
self.__in_response_header_buf = self._get_packet_pre_alloc()
self.__in_response_data_buf = self._get_packet_pre_alloc()
self._put_short_timeout_reset = 2
self._get_short_timeout_reset = 2
self._put_long_timeout = 5000
self._get_long_timeout = 5000
def __get_command(self, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
start = int(time.time() * 1000)
while (int(time.time() * 1000) - start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_command_header_buf[0], len(self.__in_command_header_buf[0]))
self._flush()
packet = self._get_packet(self._COMMAND_HEADER_PACKET_MAGIC, self.__in_command_header_buf, self._get_short_timeout)
if packet is not None:
command, datalen = struct.unpack("<II", packet)
in_command_data_buf = self._get_packet_pre_alloc(datalen)
self.put_bytes(self.__out_command_header_ack, self._put_short_timeout)
dat_packet = self._get_packet(self._COMMAND_DATA_PACKET_MAGIC, in_command_data_buf, self._get_long_timeout)
if dat_packet is not None:
self.put_bytes(self.__out_command_data_ack, self._put_short_timeout)
return (command, dat_packet)
# Avoid timeout livelocking.
self._put_short_timeout = min(self._put_short_timeout + 1, timeout)
self._get_short_timeout = min(self._get_short_timeout + 1, timeout)
return (None, None)
def __put_result(self, data, timeout): # private
self._put_short_timeout = self._put_short_timeout_reset
self._get_short_timeout = self._get_short_timeout_reset
out_header = self._set_packet(self._RESULT_HEADER_PACKET_MAGIC, struct.pack("<I", len(data)))
out_data = self._set_packet(self._RESULT_DATA_PACKET_MAGIC, data)
start = int(time.time() * 1000)
while (int(time.time() * 1000) - start) < timeout:
gc.collect() # Avoid collection during the transfer.
self._zero(self.__in_response_header_buf[0], len(self.__in_response_header_buf[0]))
self._zero(self.__in_response_data_buf[0], len(self.__in_response_data_buf[0]))
self._flush()
if self._get_packet(self._RESULT_HEADER_PACKET_MAGIC, self.__in_response_header_buf, self._get_short_timeout) is not None:
self.put_bytes(out_header, self._put_short_timeout)
if self._get_packet(self._RESULT_DATA_PACKET_MAGIC, self.__in_response_data_buf, self._get_short_timeout) is not None:
self.put_bytes(out_data, self._put_long_timeout)
return True
# Avoid timeout livelocking.
self._put_short_timeout = min(self._put_short_timeout + 1, timeout)
self._get_short_timeout = min(self._get_short_timeout + 1, timeout)
return False
def register_callback(self, cb): # public
self.__dict[self._hash(cb.__name__, len(cb.__name__))] = cb
def schedule_callback(self, cb): # public
self.__schedule_cb = cb
def setup_loop_callback(self, cb): # public
self.__loop_cb = cb
def loop(self, recv_timeout=1000, send_timeout=1000): # public
while True:
command, data = self.__get_command(recv_timeout)
if command is not None:
cb = self.__dict.get(command)
if self.__put_result(cb(data) if cb is not None else bytes(), send_timeout) and self.__schedule_cb is not None:
self.__schedule_cb()
self.__schedule_cb = None
if self.__loop_cb is not None: self.__loop_cb()
class rpc_usb_vcp_master(rpc_master):
# We need to do reads this way so that we get a short timeout while waiting for data and then
# no timeout while data is coming in. pyserial inter_byte_timeout does not work.
def __get_bytes(self, buff):
i = 0
l = len(buff)
while l:
data = self.__ser.read(min(l, 1024)) # Starts a new timeout per call.
data_len = len(data)
if not data_len: return None
buff[i:i+data_len] = data
i += data_len
l -= data_len
return buff
def __init__(self, port): # private
self.__ser = serial.Serial(port, baudrate=115200, timeout=0.01)
rpc_master.__init__(self)
def _flush(self): # protected
self.__ser.reset_input_buffer()
def get_bytes(self, buff, timeout_ms): # protected
if int(self.__ser.timeout * 100) != 1: self.__ser.timeout = 0.01 # Changing this causes control transfers.
result = self.__get_bytes(buff)
if result is None: time.sleep(timeout_ms * 0.001)
return result
def put_bytes(self, data, timeout_ms): # protected
self.__ser.write(data)
def _stream_get_bytes(self, buff, timeout_ms): # protected
if int(self.__ser.timeout) != 1: self.__ser.timeout = 1 # Changing this causes control transfers.
return self.__get_bytes(buff)
class rpc_usb_vcp_slave(rpc_slave):
# We need to do reads this way so that we get a short timeout while waiting for data and then
# no timeout while data is coming in. pyserial inter_byte_timeout does not work.
def __get_bytes(self, buff):
i = 0
l = len(buff)
while l:
data = self.__ser.read(min(l, 1024)) # Starts a new timeout per call.
data_len = len(data)
if not data_len: return None
buff[i:i+data_len] = data
i += data_len
l -= data_len
return buff
def __init__(self, port): # private
self.__ser = serial.Serial(port, baudrate=115200, timeout=0.01)
rpc_slave.__init__(self)
def _flush(self): # protected
self.__ser.reset_input_buffer()
def get_bytes(self, buff, timeout_ms): # protected
if int(self.__ser.timeout * 100) != 1: self.__ser.timeout = 0.01 # Changing this causes control transfers.
return self.__get_bytes(buff)
def put_bytes(self, data, timeout_ms): # protected
self.__ser.write(data)
def _stream_get_bytes(self, buff, timeout_ms): # protected
if int(self.__ser.timeout) != 1: self.__ser.timeout = 1 # Changing this causes control transfers.
return self.__get_bytes(buff)
class rpc_wifi_or_ethernet_master(rpc_master):
def __valid_tcp_socket(self): # private
if self.__tcp__socket is None:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(self.__myaddr)
s.listen(0)
s.settimeout(1)
self.__tcp__socket, addr = s.accept()
s.close()
except (socket.timeout, socket.error): self.__tcp__socket = None
return self.__tcp__socket is not None
def __close_tcp_socket(self): # private
self.__tcp__socket.close()
self.__tcp__socket = None
def __valid_udp_socket(self): # private
if self.__udp__socket is None:
try:
self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__udp__socket.bind(self.__myaddr)
except (socket.timeout, socket.error): self.__udp__socket = None
return self.__udp__socket is not None
def __close_udp_socket(self): # private
self.__udp__socket.close()
self.__udp__socket = None
def __init__(self, slave_ip, my_ip="", port=0x1DBA): # private
self._udp_limit = 1400
self._timeout_scale = 10
self.__myip = my_ip
self.__myaddr = (self.__myip, port)
self.__slave_addr = (slave_ip, port)
self.__tcp__socket = None
self.__udp__socket = None
print("IP Address:Port %s:%d\nRunning..." % self.__myaddr)
rpc_master.__init__(self)
def _flush(self): # protected
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(0.001)
while(True):
data, addr = self.__udp__socket.recvfrom(1400)
if not len(data): break
except socket.timeout: pass
except socket.error as err: self.__close_udp_socket()
if self.__tcp__socket is not None:
try:
self.__tcp__socket.settimeout(0.001)
while(True):
data = self.__tcp__socket.recvfrom(1400)
if not len(data): break
except socket.timeout: pass
except socket.error as err: self.__close_tcp_socket()
def get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale)
while l:
data, addr = self.__udp__socket.recvfrom(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
# We don't need to close the socket on error since it's connectionless.
except socket.timeout: pass
except socket.error: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(100)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
return buff if not l else None
def put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale)
while l:
data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__slave_addr)
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_udp_socket()
except (socket.timeout, socket.error): self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(100)
while l:
data_len = self.__tcp_socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
def _stream_get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
return buff if not l else None
def _stream_put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
if l: raise OSError # Stop Stream.
class rpc_wifi_or_ethernet_slave(rpc_slave):
def __valid_tcp_socket(self): # private
if self.__tcp__socket is None:
try:
self.__tcp__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.__tcp__socket.connect(self.__master_addr)
except (socket.timeout, socket.error): self.__tcp__socket = None
return self.__tcp__socket is not None
def __close_tcp_socket(self): # private
self.__tcp__socket.close()
self.__tcp__socket = None
def __valid_udp_socket(self): # private
if self.__udp__socket is None:
try:
self.__udp__socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.__udp__socket.bind(self.__myaddr)
except (socket.timeout, socket.error): self.__udp__socket = None
return self.__udp__socket is not None
def __close_udp_socket(self): # private
self.__udp__socket.close()
self.__udp__socket = None
def __init__(self, my_ip="", port=0x1DBA): # private
self._udp_limit = 1400
self._timeout_scale = 10
self.__myip = my_ip
self.__myaddr = (self.__myip, port)
self.__master_addr = None
self.__tcp__socket = None
self.__udp__socket = None
print("IP Address:Port %s:%d\nRunning..." % self.__myaddr)
rpc_slave.__init__(self)
def _flush(self): # protected
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(0.001)
while(True):
data, addr = self.__udp__socket.recvfrom(1400)
if not len(data): break
except socket.timeout: pass
except socket.error: self.__close_udp_socket()
if self.__tcp__socket is not None:
try:
self.__tcp__socket.settimeout(0.001)
while(True):
data = self.__tcp__socket.recvfrom(1400)
if not len(data): break
except socket.timeout: pass
except socket.error: self.__close_tcp_socket()
def get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._get_short_timeout * 0.001 * self._timeout_scale)
while l:
data, addr = self.__udp__socket.recvfrom(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
self.__master_addr = addr
i += data_len
l -= data_len
# We don't need to close the socket on error since it's connectionless.
except socket.timeout: pass
except socket.error: self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(100)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
return buff if not l else None
def put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if l <= self._udp_limit:
if self.__valid_udp_socket():
try:
self.__udp__socket.settimeout(self._put_short_timeout * 0.001 * self._timeout_scale)
while l:
data_len = self.__udp__socket.sendto(data[i:i+min(l, 1400)], self.__master_addr)
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_udp_socket()
except (socket.timeout, socket.error): self.__close_udp_socket()
elif self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(100)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
def _stream_get_bytes(self, buff, timeout_ms): # protected
i = 0
l = len(buff)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data = self.__tcp__socket.recv(min(l, 1400))
data_len = len(data)
if not data_len: break
buff[i:i+data_len] = data
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
return buff if not l else None
def _stream_put_bytes(self, data, timeout_ms): # protected
i = 0
l = len(data)
if self.__valid_tcp_socket():
try:
self.__tcp__socket.settimeout(timeout_ms * 0.001)
while l:
data_len = self.__tcp__socket.send(data[i:i+min(l, 1400)])
if not data_len: break
i += data_len
l -= data_len
if l: self.__close_tcp_socket()
except (socket.timeout, socket.error): self.__close_tcp_socket()
if l: raise OSError # Stop Stream.

View File

@ -0,0 +1,123 @@
# Image Transfer - As The Controller Device
#
# This script is meant to talk to the "image_transfer_jpg_as_the_remote_device_for_your_computer.py" on the OpenMV Cam.
#
# This script shows off how to transfer the frame buffer to your computer as a jpeg image.
import io, pygame, rpc, serial, serial.tools.list_ports, socket, struct, sys
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to control over USB or WIFI.
##############################################################
# Choose the interface you wish to control an OpenMV Cam over.
##############################################################
# Uncomment the below lines to setup your OpenMV Cam for controlling over a USB VCP.
#
# * port - Serial Port Name.
#
print("\nAvailable Ports:\n")
for port, desc, hwid in serial.tools.list_ports.comports():
print("{} : {} [{}]".format(port, desc, hwid))
sys.stdout.write("\nPlease enter a port name: ")
sys.stdout.flush()
interface = rpc.rpc_usb_vcp_master(port=raw_input())
print("")
sys.stdout.flush()
# Uncomment the below line to setup your OpenMV Cam for controlling over WiFi.
#
# * slave_ip - IP address to connect to.
# * my_ip - IP address to bind to ("" to bind to all interfaces...)
# * port - Port to route traffic to.
#
# interface = rpc.rpc_wifi_or_ethernet_master(slave_ip="xxx.xxx.xxx.xxx", my_ip="", port=0x1DBA)
##############################################################
# Call Back Handlers
##############################################################
def get_frame_buffer_call_back(pixformat_str, framesize_str, cutthrough, silent):
if not silent: print("Getting Remote Frame...")
result = interface.call("jpeg_image_snapshot", "%s,%s" % (pixformat_str, framesize_str))
if result is not None:
size = struct.unpack("<I", result)[0]
img = bytearray(size)
if cutthrough:
# Fast cutthrough data transfer with no error checking.
# Before starting the cut through data transfer we need to sync both the master and the
# slave device. On return both devices are in sync.
result = interface.call("jpeg_image_read")
if result is not None:
# GET BYTES NEEDS TO EXECUTE NEXT IMMEDIATELY WITH LITTLE DELAY NEXT.
# Read all the image data in one very large transfer.
interface.get_bytes(img, 5000) # timeout
else:
# Slower data transfer with error checking.
# Transfer 32 KB chunks.
chunk_size = (1 << 15)
if not silent: print("Reading %d bytes..." % size)
for i in range(0, size, chunk_size):
ok = False
for j in range(3): # Try up to 3 times.
result = interface.call("jpeg_image_read", struct.pack("<II", i, chunk_size))
if result is not None:
img[i:i+chunk_size] = result # Write the image data.
if not silent: print("%.2f%%" % ((i * 100) / size))
ok = True
break
if not silent: print("Retrying... %d/2" % (j + 1))
if not ok:
if not silent: print("Error!")
return None
return img
else:
if not silent: print("Failed to get Remote Frame!")
return None
pygame.init()
screen_w = 640
screen_h = 480
screen = pygame.display.set_mode((screen_w, screen_h), flags=pygame.RESIZABLE)
pygame.display.set_caption("Frame Buffer")
clock = pygame.time.Clock()
while(True):
sys.stdout.flush()
# You may change the pixformat and the framesize of the image transfered from the remote device
# by modifying the below arguments.
#
# When cutthrough is False the image will be transferred through the RPC library with CRC and
# retry protection on all data moved. For faster data transfer set cutthrough to True so that
# get_bytes() and put_bytes() are called after an RPC call completes to transfer data
# more quicly from one image buffer to another. Note: This works because once an RPC call
# completes successfully both the master and slave devices are synchronized completely.
#
img = get_frame_buffer_call_back("sensor.RGB565", "sensor.QQVGA", cutthrough=True, silent=True)
if img is not None:
try:
screen.blit(pygame.transform.scale(pygame.image.load(io.BytesIO(img), "jpg"), (screen_w, screen_h)), (0, 0))
pygame.display.update()
clock.tick()
except pygame.error: pass
print(clock.get_fps())
for event in pygame.event.get():
if event.type == pygame.QUIT:
pygame.quit()
quit()

View File

@ -0,0 +1,79 @@
# Image Transfer - As The Controller Device
#
# This script is meant to talk to the "image_transfer_jpg_streaming_as_the_remote_device_for_your_computer.py" on the OpenMV Cam.
#
# This script shows off how to transfer the frame buffer to your computer as a jpeg image.
import io, pygame, rpc, serial, serial.tools.list_ports, socket, sys
# The RPC library above is installed on your OpenMV Cam and provides mutliple classes for
# allowing your OpenMV Cam to control over USB or WIFI.
##############################################################
# Choose the interface you wish to control an OpenMV Cam over.
##############################################################
# Uncomment the below lines to setup your OpenMV Cam for controlling over a USB VCP.
#
# * port - Serial Port Name.
#
print("\nAvailable Ports:\n")
for port, desc, hwid in serial.tools.list_ports.comports():
print("{} : {} [{}]".format(port, desc, hwid))
sys.stdout.write("\nPlease enter a port name: ")
sys.stdout.flush()
interface = rpc.rpc_usb_vcp_master(port=raw_input())
print("")
sys.stdout.flush()
# Uncomment the below line to setup your OpenMV Cam for controlling over WiFi.
#
# * slave_ip - IP address to connect to.
# * my_ip - IP address to bind to ("" to bind to all interfaces...)
# * port - Port to route traffic to.
#
# interface = rpc.rpc_wifi_or_ethernet_master(slave_ip="xxx.xxx.xxx.xxx", my_ip="", port=0x1DBA)
##############################################################
# Call Back Handlers
##############################################################
pygame.init()
screen_w = 640
screen_h = 480
screen = pygame.display.set_mode((screen_w, screen_h), flags=pygame.RESIZABLE)
pygame.display.set_caption("Frame Buffer")
clock = pygame.time.Clock()
# This will be called with the bytes() object generated by the slave device.
def jpg_frame_buffer_cb(data):
sys.stdout.flush()
try:
screen.blit(pygame.transform.scale(pygame.image.load(io.BytesIO(data), "jpg"), (screen_w, screen_h)), (0, 0))
pygame.display.update()
clock.tick()
except pygame.error: pass
print(clock.get_fps())
for event in pygame.event.get():
if event.type == pygame.QUIT:
pygame.quit()
quit()
while(True):
sys.stdout.flush()
# You may change the pixformat and the framesize of the image transfered from the remote device
# by modifying the below arguments.
result = interface.call("jpeg_image_stream", "sensor.RGB565,sensor.QQVGA")
if result is not None:
# THE REMOTE DEVICE WILL START STREAMING ON SUCCESS. SO, WE NEED TO RECEIVE DATA IMMEDIATELY.
interface.stream_reader(jpg_frame_buffer_cb, queue_depth=8)
for event in pygame.event.get():
if event.type == pygame.QUIT:
pygame.quit()
quit()

View File

@ -0,0 +1,129 @@
# Remote Control - As The Controller Device
#
# This script remotely controls an OpenMV Cam using the RPC library.
#
# This script is meant to talk to the "popular_features_as_the_remote_device.py" on the OpenMV Cam.
import json, rpc, serial, serial.tools.list_ports, struct, sys
from datetime import datetime
##############################################################
# Choose the interface you wish to control an OpenMV Cam over.
##############################################################
# Uncomment the below lines to setup your OpenMV Cam for controlling over a USB VCP.
#
# * port - Serial Port Name.
#
print("\nAvailable Ports:\n")
for port, desc, hwid in serial.tools.list_ports.comports():
print("{} : {} [{}]".format(port, desc, hwid))
sys.stdout.write("\nPlease enter a port name: ")
sys.stdout.flush()
interface = rpc.rpc_usb_vcp_master(port=raw_input())
print("")
sys.stdout.flush()
# Uncomment the below line to setup your OpenMV Cam for controlling over WiFi.
#
# * slave_ip - IP address to connect to.
# * my_ip - IP address to bind to ("" to bind to all interfaces...)
# * port - Port to route traffic to.
#
# interface = rpc.rpc_wifi_or_ethernet_master(slave_ip="xxx.xxx.xxx.xxx", my_ip="", port=0x1DBA)
##############################################################
# Call Back Handlers
##############################################################
def exe_face_detection():
result = interface.call("face_detection")
if result is not None and len(result):
print("Largest Face Detected [x=%d, y=%d, w=%d, h=%d]" % struct.unpack("<HHHH", result))
def exe_person_detection():
result = interface.call("person_detection")
if result is not None:
print(result.tobytes())
def exe_qrcode_detection():
result = interface.call("qrcode_detection")
if result is not None and len(result):
print(result.tobytes())
def exe_all_qrcode_detection():
result = interface.call("all_qrcode_detection")
if result is not None and len(result):
print("QR Codes Detected:")
for obj in json.loads(result.tobytes()):
print(obj)
def exe_apriltag_detection():
result = interface.call("apriltag_detection")
if result is not None and len(result):
print("Largest Tag Detected [cx=%d, cy=%d, id=%d, rot=%d]" % struct.unpack("<HHHH",result))
def exe_all_apriltag_detection():
result = interface.call("all_apriltag_detection")
if result is not None and len(result):
print("Tags Detected:")
for obj in json.loads(result.tobytes()):
print(obj)
def exe_datamatrix_detection():
result = interface.call("datamatrix_detection")
if result is not None and len(result):
print(result.tobytes())
def exe_all_datamatrix_detection():
result = interface.call("all_datamatrix_detection")
if result is not None and len(result):
print("Data Matrices Detected:")
for obj in json.loads(result.tobytes()):
print(obj)
def exe_barcode_detection():
result = interface.call("barcode_detection")
if result is not None and len(result):
print(result.tobytes())
def exe_all_barcode_detection():
result = interface.call("all_barcode_detection")
if result is not None and len(result):
print("Bar Codes Detected:")
for obj in json.loads(result.tobytes()):
print(obj)
def exe_color_detection():
thresholds = (30, 100, 15, 127, 15, 127) # generic red thresholds
# thresholds = (30, 100, -64, -8, -32, 32) # generic green thresholds
# thresholds = (0, 30, 0, 64, -128, 0) # generic blue thresholds
result = interface.call("color_detection", struct.pack("<bbbbbb", *thresholds))
if result is not None and len(result):
print("Largest Color Detected [cx=%d, cy=%d]" % struct.unpack("<HH", result))
def exe_jpeg_snapshot():
result = interface.call("jpeg_snapshot")
if result is not None:
name = "snapshot-%s.jpg" % datetime.now().strftime("%d.%m.%Y-%H.%M.%S")
print("Writing jpeg %s..." % name)
with open(name, "wb") as snap:
snap.write(result)
# Execute remote functions in a loop. Please choose and uncomment one remote function below.
# Executing multiple at a time may run slowly if the camera needs to change camera modes
# per execution.
while(True):
exe_face_detection() # Face should be about 2ft away.
# exe_person_detection()
# exe_qrcode_detection() # Place the QRCode about 2ft away.
# exe_all_qrcode_detection() # Place the QRCode about 2ft away.
# exe_apriltag_detection()
# exe_all_apriltag_detection()
# exe_datamatrix_detection() # Place the Datamatrix about 2ft away.
# exe_all_datamatrix_detection() # Place the Datamatrix about 2ft away.
# exe_barcode_detection() # Place the Barcode about 2ft away.
# exe_all_barcode_detection() # Place the Barcode about 2ft away.
# exe_color_detection()
# exe_jpeg_snapshot()