26 Introduction to Main Program Architecture
Introduction to the Main Program Architecture
File Structure and Functions
- ugv_pt_rpi
- [folder] AccessPopup (for network connection related functions)
- [folder] sounds (for storing audio files; you can configure voice packs here)
- [folder] static (for storing captured photos)
- [folder] templates (resources for the web application)
- [folder] tutorial_cn (interactive tutorials in Chinese)
- [folder] tutorial_en (interactive tutorials in English)
- [folder] videos (for storing recorded videos)
- app.py (main program of the product, including web‑socket and flask functionality)
- asound.conf (audio card configuration file)
- audio_ctrl.py (library for audio functions)
- autorun.sh (script to configure auto‑start of the main program and jupyterlab at boot)
- base_camera.py (low‑level multi‑threaded image acquisition library for flask real‑time video, from the original flask‑video‑streaming project)
- base_ctrl.py (library for communicating with the lower computer via serial port)
- config.yaml (configuration file for various parameters)
- requirements.txt (Python project dependencies)
- robot_ctrl.py (library for robot motion and vision processing)
- serial_simple_ctrl.py (standalone program for testing serial communication)
- setup.sh (automatic installation script)
- start_jupyter.sh (starts the JupyterLab server)
Installation Script
There is a file named setup.sh in the project folder, written in shell script. It helps automatically configure the robot product's upper computer, including setting up the serial port, configuring the camera, creating a project virtual environment, and installing dependencies. These steps are already configured in the image of the TF card we ship.
How to use the installation script: the installation process will download and install many dependency libraries from the network. For regions with special network conditions, we recommend downloading the image file directly from our website to install the product.
Auto‑Run Programs
The autorun.sh file in the project folder is used to configure the main program (app.py) and JupyterLab (start_jupyter.sh) to start automatically at boot (as the user, not root), and also generates the JupyterLab configuration file.
Introduction to app.py (v0.89)
The following code block is for demonstration only and cannot be executed.
Import Flask application and JSON‑related libraries for building the web application:
from importlib import import_module import os, socket, psutil import subprocess, re, netifaces from flask import Flask, render_template, Response, jsonify, request, send_from_directory, send_file from werkzeug.utils import secure_filename import json
Import web‑socket related libraries:
from flask_socketio import SocketIO, emit
Import robot control related libraries, including vision functions and motion control:
from robot_ctrl import Camera from robot_ctrl import RobotCtrlMiddleWare
Import other libraries:
import time # time functions import logging # for setting Flask application output information import threading # multi‑threading library import yaml # for reading .yaml configuration files
Open the config.yaml configuration file and obtain the parameters:
curpath = os.path.realpath(__file__)
thisPath = os.path.dirname(curpath)
with open(thisPath + '/config.yaml', 'r') as yaml_file:
f = yaml.safe_load(yaml_file)
robot_name = f['base_config']['robot_name']
sbc_version = f['base_config']['sbc_version']
Instantiate the Flask application and configure output (disable output):
app = Flask(__name__)
log = logging.getLogger('werkzeug')
log.disabled = True
Instantiate web‑socket functionality (for communication between web client and server), video related functions (real‑time video, OpenCV), robot motion control (movement, lights, gimbal, obtaining chassis feedback, etc.):
socketio = SocketIO(app) camera = Camera() robot = RobotCtrlMiddleWare()
Network related settings:
net_interface = "wlan0" # set the wireless network card; onboard is wlan0, USB ones use other numbers wifi_mode = "None" # Store the Wi‑Fi mode; this variable will be displayed on the OLED screen eth0_ip = None # Ethernet IP address, displayed on the OLED screen wlan_ip = None # wireless network (net_interface) IP address, displayed on the OLED screen
Storage path for audio files uploaded via drag‑and‑drop on the web page:
UPLOAD_FOLDER = thisPath + '/sounds/others'
Some variables for storing upper computer information (these variables are updated by other functions when the main program runs):
pic_size = 0; vid_size = 0; cpu_read = 0; cpu_temp = 0; ram_read = 0; rssi_read= 0;
A dictionary of executable commands and corresponding functions. Each command has a corresponding code stored in config.yaml. The dictionary is used to select the command to execute. Because there are many commands, using a long chain of if‑else would harm readability.
cmd_actions = {
f['code']['min_res']: lambda: camera.set_video_resolution("240P"),
f['code']['mid_res']: lambda: camera.set_video_resolution("480P"),
f['code']['max_res']: lambda: camera.set_video_resolution("960P"),
f['code']['zoom_x1']: lambda: camera.scale_frame(1),
f['code']['zoom_x2']: lambda: camera.scale_frame(2),
f['code']['zoom_x4']: lambda: camera.scale_frame(4),
f['code']['pic_cap']: lambda: camera.capture_frame(thisPath + '/static/'),
f['code']['vid_sta']: lambda: camera.record_video(1, thisPath + '/videos/'),
f['code']['vid_end']: lambda: camera.record_video(0, thisPath + '/videos/'),
f['code']['cv_none']: lambda: camera.set_cv_mode(f['code']['cv_none']),
f['code']['cv_moti']: lambda: camera.set_cv_mode(f['code']['cv_moti']),
f['code']['cv_face']: lambda: camera.set_cv_mode(f['code']['cv_face']),
f['code']['cv_objs']: lambda: camera.set_cv_mode(f['code']['cv_objs']),
f['code']['cv_clor']: lambda: camera.set_cv_mode(f['code']['cv_clor']),
f['code']['cv_hand']: lambda: camera.set_cv_mode(f['code']['cv_hand']),
f['code']['cv_auto']: lambda: camera.set_cv_mode(f['code']['cv_auto']),
f['code']['mp_face']: lambda: camera.set_cv_mode(f['code']['mp_face']),
f['code']['mp_pose']: lambda: camera.set_cv_mode(f['code']['mp_pose']),
f['code']['re_none']: lambda: camera.set_detection_reaction(f['code']['re_none']),
f['code']['re_capt']: lambda: camera.set_detection_reaction(f['code']['re_capt']),
f['code']['re_reco']: lambda: camera.set_detection_reaction(f['code']['re_reco']),
f['code']['mc_lock']: lambda: camera.set_movtion_lock(f['code']['mc_lock']),
f['code']['mc_unlo']: lambda: camera.set_movtion_lock(f['code']['mc_unlo']),
f['code']['led_off']: robot.set_led_mode_off,
f['code']['led_aut']: robot.set_led_mode_auto,
f['code']['led_ton']: robot.set_led_mode_on,
f['code']['base_of']: robot.set_base_led_off,
f['code']['base_on']: robot.set_base_led_on,
f['code']['head_ct']: robot.head_led_ctrl,
f['code']['base_ct']: robot.base_led_ctrl,
f['code']['s_panid']: camera.set_pan_id,
f['code']['release']: camera.release_torque,
f['code']['set_mid']: camera.middle_set,
f['code']['s_tilid']: camera.set_tilt_id
}
When the web page loads, it also requests the config.yaml file from the server for configuring some information on the web page, such as displaying the product name or unifying command codes with the server. The web page sends a request through the "/config" route, and the server returns the required config.yaml file to the web client.
@app.route('/config')
def get_config():
with open(thisPath + '/config.yaml', 'r') as file:
yaml_content = file.read()
return yaml_content
Obtain the Wi‑Fi signal strength. The parameter is the name of the wireless network card (a device may have multiple wireless cards).
def get_signal_strength(interface):
try:
output = subprocess.check_output(["/sbin/iwconfig", interface]).decode("utf-8")
signal_strength = re.search(r"Signal level=(-\d+)", output)
if signal_strength:
return int(signal_strength.group(1))
return 0
except FileNotFoundError:
print("iwconfig command not found. Please ensure it's installed and in your PATH.")
return -1
except subprocess.CalledProcessError as e:
print(f"Error executing iwconfig: {e}")
return -1
except Exception as e:
print(f"An error occurred: {e}")
return -1
Obtain the Wi‑Fi mode, determining whether Wi‑Fi is in AP or STA mode.
def get_wifi_mode():
global wifi_mode
try:
result = subprocess.check_output(['/sbin/iwconfig', 'wlan0'], encoding='utf-8')
if "Mode:Master" in result or "Mode:AP" in result:
wifi_mode = "AP"
return "AP"
if "Mode:Managed" in result:
wifi_mode = "STA"
return "STA"
except subprocess.CalledProcessError as e:
print(f"Error checking Wi-Fi mode: {e}")
return None
return None
Obtain the IP address. The required parameter is the network card name:
def get_ip_address(interface):
try:
interface_info = netifaces.ifaddresses(interface)
ipv4_info = interface_info.get(netifaces.AF_INET, [{}])
return ipv4_info[0].get('addr')
except ValueError:
print(f"Interface {interface} not found.")
return None
except IndexError:
print(f"No IPv4 address assigned to {interface}.")
return None
Obtain CPU usage. This function blocks; the blocking time is the interval parameter of cpu_percent():
def get_cpu_usage():
return psutil.cpu_percent(interval=2)
Obtain the CPU temperature:
def get_cpu_temperature():
try:
temperature_str = os.popen('vcgencmd measure_temp').readline()
temperature = float(temperature_str.replace("temp=", "").replace("'C\n", ""))
return temperature
except Exception as e:
print("Error reading CPU temperature:", str(e))
return None
Obtain memory usage:
def get_memory_usage():
return psutil.virtual_memory().percent
This function integrates the functions above to obtain various information and assign them to the corresponding variables, making it easy for other parts of the program to access those variables:
def update_device_info():
global pic_size, vid_size, cpu_read, ram_read, rssi_read, cpu_temp
cpu_read = get_cpu_usage()
cpu_temp = get_cpu_temperature()
ram_read = get_memory_usage()
rssi_read= get_signal_strength(net_interface)
This function continuously obtains various feedback information, merges it, and sends it to the web client. It is executed in a separate thread created by threading when the main program starts (when the first client establishes a connection), without affecting the main program's execution. This function obtains chassis information and other vision‑related information from camera.get_status() and sends a merged JSON back to the web client. The feedback route is "/ctrl".
The execution frequency of this function is 10 Hz, but some information is obtained at a lower frequency (the 10 Hz refers to the frequency of camera.get_status()). Other information such as folder size, CPU and memory usage are obtained much less frequently because they consume more resources.
def update_data_websocket():
while 1:
try:
fb_json = camera.get_status()
except:
continue
socket_data = {
f['fb']['picture_size']:pic_size,
f['fb']['video_size']: vid_size,
f['fb']['cpu_load']: cpu_read,
f['fb']['cpu_temp']: cpu_temp,
f['fb']['ram_usage']: ram_read,
f['fb']['wifi_rssi']: rssi_read
}
try:
socket_data.update(fb_json)
socketio.emit('update', socket_data, namespace='/ctrl')
except:
pass
time.sleep(0.1)
All subsequent `@app.route()` decorators define route functions for the Flask application. Routes distinguish the types of requests sent by the client, and different types of requests are handled by different functions.
This is the main route. When a client connects (when someone visits the IP:5000 page), a random audio file from the `sounds/connected` folder is played, and the server returns the main control interface of the web application. The HTML file for the main control interface is `index.html`.
@app.route('/')
def index():
"""Video streaming home page."""
robot.play_random_audio("connected", False)
return render_template('index.html')
Routes for sending various files to the client: CSS, JS, photos, videos.
@app.route('/<path:filename>')
def serve_static(filename):
return send_from_directory('templates', filename)
@app.route('/photo/<path:filename>')
def serve_static_photo(filename):
return send_from_directory('templates', filename)
@app.route('/video/<path:filename>')
def serve_static_video(filename):
return send_from_directory('templates', filename)
Route for opening the settings page:
@app.route('/settings/<path:filename>')
def serve_static_settings(filename):
return send_from_directory('templates', filename)
Route for returning to the home page from the settings page:
@app.route('/index')
def serve_static_home(filename):
return redirect(url_for('index'))
Function for obtaining the real‑time video feed, from the open‑source project flask‑video‑streaming:
def gen(cameraInput):
"""Video streaming generator function."""
yield b'--frame\r\n'
while True:
frame = cameraInput.get_frame()
yield b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n--frame\r\n'
Route for displaying the real‑time video feed on the web page:
@app.route('/video_feed')
def video_feed():
"""Video streaming route. Put this in the src attribute of an img tag."""
return Response(gen(camera),
mimetype='multipart/x-mixed-replace; boundary=frame')
Route for obtaining the list of photo names in the photo folder:
@app.route('/get_photo_names')
def get_photo_names():
photo_files = sorted(os.listdir(thisPath + '/static'), key=lambda x: os.path.getmtime(os.path.join(thisPath + '/static', x)), reverse=True)
return jsonify(photo_files)
Route for sending an image to the web page:
@app.route('/get_photo/<filename>')
def get_photo(filename):
return send_from_directory(thisPath + '/static', filename)
Route for deleting a photo:
@app.route('/delete_photo', methods=['POST'])
def delete_photo():
filename = request.form.get('filename')
try:
os.remove(os.path.join(thisPath + '/static', filename))
return jsonify(success=True)
except Exception as e:
print(e)
return jsonify(success=False)
The following functions are similar but for operating on videos:
@app.route('/delete_video', methods=['POST'])
def delete_video():
filename = request.form.get('filename')
try:
os.remove(os.path.join(thisPath + '/videos', filename))
return jsonify(success=True)
except Exception as e:
print(e)
return jsonify(success=False)
@app.route('/get_video_names')
def get_video_names():
video_files = sorted(
[filename for filename in os.listdir(thisPath + '/videos/') if filename.endswith('.mp4')],
key=lambda filename: os.path.getctime(os.path.join(thisPath + '/videos/', filename)),
reverse=True
)
return jsonify(video_files)
@app.route('/videos/<path:filename>')
def videos(filename):
return send_from_directory(thisPath + '/videos', filename)
A function to get the size of a folder by iterating over all files inside. This consumes significant resources.
def get_folder_size(folder_path):
total_size = 0
for dirpath, dirnames, filenames in os.walk(folder_path):
for filename in filenames:
file_path = os.path.join(dirpath, filename)
total_size += os.path.getsize(file_path)
# Convert total_size to MB
size_in_mb = total_size / (1024 * 1024)
return round(size_in_mb,2)
A web‑socket route for receiving JSON commands from the client. Some commands are high‑frequency and require as low latency as possible, so web‑socket is used instead of HTTP. Web‑socket is connection‑oriented: one connection, multiple communications. HTTP is connectionless: each request requires establishing a connection, communicating, and destroying the connection. HTTP is not suitable for high‑frequency, low‑latency communication (its advantage is simplicity).
@socketio.on('json', namespace='/json')
def handle_socket_json(json):
try:
robot.json_command_handler(json)
except Exception as e:
print("Error handling JSON data:", e)
return
A function that updates the information displayed on the OLED. This function runs in a separate thread created by threading when the main program runs. The function mentioned earlier that sends feedback to the web page at 10 Hz uses some variables updated by this function. The information here does not require high real‑time performance, so the loop inside this function obtains information at a lower frequency. (This function does not use time.sleep() for delays; it relies on update_device_info() which internally calls get_cpu_usage() to create delays.)
def oled_update():
global eth0_ip, wlan_ip
robot.base_oled(0, f"E: No Ethernet")
robot.base_oled(1, f"W: NO {net_interface}")
robot.base_oled(2, "F/J:5000/8888")
get_wifi_mode()
start_time = time.time()
last_folder_check_time = 0
while True:
current_time = time.time()
if current_time - last_folder_check_time > 600:
pic_size = get_folder_size(thisPath + '/static')
vid_size = get_folder_size(thisPath + '/videos')
last_folder_check_time = current_time
update_device_info() # the interval of this loop is set in here
get_wifi_mode()
if get_ip_address('eth0') != eth0_ip:
eth0_ip = get_ip_address('eth0');
if eth0_ip:
robot.base_oled(0, f"E:{eth0_ip}")
else:
robot.base_oled(0, f"E: No Ethernet")
if get_ip_address(net_interface) != wlan_ip:
wlan_ip = get_ip_address(net_interface)
if wlan_ip:
robot.base_oled(1, f"W:{wlan_ip}")
else:
robot.base_oled(1, f"W: NO {net_interface}")
elapsed_time = current_time - start_time
hours = int(elapsed_time // 3600)
minutes = int((elapsed_time % 3600) // 60)
seconds = int(elapsed_time % 60)
robot.base_oled(3, f"{wifi_mode} {hours:02d}:{minutes:02d}:{seconds:02d} {rssi_read}dBm")
This is a route for handling command line information sent from the client:
@app.route('/send_command', methods=['POST'])
def handle_command():
command = request.form['command']
print("Received command:", command)
# camera.info_update("CMD:" + command, (0,255,255), 0.36)
camera.cmd_process(command)
return jsonify({"status": "success", "message": "Command received"})
Route for obtaining the list of audio files stored in the sounds/other folder:
@app.route('/getAudioFiles', methods=['GET'])
def get_audio_files():
files = [f for f in os.listdir(UPLOAD_FOLDER) if os.path.isfile(os.path.join(UPLOAD_FOLDER, f))]
return jsonify(files)
Route for implementing drag‑and‑drop upload functionality on the web page. Uploaded audio files are stored in the sounds/other folder:
@app.route('/uploadAudio', methods=['POST'])
def upload_audio():
if 'file' not in request.files:
return jsonify({'error': 'No file part'})
file = request.files['file']
if file.filename == '':
return jsonify({'error': 'No selected file'})
if file:
filename = secure_filename(file.filename)
file.save(os.path.join(UPLOAD_FOLDER, filename))
return jsonify({'success': 'File uploaded successfully'})
Route for playing an audio file from the sounds/other folder:
@app.route('/playAudio', methods=['POST'])
def play_audio():
audio_file = request.form['audio_file']
print(thisPath + '/sounds/others/' + audio_file)
robot.audio_play(thisPath + '/sounds/others/' + audio_file)
return jsonify({'success': 'Audio is playing'})
Route for stopping playback:
@app.route('/stop_audio', methods=['POST'])
def audio_stop():
robot.audio_stop()
return jsonify({'success': 'Audio stop'})
This function executes some command line instructions automatically at boot. It is executed automatically when the main program starts. You can freely add more instructions.
- base -c {"T":142,"cmd":50} : Sends an instruction to the chassis to set the feedback interval. By default the chassis loop has no delay; giving a parameter of 50 adds an extra 50 ms delay, which helps improve the efficiency of the upper computer (decoding serial information from the lower computer also consumes resources).
- base -c {"T":131,"cmd":1} : Enables chassis feedback streaming, so that feedback is no longer request‑response. The chassis can automatically send continuous feedback (although the ROS driver also uses this mode by default, it is written here anyway).
- base -c {"T":143,"cmd":0} : Disables echo. When sending commands to the chassis, the chassis will not echo the original command (unnecessary, and it would consume serial decoding resources on the upper computer, especially when using high‑frequency commands to control the chassis).
- base -c {"T":4,"cmd":2} : Sets the peripheral type of the chassis: 0 = no peripheral, 1 = robotic arm, 2 = gimbal.
- base -c {"T":300,"mode":0,"mac":"EF:EF:EF:EF:EF:EF"} : The chassis will not be controlled by ESP‑NOW broadcast signals; it can only be controlled by ESP‑NOW commands from the MAC address EF:EF:EF:EF:EF:EF. You can freely change this MAC address (since the upper computer program runs successfully, there is no need for ESP‑NOW broadcast control).
- send -a -b : Adds the broadcast address to the lower computer's ESP‑NOW peer list, facilitating future device‑to‑device communication.
def cmd_on_boot():
cmd_list = [
'base -c {"T":142,"cmd":50}', # set feedback interval
'base -c {"T":131,"cmd":1}', # serial feedback flow on
'base -c {"T":143,"cmd":0}', # serial echo off
'base -c {"T":4,"cmd":2}', # select the module - 0:None 1:RoArm-M2-S 2:Gimbal
'base -c {"T":300,"mode":0,"mac":"EF:EF:EF:EF:EF:EF"}', # the base won't be ctrl by esp-now broadcast cmd, but it can still recv broadcast megs.
'send -a -b' # add broadcast mac addr to peer
]
for i in range(0, len(cmd_list)):
camera.cmd_process(cmd_list[i])
What is executed when the main program runs:
if __name__ == '__main__':
# Play a random audio file from the sounds/robot_started folder
robot.play_random_audio("robot_started", False)
# Turn on the LED lights
robot.set_led_mode_on()
# Create a separate thread for update_data_websocket()
date_update_thread = threading.Thread(target=update_data_websocket, daemon=True)
date_update_thread.start()
# Create a separate thread for oled_update()
oled_update_thread = threading.Thread(target=oled_update, daemon=True)
oled_update_thread.start()
# Get the size of the photo folder
pic_size = get_folder_size(thisPath + '/static')
# Get the size of the video folder
vid_size = get_folder_size(thisPath + '/videos')
# Turn off the LED lights
robot.set_led_mode_off()
# Run commands automatically at boot
cmd_on_boot()
# Start the Flask application server
socketio.run(app, host='0.0.0.0', port=5000, allow_unsafe_werkzeug=True)