mirror of
https://github.com/openmv/openmv.git
synced 2025-09-26 23:09:13 +08:00
tools/pyopenmv: Remove obsolete tools.
Signed-off-by: iabdalkader <i.abdalkader@gmail.com>
This commit is contained in:
parent
64a2016a2f
commit
af6fb16cc6
@ -1,305 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# This file is part of the OpenMV project.
|
||||
#
|
||||
# Copyright (c) 2013-2021 Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Copyright (c) 2013-2021 Kwabena W. Agyeman <kwagyeman@openmv.io>
|
||||
#
|
||||
# This work is licensed under the MIT license, see the file LICENSE for details.
|
||||
#
|
||||
# Openmv module.
|
||||
|
||||
import struct
|
||||
import sys,time
|
||||
import serial
|
||||
import platform
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
__serial = None
|
||||
__FB_HDR_SIZE = 12
|
||||
|
||||
# USB Debug commands
|
||||
__USBDBG_CMD = 48
|
||||
__USBDBG_FW_VERSION = 0x80
|
||||
__USBDBG_FRAME_SIZE = 0x81
|
||||
__USBDBG_FRAME_DUMP = 0x82
|
||||
__USBDBG_ARCH_STR = 0x83
|
||||
__USBDBG_SCRIPT_EXEC = 0x05
|
||||
__USBDBG_SCRIPT_STOP = 0x06
|
||||
__USBDBG_SCRIPT_SAVE = 0x07
|
||||
__USBDBG_SCRIPT_RUNNING = 0x87
|
||||
__USBDBG_TEMPLATE_SAVE = 0x08
|
||||
__USBDBG_DESCRIPTOR_SAVE= 0x09
|
||||
__USBDBG_ATTR_READ = 0x8A
|
||||
__USBDBG_ATTR_WRITE = 0x0B
|
||||
__USBDBG_SYS_RESET = 0x0C
|
||||
__USBDBG_SYS_RESET_TO_BL= 0x0E
|
||||
__USBDBG_FB_ENABLE = 0x0D
|
||||
__USBDBG_TX_BUF_LEN = 0x8E
|
||||
__USBDBG_TX_BUF = 0x8F
|
||||
__USBDBG_GET_STATE = 0x93
|
||||
__USBDBG_PROFILE_SIZE = 0x94
|
||||
__USBDBG_PROFILE_DUMP = 0x95
|
||||
__USBDBG_PROFILE_MODE = 0x16
|
||||
__USBDBG_PROFILE_EVENT = 0x17
|
||||
__USBDBG_PROFILE_RESET = 0x18
|
||||
|
||||
__USBDBG_FLAG_SCRIPT_RUNNING = (1 << 0)
|
||||
__USBDBG_FLAG_TEXTBUF_NOTEMPTY = (1 << 1)
|
||||
__USBDBG_FLAG_FRAMEBUF_LOCKED = (1 << 2)
|
||||
__USBDBG_FLAG_PROFILE_ENABLED = (1 << 3)
|
||||
__USBDBG_FLAG_PROFILE_HAS_PMU = (1 << 4)
|
||||
|
||||
ATTR_CONTRAST = 0
|
||||
ATTR_BRIGHTNESS = 1
|
||||
ATTR_SATURATION = 2
|
||||
ATTR_GAINCEILING = 3
|
||||
|
||||
__BOOTLDR_START = 0xABCD0001
|
||||
__BOOTLDR_RESET = 0xABCD0002
|
||||
__BOOTLDR_ERASE = 0xABCD0004
|
||||
__BOOTLDR_WRITE = 0xABCD0008
|
||||
|
||||
def init(port, baudrate=921600, timeout=0.3):
|
||||
global __serial
|
||||
# open CDC port
|
||||
__serial = serial.Serial(port, baudrate=baudrate, timeout=timeout)
|
||||
|
||||
def disconnect():
|
||||
global __serial
|
||||
try:
|
||||
if (__serial):
|
||||
__serial.close()
|
||||
__serial = None
|
||||
except:
|
||||
pass
|
||||
|
||||
def write_pack(fmt, *values):
|
||||
__serial.write(struct.pack(fmt, *values))
|
||||
|
||||
def read_unpack(fmt):
|
||||
return struct.unpack(fmt, __serial.read(struct.calcsize(fmt)))
|
||||
|
||||
def set_timeout(timeout):
|
||||
__serial.timeout = timeout
|
||||
|
||||
def fb_size():
|
||||
# read fb header
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_FRAME_SIZE, __FB_HDR_SIZE)
|
||||
return read_unpack("III")
|
||||
|
||||
def read_state():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_GET_STATE, 64)
|
||||
flags, w, h, size, text_buf = read_unpack("IIII48s")
|
||||
|
||||
text = None
|
||||
profile = False
|
||||
|
||||
if flags & __USBDBG_FLAG_PROFILE_ENABLED:
|
||||
profile = True
|
||||
|
||||
if flags & __USBDBG_FLAG_TEXTBUF_NOTEMPTY:
|
||||
text = text_buf.split(b'\0', 1)[0].decode()
|
||||
|
||||
if flags & __USBDBG_FLAG_FRAMEBUF_LOCKED == 0:
|
||||
return 0, 0, None, 0, text, "", profile
|
||||
|
||||
num_bytes = size if size > 2 else (w * h * size)
|
||||
|
||||
# read fb data
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_FRAME_DUMP, num_bytes)
|
||||
buff = __serial.read(num_bytes)
|
||||
|
||||
if size == 1: # Grayscale
|
||||
fmt = "GRAY"
|
||||
y = np.fromstring(buff, dtype=np.uint8)
|
||||
buff = np.column_stack((y, y, y))
|
||||
elif size == 2: # RGB565
|
||||
fmt = "RGB"
|
||||
arr = np.fromstring(buff, dtype=np.uint16)
|
||||
r = (((arr & 0xF800) >>11)*255.0/31.0).astype(np.uint8)
|
||||
g = (((arr & 0x07E0) >>5) *255.0/63.0).astype(np.uint8)
|
||||
b = (((arr & 0x001F) >>0) *255.0/31.0).astype(np.uint8)
|
||||
buff = np.column_stack((r,g,b))
|
||||
else: # JPEG
|
||||
fmt = "JPEG"
|
||||
try:
|
||||
buff = np.asarray(Image.frombuffer("RGB", (w, h), buff, "jpeg", "RGB", ""))
|
||||
except Exception as e:
|
||||
raise ValueError(f"JPEG decode error (%e)")
|
||||
|
||||
if (buff.size != (w*h*3)):
|
||||
raise ValueError(f"Unexpected frame size. Expected: {w*h*3} received: {buff.size}")
|
||||
|
||||
return w, h, buff.reshape((h, w, 3)), num_bytes, text, fmt, profile
|
||||
|
||||
|
||||
def fb_dump():
|
||||
size = fb_size()
|
||||
|
||||
if (not size[0]):
|
||||
# frame not ready
|
||||
return None
|
||||
|
||||
if (size[2] > 2): #JPEG
|
||||
num_bytes = size[2]
|
||||
else:
|
||||
num_bytes = size[0]*size[1]*size[2]
|
||||
|
||||
# read fb data
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_FRAME_DUMP, num_bytes)
|
||||
buff = __serial.read(num_bytes)
|
||||
|
||||
if size[2] == 1: # Grayscale
|
||||
y = np.fromstring(buff, dtype=np.uint8)
|
||||
buff = np.column_stack((y, y, y))
|
||||
elif size[2] == 2: # RGB565
|
||||
arr = np.fromstring(buff, dtype=np.uint16)
|
||||
r = (((arr & 0xF800) >>11)*255.0/31.0).astype(np.uint8)
|
||||
g = (((arr & 0x07E0) >>5) *255.0/63.0).astype(np.uint8)
|
||||
b = (((arr & 0x001F) >>0) *255.0/31.0).astype(np.uint8)
|
||||
buff = np.column_stack((r,g,b))
|
||||
else: # JPEG
|
||||
try:
|
||||
buff = np.asarray(Image.frombuffer("RGB", size[0:2], buff, "jpeg", "RGB", ""))
|
||||
except Exception as e:
|
||||
print ("JPEG decode error (%s)"%(e))
|
||||
return None
|
||||
|
||||
if (buff.size != (size[0]*size[1]*3)):
|
||||
return None
|
||||
|
||||
return (size[0], size[1], buff.reshape((size[1], size[0], 3)))
|
||||
|
||||
def read_profile():
|
||||
records = []
|
||||
|
||||
# Read and unpack profiling data size
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_PROFILE_SIZE, 12)
|
||||
record_count, record_size, event_count = read_unpack("<III")
|
||||
|
||||
if record_count:
|
||||
offset = 0
|
||||
record_format = f"<5I2Q{event_count}QI"
|
||||
profile_size = record_count * record_size
|
||||
|
||||
# Read profiling data
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_PROFILE_DUMP, profile_size)
|
||||
profile_data = __serial.read(profile_size)
|
||||
|
||||
for i in range(record_count):
|
||||
# Unpack the record
|
||||
profile = struct.unpack(record_format, profile_data[offset:offset + record_size])
|
||||
|
||||
# Parse the profile data
|
||||
records.append({
|
||||
'address': profile[0],
|
||||
'caller': profile[1],
|
||||
'call_count': profile[2],
|
||||
'min_ticks': profile[3],
|
||||
'max_ticks': profile[4],
|
||||
'total_ticks': profile[5],
|
||||
'total_cycles': profile[6],
|
||||
'events': profile[7:7 + event_count] if event_count > 0 else []
|
||||
})
|
||||
|
||||
offset += record_size
|
||||
|
||||
return records
|
||||
|
||||
def set_profile_mode(mode):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_PROFILE_MODE, 4)
|
||||
write_pack("<I", mode)
|
||||
|
||||
def set_event_counter(event_num, event_type):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_PROFILE_EVENT, 8)
|
||||
write_pack("<II", event_num, event_type)
|
||||
|
||||
def reset_profiler():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_PROFILE_RESET, 0)
|
||||
|
||||
def exec_script(buf):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_EXEC, len(buf))
|
||||
__serial.write(buf.encode())
|
||||
|
||||
def stop_script():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_STOP, 0)
|
||||
|
||||
def script_running():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_RUNNING, 4)
|
||||
return read_unpack("I")[0]
|
||||
|
||||
def save_template(x, y, w, h, path):
|
||||
buf = struct.pack("IIII", x, y, w, h) + path
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_TEMPLATE_SAVE, len(buf))
|
||||
__serial.write(buf)
|
||||
|
||||
def save_descriptor(x, y, w, h, path):
|
||||
buf = struct.pack("HHHH", x, y, w, h) + path
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_DESCRIPTOR_SAVE, len(buf))
|
||||
__serial.write(buf)
|
||||
|
||||
def set_attr(attr, value):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_ATTR_WRITE, 8)
|
||||
write_pack("<II", attr, value)
|
||||
|
||||
def get_attr(attr):
|
||||
write_pack("<BBIh", __USBDBG_CMD, __USBDBG_ATTR_READ, 1, attr)
|
||||
return __serial.read(1)
|
||||
|
||||
def reset():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_SYS_RESET, 0)
|
||||
|
||||
def reset_to_bl():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_SYS_RESET_TO_BL, 0)
|
||||
|
||||
def bootloader_start():
|
||||
write_pack("<I", __BOOTLDR_START)
|
||||
return read_unpack("I")[0] == __BOOTLDR_START
|
||||
|
||||
def bootloader_reset():
|
||||
write_pack("<I", __BOOTLDR_RESET)
|
||||
|
||||
def flash_erase(sector):
|
||||
write_pack("<II", __BOOTLDR_ERASE, sector)
|
||||
|
||||
def flash_write(buf):
|
||||
write_pack("<I", __BOOTLDR_WRITE) + buf
|
||||
|
||||
def tx_buf_len():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_TX_BUF_LEN, 4)
|
||||
return read_unpack("I")[0]
|
||||
|
||||
def tx_buf(bytes):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_TX_BUF, bytes)
|
||||
return __serial.read(bytes)
|
||||
|
||||
def fw_version():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_FW_VERSION, 12)
|
||||
return read_unpack("III")
|
||||
|
||||
def enable_fb(enable):
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_FB_ENABLE, 4)
|
||||
write_pack("<I", enable)
|
||||
|
||||
def arch_str():
|
||||
write_pack("<BBI", __USBDBG_CMD, __USBDBG_ARCH_STR, 64)
|
||||
return __serial.read(64).split(b'\0', 1)[0]
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv)!= 3:
|
||||
print ('usage: pyopenmv.py <port> <script>')
|
||||
sys.exit(1)
|
||||
|
||||
with open(sys.argv[2], 'r') as fin:
|
||||
buf = fin.read()
|
||||
|
||||
disconnect()
|
||||
init(sys.argv[1])
|
||||
stop_script()
|
||||
exec_script(buf)
|
||||
tx_len = tx_buf_len()
|
||||
time.sleep(0.250)
|
||||
if (tx_len):
|
||||
print(tx_buf(tx_len).decode())
|
||||
disconnect()
|
@ -1,564 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# This file is part of the OpenMV project.
|
||||
#
|
||||
# Copyright (c) 2013-2025 Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Copyright (c) 2013-2025 Kwabena W. Agyeman <kwagyeman@openmv.io>
|
||||
#
|
||||
# This work is licensed under the MIT license, see the file LICENSE for details.
|
||||
#
|
||||
# An example script using pyopenmv to grab the framebuffer.
|
||||
|
||||
import sys
|
||||
import numpy as np
|
||||
import pygame
|
||||
import pyopenmv
|
||||
import argparse
|
||||
import time
|
||||
|
||||
test_script = """
|
||||
import sensor, image, time
|
||||
sensor.reset()
|
||||
sensor.set_pixformat(sensor.RGB565)
|
||||
sensor.set_framesize(sensor.QVGA)
|
||||
sensor.skip_frames(time = 2000)
|
||||
clock = time.clock()
|
||||
|
||||
while(True):
|
||||
clock.tick()
|
||||
img = sensor.snapshot()
|
||||
print(clock.fps(), " FPS")
|
||||
"""
|
||||
|
||||
bench_script = """
|
||||
import sensor, image, time
|
||||
sensor.reset()
|
||||
sensor.set_pixformat(sensor.RGB565)
|
||||
sensor.set_framesize(sensor.VGA)
|
||||
img = sensor.snapshot().compress()
|
||||
while(True):
|
||||
img.flush()
|
||||
"""
|
||||
|
||||
def addr_to_symbol(symbols, address):
|
||||
# Binary search for speed
|
||||
lo, hi = 0, len(symbols) - 1
|
||||
while lo <= hi:
|
||||
mid = (lo + hi) // 2
|
||||
start, end, name = symbols[mid]
|
||||
if start <= address < end:
|
||||
return name
|
||||
elif address < start:
|
||||
hi = mid - 1
|
||||
else:
|
||||
lo = mid + 1
|
||||
return None
|
||||
|
||||
def get_color_by_percentage(percentage, base_color=(220, 220, 220)):
|
||||
"""
|
||||
Return a color based on percentage with fine-grained intensity levels.
|
||||
"""
|
||||
def clamp(value):
|
||||
return max(0, min(255, int(value)))
|
||||
|
||||
if percentage >= 50:
|
||||
# Very high - bright red
|
||||
intensity = min(1.0, (percentage - 50) / 50)
|
||||
return (255, clamp(120 - 120 * intensity), clamp(120 - 120 * intensity))
|
||||
elif percentage >= 30:
|
||||
# High - red-orange
|
||||
intensity = (percentage - 30) / 20
|
||||
return (255, clamp(160 + 40 * intensity), clamp(160 - 40 * intensity))
|
||||
elif percentage >= 20:
|
||||
# Medium-high - orange
|
||||
intensity = (percentage - 20) / 10
|
||||
return (255, clamp(200 + 55 * intensity), clamp(180 - 20 * intensity))
|
||||
elif percentage >= 15:
|
||||
# Medium - yellow-orange
|
||||
intensity = (percentage - 15) / 5
|
||||
return (255, clamp(220 + 35 * intensity), clamp(180 + 20 * intensity))
|
||||
elif percentage >= 10:
|
||||
# Medium-low - yellow
|
||||
intensity = (percentage - 10) / 5
|
||||
return (clamp(255 - 75 * intensity), 255, clamp(180 + 75 * intensity))
|
||||
elif percentage >= 5:
|
||||
# Low - light green
|
||||
intensity = (percentage - 5) / 5
|
||||
return (clamp(180 + 75 * intensity), 255, clamp(180 + 75 * intensity))
|
||||
elif percentage >= 2:
|
||||
# Very low - green
|
||||
intensity = (percentage - 2) / 3
|
||||
return (clamp(160 + 95 * intensity), clamp(255 - 55 * intensity), clamp(160 + 95 * intensity))
|
||||
elif percentage >= 1:
|
||||
# Minimal - light blue-green
|
||||
intensity = (percentage - 1) / 1
|
||||
return (clamp(140 + 120 * intensity), clamp(200 + 55 * intensity), clamp(255 - 95 * intensity))
|
||||
else:
|
||||
# Zero or negligible - base color
|
||||
return base_color
|
||||
|
||||
def draw_rounded_rect(surface, color, rect, radius=5):
|
||||
x, y, w, h = rect
|
||||
if w <= 0 or h <= 0:
|
||||
return
|
||||
pygame.draw.rect(surface, color, (x + radius, y, w - 2*radius, h))
|
||||
pygame.draw.rect(surface, color, (x, y + radius, w, h - 2*radius))
|
||||
pygame.draw.circle(surface, color, (x + radius, y + radius), radius)
|
||||
pygame.draw.circle(surface, color, (x + w - radius, y + radius), radius)
|
||||
pygame.draw.circle(surface, color, (x + radius, y + h - radius), radius)
|
||||
pygame.draw.circle(surface, color, (x + w - radius, y + h - radius), radius)
|
||||
|
||||
|
||||
def draw_table(overlay_surface, config, title, headers, col_widths):
|
||||
"""Draw the common table background, title, and header."""
|
||||
# Draw main table background
|
||||
table_rect = (0, 0, config['width'], config['height'])
|
||||
draw_rounded_rect(overlay_surface, config['colors']['bg'], table_rect, int(8 * config['scale_factor']))
|
||||
pygame.draw.rect(overlay_surface, config['colors']['border'], table_rect, max(1, int(2 * config['scale_factor'])))
|
||||
|
||||
# Table title
|
||||
title_text = config['fonts']['title'].render(title, True, config['colors']['header_text'])
|
||||
title_rect = title_text.get_rect()
|
||||
title_x = (config['width'] - title_rect.width) // 2
|
||||
overlay_surface.blit(title_text, (title_x, int(12 * config['scale_factor'])))
|
||||
|
||||
# Header
|
||||
header_y = int(50 * config['scale_factor'])
|
||||
header_height = int(40 * config['scale_factor'])
|
||||
|
||||
# Draw header background
|
||||
header_rect = (int(5 * config['scale_factor']), header_y,
|
||||
config['width'] - int(10 * config['scale_factor']), header_height)
|
||||
draw_rounded_rect(overlay_surface, config['colors']['header_bg'], header_rect, int(4 * config['scale_factor']))
|
||||
|
||||
# Draw header text and separators
|
||||
current_x = int(10 * config['scale_factor'])
|
||||
for i, (header, width) in enumerate(zip(headers, col_widths)):
|
||||
header_surface = config['fonts']['header'].render(header, True, config['colors']['header_text'])
|
||||
overlay_surface.blit(header_surface, (current_x, header_y + int(6 * config['scale_factor'])))
|
||||
|
||||
if i < len(headers) - 1:
|
||||
sep_x = current_x + width - int(5 * config['scale_factor'])
|
||||
pygame.draw.line(overlay_surface, config['colors']['border'],
|
||||
(sep_x, header_y + 2), (sep_x, header_y + header_height - 2), 1)
|
||||
current_x += width
|
||||
|
||||
|
||||
def draw_event_table(overlay_surface, config, profile_data, profile_mode, symbols):
|
||||
"""Draw the event counter mode table."""
|
||||
|
||||
# Prepare data
|
||||
num_events = len(profile_data[0]['events']) if profile_data else 0
|
||||
if not num_events:
|
||||
sorted_data = sorted(profile_data, key=lambda x: x['address'])
|
||||
else:
|
||||
sort_func = lambda x: x['events'][0] // max(1, x['call_count'])
|
||||
sorted_data = sorted(profile_data, key=sort_func, reverse=True)
|
||||
|
||||
headers = ["Function"] + [f"E{i}" for i in range(num_events)]
|
||||
proportions = [0.30] + [0.70/num_events] * num_events
|
||||
col_widths = [config['width'] * prop for prop in proportions]
|
||||
profile_mode = "Exclusive" if profile_mode else "Inclusive"
|
||||
|
||||
# Calculate event totals for percentage calculation
|
||||
event_totals = [0] * num_events
|
||||
for record in sorted_data:
|
||||
for i, event_count in enumerate(record['events']):
|
||||
event_totals[i] += event_count // max(1, record['call_count'])
|
||||
|
||||
# Draw table structure
|
||||
draw_table(overlay_surface, config, f"Event Counters ({profile_mode})", headers, col_widths)
|
||||
|
||||
# Draw data rows
|
||||
row_height = int(30 * config['scale_factor'])
|
||||
data_start_y = int(50 * config['scale_factor'] + 40 * config['scale_factor'] + 8 * config['scale_factor'])
|
||||
available_height = config['height'] - data_start_y - int(60 * config['scale_factor'])
|
||||
visible_rows = min(len(sorted_data), available_height // row_height)
|
||||
|
||||
for i in range(visible_rows):
|
||||
record = sorted_data[i]
|
||||
row_y = data_start_y + i * row_height
|
||||
|
||||
# Draw row background
|
||||
row_color = config['colors']['row_alt'] if i % 2 == 0 else config['colors']['row_normal']
|
||||
row_rect = (int(5 * config['scale_factor']), row_y,
|
||||
config['width'] - int(10 * config['scale_factor']), row_height)
|
||||
pygame.draw.rect(overlay_surface, row_color, row_rect)
|
||||
|
||||
# Function name
|
||||
name = addr_to_symbol(symbols, record['address']) if symbols else "<no symbols>"
|
||||
max_name_chars = int(col_widths[0] // (11 * config['scale_factor']))
|
||||
display_name = name if len(name) <= max_name_chars else name[:max_name_chars - 3] + "..."
|
||||
|
||||
row_data = [display_name]
|
||||
|
||||
# Event data
|
||||
for j, event_count in enumerate(record['events']):
|
||||
event_scale = ""
|
||||
event_count //= max(1, record['call_count'])
|
||||
if event_count > 1_000_000_000:
|
||||
event_count //= 1_000_000_000
|
||||
event_scale = "B"
|
||||
elif event_count > 1_000_000:
|
||||
event_count //= 1_000_000
|
||||
event_scale = "M"
|
||||
row_data.append(f"{event_count:,}{event_scale}")
|
||||
|
||||
# Determine row color based on sorting key (event 0)
|
||||
if len(record['events']) > 0 and event_totals[0] > 0:
|
||||
sort_key_value = record['events'][0] // max(1, record['call_count'])
|
||||
percentage = (sort_key_value / event_totals[0] * 100)
|
||||
row_text_color = get_color_by_percentage(percentage, config['colors']['content_text'])
|
||||
else:
|
||||
row_text_color = config['colors']['content_text']
|
||||
|
||||
# Draw row data with uniform color
|
||||
current_x = 10
|
||||
for j, (data, width) in enumerate(zip(row_data, col_widths)):
|
||||
text_surface = config['fonts']['content'].render(str(data), True, row_text_color)
|
||||
overlay_surface.blit(text_surface, (current_x, row_y + int(8 * config['scale_factor'])))
|
||||
|
||||
if j < len(row_data) - 1:
|
||||
sep_x = current_x + width - 8
|
||||
pygame.draw.line(overlay_surface, (60, 70, 85),
|
||||
(sep_x, row_y), (sep_x, row_y + row_height), 1)
|
||||
current_x += width
|
||||
|
||||
# Draw summary
|
||||
summary_y = config['height'] - int(50 * config['scale_factor'])
|
||||
total_functions = len(profile_data)
|
||||
grand_total = sum(event_totals)
|
||||
summary_text = (
|
||||
f"Profiles: {total_functions} | "
|
||||
f"Events: {num_events} | "
|
||||
f"Total Events: {grand_total:,}"
|
||||
)
|
||||
|
||||
summary_surface = config['fonts']['summary'].render(summary_text, True, config['colors']['content_text'])
|
||||
summary_rect = summary_surface.get_rect()
|
||||
summary_x = (config['width'] - summary_rect.width) // 2
|
||||
overlay_surface.blit(summary_surface, (summary_x, summary_y))
|
||||
|
||||
# Instructions
|
||||
instruction_str = "Press 'P' to toggle event counter overlay"
|
||||
instruction_text = config['fonts']['instruction'].render(instruction_str, True, (180, 180, 180))
|
||||
overlay_surface.blit(instruction_text, (0, summary_y + int(20 * config['scale_factor'])))
|
||||
|
||||
|
||||
def draw_profile_table(overlay_surface, config, profile_data, profile_mode, symbols):
|
||||
"""Draw the profile mode table."""
|
||||
|
||||
# Prepare data
|
||||
sort_func = lambda x: x['total_ticks']
|
||||
sorted_data = sorted(profile_data, key=sort_func, reverse=True)
|
||||
total_ticks_all = sum(record['total_ticks'] for record in profile_data)
|
||||
profile_mode = "Exclusive" if profile_mode else "Inclusive"
|
||||
|
||||
headers = ["Function", "Calls", "Min", "Max", "Total", "Avg", "Cycles", "%"]
|
||||
proportions = [0.30, 0.08, 0.10, 0.10, 0.13, 0.10, 0.13, 0.05]
|
||||
col_widths = [config['width'] * prop for prop in proportions]
|
||||
|
||||
# Draw table structure
|
||||
draw_table(overlay_surface, config, f"Performance Profile ({profile_mode})", headers, col_widths)
|
||||
|
||||
# Draw data rows
|
||||
row_height = int(30 * config['scale_factor'])
|
||||
data_start_y = int(50 * config['scale_factor'] + 40 * config['scale_factor'] + 8 * config['scale_factor'])
|
||||
available_height = config['height'] - data_start_y - int(60 * config['scale_factor'])
|
||||
visible_rows = min(len(sorted_data), available_height // row_height)
|
||||
|
||||
for i in range(visible_rows):
|
||||
record = sorted_data[i]
|
||||
row_y = data_start_y + i * row_height
|
||||
|
||||
# Draw row background
|
||||
row_color = config['colors']['row_alt'] if i % 2 == 0 else config['colors']['row_normal']
|
||||
row_rect = (int(5 * config['scale_factor']), row_y,
|
||||
config['width'] - int(10 * config['scale_factor']), row_height)
|
||||
pygame.draw.rect(overlay_surface, row_color, row_rect)
|
||||
|
||||
# Function name
|
||||
name = addr_to_symbol(symbols, record['address']) if symbols else "<no symbols>"
|
||||
max_name_chars = int(col_widths[0] // (11 * config['scale_factor']))
|
||||
display_name = name if len(name) <= max_name_chars else name[:max_name_chars - 3] + "..."
|
||||
|
||||
# Calculate values
|
||||
call_count = record['call_count']
|
||||
min_ticks = record['min_ticks'] if call_count else 0
|
||||
max_ticks = record['max_ticks'] if call_count else 0
|
||||
total_ticks = record['total_ticks']
|
||||
avg_cycles = record['total_cycles'] // max(1, call_count)
|
||||
avg_ticks = total_ticks // max(1, call_count)
|
||||
percentage = (total_ticks / total_ticks_all * 100)
|
||||
|
||||
ticks_scale = ""
|
||||
if total_ticks > 1_000_000_000:
|
||||
total_ticks //= 1_000_000
|
||||
ticks_scale = "M"
|
||||
|
||||
row_data = [
|
||||
display_name,
|
||||
f"{call_count:,}",
|
||||
f"{min_ticks:,}",
|
||||
f"{max_ticks:,}",
|
||||
f"{total_ticks:,}{ticks_scale}",
|
||||
f"{avg_ticks:,}",
|
||||
f"{avg_cycles:,}",
|
||||
f"{percentage:.1f}%"
|
||||
]
|
||||
|
||||
# Determine row color based on percentage
|
||||
text_color = get_color_by_percentage(percentage, config['colors']['content_text'])
|
||||
|
||||
# Draw row data
|
||||
current_x = int(10 * config['scale_factor'])
|
||||
for j, (data, width) in enumerate(zip(row_data, col_widths)):
|
||||
text_surface = config['fonts']['content'].render(str(data), True, text_color)
|
||||
overlay_surface.blit(text_surface, (current_x, row_y + int(8 * config['scale_factor'])))
|
||||
|
||||
if j < len(row_data) - 1:
|
||||
sep_x = current_x + width - int(8 * config['scale_factor'])
|
||||
pygame.draw.line(overlay_surface, (60, 70, 85),
|
||||
(sep_x, row_y), (sep_x, row_y + row_height), 1)
|
||||
current_x += width
|
||||
|
||||
# Draw summary
|
||||
summary_y = config['height'] - int(50 * config['scale_factor'])
|
||||
total_calls = sum(record['call_count'] for record in profile_data)
|
||||
total_cycles = sum(record['total_cycles'] for record in profile_data)
|
||||
total_ticks_summary = sum(record['total_ticks'] for record in profile_data)
|
||||
|
||||
summary_text = (
|
||||
f"Profiles: {len(profile_data)} | "
|
||||
f"Total Calls: {total_calls:,} | "
|
||||
f"Total Ticks: {total_ticks_summary:,} | "
|
||||
f"Total Cycles: {total_cycles:,}"
|
||||
)
|
||||
|
||||
summary_surface = config['fonts']['summary'].render(summary_text, True, config['colors']['content_text'])
|
||||
summary_rect = summary_surface.get_rect()
|
||||
summary_x = (config['width'] - summary_rect.width) // 2
|
||||
overlay_surface.blit(summary_surface, (summary_x, summary_y))
|
||||
|
||||
# Instructions
|
||||
instruction_str = "Press 'P' to toggle event counter overlay"
|
||||
instruction_text = config['fonts']['instruction'].render(instruction_str, True, (180, 180, 180))
|
||||
overlay_surface.blit(instruction_text, (0, summary_y + int(20 * config['scale_factor'])))
|
||||
|
||||
def draw_profile_overlay(screen, screen_width, screen_height, profile_data,
|
||||
profile_mode, profile_type, scale, symbols, alpha=250):
|
||||
"""Main entry point for drawing the profile overlay."""
|
||||
# Calculate dimensions and create surface
|
||||
base_width, base_height = 800, 800
|
||||
screen_width *= scale
|
||||
screen_height *= scale
|
||||
scale_factor = min(screen_width / base_width, screen_height / base_height)
|
||||
|
||||
overlay_surface = pygame.Surface((screen_width, screen_height), pygame.SRCALPHA)
|
||||
overlay_surface.set_alpha(alpha)
|
||||
|
||||
# Setup common configuration
|
||||
config = {
|
||||
'width': screen_width,
|
||||
'height': screen_height,
|
||||
'scale_factor': scale_factor,
|
||||
'colors': {
|
||||
'bg': (40, 50, 65),
|
||||
'border': (70, 80, 100),
|
||||
'header_bg': (60, 80, 120),
|
||||
'header_text': (255, 255, 255),
|
||||
'content_text': (220, 220, 220),
|
||||
'row_alt': (35, 45, 60),
|
||||
'row_normal': (45, 55, 70)
|
||||
},
|
||||
'fonts': {
|
||||
'title': pygame.font.SysFont("arial", int(28 * scale_factor), bold=True),
|
||||
'header': pygame.font.SysFont("monospace", int(20 * scale_factor), bold=True),
|
||||
'content': pygame.font.SysFont("monospace", int(18 * scale_factor)),
|
||||
'summary': pygame.font.SysFont("arial", int(20 * scale_factor)),
|
||||
'instruction': pygame.font.SysFont("arial", int(22 * scale_factor))
|
||||
}
|
||||
}
|
||||
|
||||
# Draw based on mode
|
||||
if profile_type == 1:
|
||||
draw_profile_table(overlay_surface, config, profile_data, profile_mode, symbols)
|
||||
elif profile_type == 2:
|
||||
draw_event_table(overlay_surface, config, profile_data, profile_mode, symbols)
|
||||
|
||||
screen.blit(overlay_surface, (0, 0))
|
||||
|
||||
def pygame_test(port, script, poll_rate, scale, benchmark, symbols):
|
||||
# init pygame
|
||||
pygame.init()
|
||||
pyopenmv.disconnect()
|
||||
|
||||
connected = False
|
||||
for i in range(10):
|
||||
try:
|
||||
# opens CDC port.
|
||||
# Set small timeout when connecting
|
||||
pyopenmv.init(port, baudrate=921600, timeout=0.050)
|
||||
connected = True
|
||||
break
|
||||
except Exception as e:
|
||||
connected = False
|
||||
time.sleep(0.100)
|
||||
|
||||
if not connected:
|
||||
print("Failed to connect to OpenMV's serial port.\n"
|
||||
"Please install OpenMV's udev rules first:\n"
|
||||
"sudo cp openmv/udev/50-openmv.rules /etc/udev/rules.d/\n"
|
||||
"sudo udevadm control --reload-rules\n\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Set higher timeout after connecting for lengthy transfers.
|
||||
pyopenmv.set_timeout(1*2) # SD Cards can cause big hicups.
|
||||
pyopenmv.stop_script()
|
||||
pyopenmv.enable_fb(True)
|
||||
pyopenmv.reset_profiler()
|
||||
|
||||
# Configure some event counters.
|
||||
pyopenmv.set_event_counter(0, 0x0039)
|
||||
pyopenmv.set_event_counter(1, 0x0023)
|
||||
pyopenmv.set_event_counter(2, 0x0024)
|
||||
pyopenmv.set_event_counter(3, 0x0001)
|
||||
pyopenmv.set_event_counter(4, 0x0003)
|
||||
pyopenmv.set_event_counter(5, 0xC102)
|
||||
pyopenmv.set_event_counter(6, 0x02CC)
|
||||
pyopenmv.set_event_counter(7, 0xC303)
|
||||
|
||||
pyopenmv.exec_script(script)
|
||||
|
||||
# init screen
|
||||
running = True
|
||||
screen = None
|
||||
|
||||
# Profiling control
|
||||
profile_type = 0
|
||||
profile_mode = 0
|
||||
profile_data = []
|
||||
last_profile_read = 0
|
||||
|
||||
clock = pygame.time.Clock()
|
||||
fps_clock = pygame.time.Clock()
|
||||
font = pygame.font.SysFont("monospace", 30)
|
||||
|
||||
if not benchmark:
|
||||
pygame.display.set_caption("OpenMV Camera")
|
||||
else:
|
||||
pygame.display.set_caption("OpenMV Camera (Benchmark)")
|
||||
screen = pygame.display.set_mode((640, 120), pygame.DOUBLEBUF, 32)
|
||||
|
||||
try:
|
||||
while running:
|
||||
# Read state
|
||||
w, h, data, size, text, fmt, profiling = pyopenmv.read_state()
|
||||
|
||||
if text is not None:
|
||||
print(text, end="")
|
||||
|
||||
# Read profiling data (maximum 10Hz)
|
||||
if profiling and profile_type:
|
||||
current_time = time.time()
|
||||
if current_time - last_profile_read >= 0.1: # 10Hz = 0.1s interval
|
||||
tmp_data = pyopenmv.read_profile()
|
||||
if tmp_data:
|
||||
profile_data = tmp_data
|
||||
last_profile_read = current_time
|
||||
|
||||
#if profile_data:
|
||||
# for r in profile_data:
|
||||
# print(f"Func: {addr_to_symbol(symbols, r['address'])}@0x{r['address']:x} ")
|
||||
# print(f"Call: {addr_to_symbol(symbols, r['caller'])}@0x{r['caller']:x}")
|
||||
# sys.exit(0)
|
||||
|
||||
if data is not None:
|
||||
fps = fps_clock.get_fps()
|
||||
|
||||
# Create image from RGB888
|
||||
if not benchmark:
|
||||
image = pygame.image.frombuffer(data.flat[0:], (w, h), 'RGB')
|
||||
image = pygame.transform.smoothscale(image, (w * scale, h * scale))
|
||||
|
||||
if screen is None:
|
||||
screen = pygame.display.set_mode((w * scale, h * scale), pygame.DOUBLEBUF, 32)
|
||||
|
||||
# blit stuff
|
||||
if benchmark:
|
||||
screen.fill((0, 0, 0))
|
||||
else:
|
||||
screen.blit(image, (0, 0))
|
||||
|
||||
# FPS text
|
||||
fps_text = f"{fps:.2f} FPS {fps * size / 1024**2:.2f} MB/s {w}x{h} {fmt}"
|
||||
screen.blit(font.render(fps_text, 5, (255, 0, 0)), (0, 0))
|
||||
|
||||
# Draw profile overlay if enabled
|
||||
if profile_type and profile_data:
|
||||
draw_profile_overlay(screen, w, h, profile_data, profile_mode, profile_type, scale, symbols)
|
||||
|
||||
# update display
|
||||
pygame.display.flip()
|
||||
fps_clock.tick(1000//poll_rate)
|
||||
|
||||
for event in pygame.event.get():
|
||||
if event.type == pygame.QUIT:
|
||||
running = False
|
||||
elif event.type == pygame.KEYDOWN:
|
||||
if event.key == pygame.K_ESCAPE:
|
||||
running = False
|
||||
elif event.key == pygame.K_c:
|
||||
pygame.image.save(image, "capture.png")
|
||||
elif event.key == pygame.K_p:
|
||||
profile_type = (profile_type + 1) % 3
|
||||
elif event.key == pygame.K_m:
|
||||
profile_mode = not profile_mode
|
||||
pyopenmv.set_profile_mode(profile_mode)
|
||||
elif event.key == pygame.K_r:
|
||||
pyopenmv.reset_profiler()
|
||||
|
||||
clock.tick(1000//poll_rate)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
pygame.quit()
|
||||
pyopenmv.stop_script()
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = argparse.ArgumentParser(description='pyopenmv module')
|
||||
parser.add_argument('--port', action = 'store', help='Serial port (dev/ttyACM0)', default='/dev/ttyACM0')
|
||||
parser.add_argument("--script", action = "store", default=None, help = "Script file")
|
||||
parser.add_argument('--poll', action = 'store', help='Poll rate (default 4ms)', default=4, type=int)
|
||||
parser.add_argument('--bench', action = 'store_true', help='Run throughput benchmark.', default=False)
|
||||
parser.add_argument('--scale', action = 'store', help='Set frame scaling factor (default 4x).', default=4, type=int)
|
||||
parser.add_argument('--firmware', action = 'store', help='Firmware for address to symbol', default=None)
|
||||
|
||||
args = parser.parse_args()
|
||||
if args.script is not None:
|
||||
with open(args.script) as f:
|
||||
args.script = f.read()
|
||||
else:
|
||||
args.script = bench_script if args.bench else test_script
|
||||
|
||||
symbols = []
|
||||
|
||||
if args.firmware:
|
||||
from elftools.elf.elffile import ELFFile
|
||||
|
||||
with open(args.firmware, 'rb') as f:
|
||||
elf = ELFFile(f)
|
||||
symtab = elf.get_section_by_name('.symtab')
|
||||
if not symtab:
|
||||
raise ValueError("No symbol table found in ELF.")
|
||||
|
||||
for sym in symtab.iter_symbols():
|
||||
addr = sym['st_value']
|
||||
size = sym['st_size']
|
||||
name = sym.name
|
||||
if name and size > 0: # ignore empty symbols
|
||||
symbols.append((addr, addr + size, name))
|
||||
|
||||
symbols.sort()
|
||||
|
||||
pygame_test(args.port, args.script, args.poll, args.scale, args.bench, symbols)
|
@ -1,281 +0,0 @@
|
||||
#!/usr/bin/env python2
|
||||
# This file is part of the OpenMV project.
|
||||
#
|
||||
# Copyright (c) 2013-2021 Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Copyright (c) 2013-2021 Kwabena W. Agyeman <kwagyeman@openmv.io>
|
||||
#
|
||||
# This work is licensed under the MIT license, see the file LICENSE for details.
|
||||
#
|
||||
# Openmv module with support for multiple cams.
|
||||
|
||||
import struct
|
||||
import sys,time
|
||||
import serial
|
||||
import platform
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
__serial = []
|
||||
__port = []
|
||||
|
||||
__FB_HDR_SIZE =12
|
||||
|
||||
# USB Debug commands
|
||||
__USBDBG_CMD = 48
|
||||
__USBDBG_FW_VERSION = 0x80
|
||||
__USBDBG_FRAME_SIZE = 0x81
|
||||
__USBDBG_FRAME_DUMP = 0x82
|
||||
__USBDBG_ARCH_STR = 0x83
|
||||
__USBDBG_SCRIPT_EXEC = 0x05
|
||||
__USBDBG_SCRIPT_STOP = 0x06
|
||||
__USBDBG_SCRIPT_SAVE = 0x07
|
||||
__USBDBG_SCRIPT_RUNNING = 0x87
|
||||
__USBDBG_TEMPLATE_SAVE = 0x08
|
||||
__USBDBG_DESCRIPTOR_SAVE= 0x09
|
||||
__USBDBG_ATTR_READ = 0x8A
|
||||
__USBDBG_ATTR_WRITE = 0x0B
|
||||
__USBDBG_SYS_RESET = 0x0C
|
||||
__USBDBG_FB_ENABLE = 0x0D
|
||||
__USBDBG_TX_BUF_LEN = 0x8E
|
||||
__USBDBG_TX_BUF = 0x8F
|
||||
|
||||
ATTR_CONTRAST =0
|
||||
ATTR_BRIGHTNESS =1
|
||||
ATTR_SATURATION =2
|
||||
ATTR_GAINCEILING=3
|
||||
|
||||
__BOOTLDR_START = 0xABCD0001
|
||||
__BOOTLDR_RESET = 0xABCD0002
|
||||
__BOOTLDR_ERASE = 0xABCD0004
|
||||
__BOOTLDR_WRITE = 0xABCD0008
|
||||
|
||||
def init(port, baudrate=921600, timeout=0.3):
|
||||
global __serial
|
||||
global __port
|
||||
# open CDC port
|
||||
__serial.append(serial.Serial(port, baudrate=baudrate, timeout=timeout))
|
||||
__port.append(port)
|
||||
|
||||
def disconnect(port):
|
||||
global __serial
|
||||
global __port
|
||||
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].close()
|
||||
__serial.pop(idx)
|
||||
__port.pop(idx)
|
||||
except:
|
||||
pass
|
||||
|
||||
def set_timeout(port, timeout):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].timeout = timeout
|
||||
except:
|
||||
pass
|
||||
|
||||
def fb_size(port):
|
||||
|
||||
# read fb header
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_FRAME_SIZE, __FB_HDR_SIZE))
|
||||
return struct.unpack("III", __serial[idx].read(12))
|
||||
except:
|
||||
return None
|
||||
|
||||
def fb_dump(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
size = fb_size(port)
|
||||
|
||||
if (not size[0]):
|
||||
# frame not ready
|
||||
return None
|
||||
|
||||
if (size[2] > 2): #JPEG
|
||||
num_bytes = size[2]
|
||||
else:
|
||||
num_bytes = size[0]*size[1]*size[2]
|
||||
|
||||
# read fb data
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_FRAME_DUMP, num_bytes))
|
||||
buff = __serial[idx].read(num_bytes)
|
||||
|
||||
if size[2] == 1: # Grayscale
|
||||
y = np.fromstring(buff, dtype=np.uint8)
|
||||
buff = np.column_stack((y, y, y))
|
||||
elif size[2] == 2: # RGB565
|
||||
arr = np.fromstring(buff, dtype=np.uint16).newbyteorder('S')
|
||||
r = (((arr & 0xF800) >>11)*255.0/31.0).astype(np.uint8)
|
||||
g = (((arr & 0x07E0) >>5) *255.0/63.0).astype(np.uint8)
|
||||
b = (((arr & 0x001F) >>0) *255.0/31.0).astype(np.uint8)
|
||||
buff = np.column_stack((r,g,b))
|
||||
else: # JPEG
|
||||
try:
|
||||
buff = np.asarray(Image.frombuffer("RGB", size[0:2], buff, "jpeg", "RGB", ""))
|
||||
except Exception as e:
|
||||
print ("JPEG decode error (%s)"%(e))
|
||||
return None
|
||||
|
||||
if (buff.size != (size[0]*size[1]*3)):
|
||||
return None
|
||||
|
||||
return (size[0], size[1], buff.reshape((size[1], size[0], 3)))
|
||||
except:
|
||||
return None
|
||||
|
||||
def exec_script(port, buf):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_EXEC, len(buf)))
|
||||
__serial[idx].write(buf.encode())
|
||||
except:
|
||||
pass
|
||||
|
||||
def stop_script(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_STOP, 0))
|
||||
except:
|
||||
pass
|
||||
|
||||
def script_running(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_SCRIPT_RUNNING, 4))
|
||||
return struct.unpack("I", __serial[idx].read(4))[0]
|
||||
except:
|
||||
return None
|
||||
|
||||
def save_template(port, x, y, w, h, path):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
buf = struct.pack("IIII", x, y, w, h) + path
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_TEMPLATE_SAVE, len(buf)))
|
||||
__serial[idx].write(buf)
|
||||
except:
|
||||
pass
|
||||
|
||||
def save_descriptor(port, x, y, w, h, path):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
buf = struct.pack("HHHH", x, y, w, h) + path
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_DESCRIPTOR_SAVE, len(buf)))
|
||||
__serial[idx].write(buf)
|
||||
except:
|
||||
pass
|
||||
|
||||
def set_attr(port, attr, value):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_ATTR_WRITE, 8))
|
||||
__serial[idx].write(struct.pack("<II", attr, value))
|
||||
except:
|
||||
pass
|
||||
|
||||
def get_attr(port, attr):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBIh", __USBDBG_CMD, __USBDBG_ATTR_READ, 1, attr))
|
||||
return __serial[idx].read(1)
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def reset(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_SYS_RESET, 0))
|
||||
except:
|
||||
pass
|
||||
|
||||
def bootloader_start(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<I", __BOOTLDR_START))
|
||||
return struct.unpack("I", __serial[idx].read(4))[0] == __BOOTLDR_START
|
||||
except:
|
||||
pass
|
||||
|
||||
def bootloader_reset(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<I", __BOOTLDR_RESET))
|
||||
except:
|
||||
pass
|
||||
|
||||
def flash_erase(port, sector):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<II", __BOOTLDR_ERASE, sector))
|
||||
except:
|
||||
pass
|
||||
|
||||
def flash_write(port, buf):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<I", __BOOTLDR_WRITE) + buf)
|
||||
except:
|
||||
pass
|
||||
|
||||
def tx_buf_len(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_TX_BUF_LEN, 4))
|
||||
return struct.unpack("I", __serial[idx].read(4))[0]
|
||||
except:
|
||||
return None
|
||||
|
||||
def tx_buf(port, bytes):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_TX_BUF, bytes))
|
||||
return __serial[idx].read(bytes)
|
||||
except:
|
||||
return None
|
||||
|
||||
def fw_version(port):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_FW_VERSION, 12))
|
||||
return struct.unpack("III", __serial[idx].read(12))
|
||||
except:
|
||||
return None
|
||||
|
||||
def enable_fb(port, enable):
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_FB_ENABLE, 4))
|
||||
__serial[idx].write(struct.pack("<I", enable))
|
||||
except:
|
||||
pass
|
||||
|
||||
def arch_str():
|
||||
try:
|
||||
idx = __port.index(port)
|
||||
__serial[idx].write(struct.pack("<BBI", __USBDBG_CMD, __USBDBG_ARCH_STR, 64))
|
||||
return __serial[idx].read(64).split('\0', 1)[0]
|
||||
except:
|
||||
return None
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv)!= 3:
|
||||
print ('usage: pyopenmv.py <port> <script>')
|
||||
sys.exit(1)
|
||||
|
||||
with open(sys.argv[2], 'r') as fin:
|
||||
buf = fin.read()
|
||||
|
||||
portname = sys.argv[1]
|
||||
|
||||
disconnect(portname)
|
||||
init(portname)
|
||||
stop_script(portname)
|
||||
exec_script(portname, buf)
|
||||
tx_len = tx_buf_len(portname)
|
||||
time.sleep(0.250)
|
||||
if (tx_len):
|
||||
print(tx_buf(portname, tx_len).decode())
|
||||
disconnect(portname)
|
@ -1,77 +0,0 @@
|
||||
#!/usr/bin/env python2
|
||||
# This file is part of the OpenMV project.
|
||||
#
|
||||
# Copyright (c) 2013-2021 Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Copyright (c) 2013-2021 Kwabena W. Agyeman <kwagyeman@openmv.io>
|
||||
#
|
||||
# This work is licensed under the MIT license, see the file LICENSE for details.
|
||||
#
|
||||
# This script stress-tests script execution.
|
||||
|
||||
import sys, os
|
||||
import pyopenmv
|
||||
import argparse
|
||||
from time import sleep
|
||||
from random import randint
|
||||
|
||||
def main():
|
||||
# CMD args parser
|
||||
parser = argparse.ArgumentParser(description='openmv stress test')
|
||||
parser.add_argument("-j", "--disable_fb", action = "store_true", help = "Disable FB JPEG compression")
|
||||
parser.add_argument("-p", "--port", action = "store", help = "OpenMV serial port")
|
||||
parser.add_argument("-t", "--time", action = "store", default = 100, help = "Max time before stopping the script")
|
||||
parser.add_argument("-s", "--script", action = "store",\
|
||||
default="../scripts/examples/00-HelloWorld/helloworld.py", help = "OpenMV script file")
|
||||
|
||||
# Parse CMD args
|
||||
args = parser.parse_args()
|
||||
|
||||
# init openmv
|
||||
if (args.port):
|
||||
portname = args.port
|
||||
elif 'darwin' in sys.platform:
|
||||
portname = "/dev/cu.usbmodem14221"
|
||||
else:
|
||||
portname = "/dev/openmvcam"
|
||||
|
||||
print("\n>>>Reading script: %s\n" %(args.script))
|
||||
with open(args.script, "r") as f:
|
||||
script = f.read()
|
||||
print("%s\n" %(script))
|
||||
|
||||
connected = False
|
||||
for i in range(10):
|
||||
try:
|
||||
# Open serial port.
|
||||
# Set small timeout when connecting
|
||||
pyopenmv.init(portname, baudrate=921600, timeout=0.050)
|
||||
connected = True
|
||||
break
|
||||
except Exception as e:
|
||||
connected = False
|
||||
sleep(0.100)
|
||||
|
||||
if not connected:
|
||||
print ( "Failed to connect to OpenMV's serial port.\n"
|
||||
"Please install OpenMV's udev rules first:\n"
|
||||
"sudo cp openmv/udev/50-openmv.rules /etc/udev/rules.d/\n"
|
||||
"sudo udevadm control --reload-rules\n\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Set higher timeout after connecting.
|
||||
pyopenmv.set_timeout(0.500)
|
||||
|
||||
# Enable/Disable framebuffer compression.
|
||||
print(">>>Enable FB JPEG compression %s" %(str(not args.disable_fb)))
|
||||
pyopenmv.enable_fb(not args.disable_fb)
|
||||
|
||||
# Interrupt running script.
|
||||
pyopenmv.stop_script()
|
||||
max_timeout = int(args.time)
|
||||
for i in range(1000):
|
||||
pyopenmv.exec_script(script)
|
||||
sleep(randint(0, max_timeout)/1000)
|
||||
pyopenmv.stop_script()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
Loading…
Reference in New Issue
Block a user