two
This commit is contained in:
Vendored
+3
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"cmake.sourceDirectory": "E:/code/code/code/new/new/new/new/new/nw/archery - 副本/cpp_ext"
|
||||
}
|
||||
@@ -109,6 +109,7 @@
|
||||
from maix import app, uart, pinmap, time
|
||||
import hashlib
|
||||
import hmac
|
||||
import re
|
||||
import ujson
|
||||
|
||||
# ========== 配置 ==========
|
||||
@@ -130,53 +131,109 @@ def generate_token(device_id):
|
||||
return "Arrow_" + hmac.new((SALT + device_id).encode(), SALT2.encode(), hashlib.sha256).hexdigest()
|
||||
|
||||
def send_cmd(cmd_str, timeout_ms=3000):
|
||||
"""发送 AT 指令并等待 OK / ERROR"""
|
||||
"""发送 AT 指令并返回完整响应;超时返回已收到的内容。"""
|
||||
print("[AT] =>", cmd_str)
|
||||
http_serial.write((cmd_str + "\r\n").encode())
|
||||
buffer = b""
|
||||
start = time.ticks_ms()
|
||||
while time.ticks_ms() - start < timeout_ms:
|
||||
while time.ticks_diff(time.ticks_ms(), start) < timeout_ms:
|
||||
data = http_serial.read(128)
|
||||
if data:
|
||||
buffer += data
|
||||
try:
|
||||
decoded = buffer.decode()
|
||||
print("<= ", decoded.strip())
|
||||
if "OK" in decoded:
|
||||
return True
|
||||
if "+CME ERROR" in decoded or "ERROR" in decoded:
|
||||
return False
|
||||
decoded = buffer.decode("utf-8", "ignore")
|
||||
if "OK" in decoded or "+CME ERROR" in decoded or "ERROR" in decoded:
|
||||
print("[AT] <=", decoded.strip())
|
||||
return decoded
|
||||
except:
|
||||
pass
|
||||
time.sleep_ms(10)
|
||||
decoded = buffer.decode("utf-8", "ignore")
|
||||
print("[AT] !! timeout", timeout_ms, "ms, response:", decoded.strip() or "<empty>")
|
||||
return decoded
|
||||
|
||||
|
||||
def response_ok(response):
|
||||
return "OK" in response and "ERROR" not in response
|
||||
|
||||
|
||||
def wait_modem_ready():
|
||||
"""等待模组响应,并确认 PDP 上下文已经获得 IP。"""
|
||||
for attempt in range(15):
|
||||
if response_ok(send_cmd("AT", 1000)):
|
||||
break
|
||||
print("[4G] 等待模组启动", attempt + 1, "/15")
|
||||
time.sleep_ms(1000)
|
||||
else:
|
||||
print("[4G] UART2 无 AT 响应,请检查模组供电、A28/A29 接线和串口占用")
|
||||
return False
|
||||
|
||||
send_cmd("ATE0", 1000)
|
||||
cpin = send_cmd("AT+CPIN?", 3000)
|
||||
if "READY" not in cpin:
|
||||
print("[4G] SIM 卡未就绪:", cpin.strip())
|
||||
return False
|
||||
|
||||
addr = send_cmd("AT+CGPADDR=1", 3000)
|
||||
match = re.search(r'\+CGPADDR:\s*1,"([^\"]+)"', addr)
|
||||
if match and match.group(1) != "0.0.0.0":
|
||||
print("[4G] PDP ready, IP:", match.group(1))
|
||||
return True
|
||||
|
||||
send_cmd("AT+MIPCALL=1,1", 15000)
|
||||
for _ in range(20):
|
||||
addr = send_cmd("AT+CGPADDR=1", 3000)
|
||||
match = re.search(r'\+CGPADDR:\s*1,"([^\"]+)"', addr)
|
||||
if match and match.group(1) != "0.0.0.0":
|
||||
print("[4G] PDP ready, IP:", match.group(1))
|
||||
return True
|
||||
time.sleep_ms(1000)
|
||||
|
||||
print("[4G] PDP 未获得 IP,请检查 SIM 流量、信号和 APN")
|
||||
return False
|
||||
|
||||
|
||||
def clear_http_instances():
|
||||
for instance_id in range(6):
|
||||
send_cmd(f"AT+MHTTPDEL={instance_id}", 1200)
|
||||
|
||||
def create_http_instance(url):
|
||||
cmd = f'AT+MHTTPCREATE="{url}"'
|
||||
if send_cmd(cmd):
|
||||
# 尝试提取 instance ID(如果模块返回)
|
||||
# 注意:部分模块不会返回 ID,可忽略,直接用 0 或 1
|
||||
return True
|
||||
return False
|
||||
response = send_cmd(cmd, 8000)
|
||||
match = re.search(r"\+MHTTPCREATE:\s*(\d+)", response)
|
||||
if not response_ok(response) or not match:
|
||||
print("❌ 创建 HTTP 实例失败,模组响应:", response.strip() or "<empty>")
|
||||
return None
|
||||
return int(match.group(1))
|
||||
|
||||
def send_http_request(url, api_path, token, device_id, json_data):
|
||||
# 1. 创建 HTTP 实例
|
||||
if not create_http_instance(url):
|
||||
print("❌ 创建 HTTP 实例失败")
|
||||
instance_id = create_http_instance(url)
|
||||
if instance_id is None:
|
||||
return False
|
||||
|
||||
# 2. 设置 Headers(假设实例 ID 为 0,或根据模块默认)
|
||||
instance_id = 0 # 大多数模块默认实例为 0;若支持多实例,需解析返回值
|
||||
send_cmd(f'AT+MHTTPCFG="header",{instance_id},"Content-Type: application/json"')
|
||||
send_cmd(f'AT+MHTTPCFG="header",{instance_id},"Authorization: {token}"')
|
||||
send_cmd(f'AT+MHTTPCFG="header",{instance_id},"DeviceId: {device_id}"')
|
||||
# 2. 设置 Headers
|
||||
commands = (
|
||||
f'AT+MHTTPCFG="header",{instance_id},"Content-Type: application/json"',
|
||||
f'AT+MHTTPCFG="header",{instance_id},"Authorization: {token}"',
|
||||
f'AT+MHTTPCFG="header",{instance_id},"DeviceId: {device_id}"',
|
||||
)
|
||||
for command in commands:
|
||||
if not response_ok(send_cmd(command)):
|
||||
print("❌ HTTP Header 配置失败")
|
||||
send_cmd(f"AT+MHTTPDEL={instance_id}", 2000)
|
||||
return False
|
||||
|
||||
# 3. 发送 Body
|
||||
json_str = ujson.dumps(json_data)
|
||||
send_cmd(f'AT+MHTTPCONTENT={instance_id},0,0,"{json_str}"')
|
||||
at_json = json_str.replace("\\", "\\\\").replace('"', '\\"')
|
||||
if not response_ok(send_cmd(f'AT+MHTTPCONTENT={instance_id},0,0,"{at_json}"', 8000)):
|
||||
print("❌ HTTP Body 配置失败")
|
||||
send_cmd(f"AT+MHTTPDEL={instance_id}", 2000)
|
||||
return False
|
||||
|
||||
# 4. 发起 POST 请求
|
||||
if send_cmd(f'AT+MHTTPREQUEST={instance_id},2,0,"{api_path}"'):
|
||||
if response_ok(send_cmd(f'AT+MHTTPREQUEST={instance_id},2,0,"{api_path}"', 15000)):
|
||||
print("✅ HTTP 请求已发送")
|
||||
return True
|
||||
else:
|
||||
@@ -199,7 +256,7 @@ def read_response(timeout_ms=5000):
|
||||
print("🚀 启动直接上传流程...")
|
||||
|
||||
token = generate_token(device_id)
|
||||
print("🔑 Token:", token)
|
||||
print("🔑 Token 已生成:", token[:12] + "...")
|
||||
|
||||
# 构造模拟数据
|
||||
timestamp = int(time.time() * 1000)
|
||||
@@ -216,9 +273,16 @@ json_data = {
|
||||
}
|
||||
|
||||
# 执行上传
|
||||
if send_http_request(url, api_path, token, device_id, json_data):
|
||||
upload_ok = False
|
||||
if not wait_modem_ready():
|
||||
print("💥 4G 模组未就绪")
|
||||
else:
|
||||
clear_http_instances()
|
||||
upload_ok = send_http_request(url, api_path, token, device_id, json_data)
|
||||
|
||||
if upload_ok:
|
||||
read_response()
|
||||
else:
|
||||
print("💥 上传流程失败")
|
||||
|
||||
print("🔚 程序结束")
|
||||
print("🔚 程序结束")
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -18,6 +18,10 @@ files:
|
||||
- laser_manager.py
|
||||
- logger_manager.py
|
||||
- main.py
|
||||
- model_270139.cvimodel
|
||||
- model_270139.mud
|
||||
- model_270820.cvimodel
|
||||
- model_270820.mud
|
||||
- model_285484.cvimodel
|
||||
- model_285484.mud
|
||||
- network.py
|
||||
@@ -25,6 +29,7 @@ files:
|
||||
- ota_manager.py
|
||||
- power.py
|
||||
- server.pem
|
||||
- set_autostart.py
|
||||
- shoot_manager.py
|
||||
- shot_id_generator.py
|
||||
- target_roi_yolo.py
|
||||
|
||||
@@ -96,6 +96,11 @@ ADC_LASER_THRESHOLD = 3000
|
||||
|
||||
# ==================== 激光配置 ====================
|
||||
MODULE_ADDR = 0x00
|
||||
# 激光开关改由 A14 GPIO 控制:低电平开启,高电平关闭。
|
||||
LASER_CONTROL_PIN = "A14"
|
||||
LASER_CONTROL_GPIO = "GPIOA14"
|
||||
LASER_CONTROL_ON_LEVEL = 0
|
||||
LASER_CONTROL_OFF_LEVEL = 1
|
||||
LASER_ON_CMD = bytes([0xAA, MODULE_ADDR, 0x01, 0xBE, 0x00, 0x01, 0x00, 0x01, 0xC1])
|
||||
LASER_OFF_CMD = bytes([0xAA, MODULE_ADDR, 0x01, 0xBE, 0x00, 0x01, 0x00, 0x00, 0xC0])
|
||||
DISTANCE_QUERY_CMD = bytes([0xAA, MODULE_ADDR, 0x00, 0x20, 0x00, 0x01, 0x00, 0x00, 0x21]) # 激光测距查询命令
|
||||
@@ -348,11 +353,29 @@ PIN_MAPPINGS = {
|
||||
"A28": "UART2_TX",
|
||||
"A15": "I2C5_SCL",
|
||||
"A27": "I2C5_SDA",
|
||||
"A14": "GPIOA14", # 激光开关:低开、高关
|
||||
"A24": "GPIOA24", # 电源板关机控制
|
||||
"A25": "GPIOA25", # 电源状态绿灯
|
||||
"A23": "GPIOA23", # 电源状态红灯
|
||||
}
|
||||
|
||||
# ==================== 电源配置 ====================
|
||||
AUTO_POWER_OFF_IN_SECONDS = 10 * 60 # 自动关机时间(秒),0表示不自动关机
|
||||
# 充电时自动关机暂时禁用;需要恢复时改为 True。
|
||||
CHARGING_AUTO_POWER_OFF_ENABLED = False
|
||||
|
||||
# 一代电源控制:A24 由电源板负责按键/关机信号,软件关机时输出高电平。
|
||||
|
||||
# 电源状态指示灯
|
||||
STATUS_LED_GREEN_GPIO = "GPIOA25"
|
||||
STATUS_LED_RED_GPIO = "GPIOA23"
|
||||
STATUS_LED_GREEN_ENABLED = True
|
||||
STATUS_LED_RED_ENABLED = True
|
||||
STATUS_LED_ACTIVE_LEVEL = 1
|
||||
STATUS_LED_LOW_BATTERY_PERCENT = 10
|
||||
STATUS_LED_FULL_BATTERY_PERCENT = 90
|
||||
STATUS_LED_CHARGING_BLINK_MS = 500
|
||||
STATUS_LED_POLL_MS = 1000
|
||||
|
||||
BATTERY_SOC_LPF_ALPHA = 0.5
|
||||
BATTERY_SOC_AVG_WINDOW = 5
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Run independently and keep A24 at a high logic level."""
|
||||
|
||||
from maix import app, gpio, pinmap, time
|
||||
|
||||
|
||||
PIN = "A17"
|
||||
GPIO_NAME = "GPIOA17"
|
||||
|
||||
|
||||
def main():
|
||||
pinmap.set_pin_function(PIN, GPIO_NAME)
|
||||
output = gpio.GPIO(GPIO_NAME, gpio.Mode.OUT)
|
||||
output.value(1)
|
||||
print(f"{PIN} is HIGH. Stop the script to set it LOW.")
|
||||
|
||||
try:
|
||||
while not app.need_exit():
|
||||
# Refresh the output in case another component changes its state.
|
||||
output.value(1)
|
||||
time.sleep_ms(100)
|
||||
finally:
|
||||
output.value(0)
|
||||
print(f"{PIN} is LOW.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+79
-1
@@ -5,6 +5,7 @@
|
||||
提供硬件对象的统一管理和访问
|
||||
"""
|
||||
from maix import time
|
||||
import _thread
|
||||
import config
|
||||
from at_client import ATClient
|
||||
|
||||
@@ -28,6 +29,7 @@ class HardwareManager:
|
||||
self._bus = None # I2C总线
|
||||
self._adc_obj = None # ADC对象
|
||||
self._at_client = None # AT客户端
|
||||
self._status_led_monitor_started = False
|
||||
|
||||
self._last_active_time = 0 # 用于记录用户的最后一次活跃的时间
|
||||
self._stop_timer = False # 用于停止定时器的标志
|
||||
@@ -104,11 +106,87 @@ class HardwareManager:
|
||||
# 物理引脚是 A24,对应 GPIO 功能是 GPIOA24
|
||||
# 注意:这里需要先在 config.PIN_MAPPINGS 中配置好 "A24": "GPIOA24"
|
||||
from maix import gpio
|
||||
# 输出高电平关闭
|
||||
# 一代电源板关机信号为高电平
|
||||
gpio.GPIO("GPIOA24", gpio.Mode.OUT).value(1)
|
||||
except Exception as e:
|
||||
print(f"关机失败: {e}")
|
||||
|
||||
def start_status_led_monitor(self):
|
||||
"""后台更新状态灯:正常/充满绿常亮、充电绿闪烁、低电量红常亮。"""
|
||||
if self._status_led_monitor_started:
|
||||
return
|
||||
self._status_led_monitor_started = True
|
||||
_thread.start_new_thread(self._status_led_loop, ())
|
||||
|
||||
def _status_led_loop(self):
|
||||
from maix import gpio
|
||||
from power import get_bus_voltage, is_charging, voltage_to_percent
|
||||
|
||||
try:
|
||||
green = None
|
||||
if getattr(config, "STATUS_LED_GREEN_ENABLED", True):
|
||||
green = gpio.GPIO(config.STATUS_LED_GREEN_GPIO, gpio.Mode.OUT)
|
||||
red = None
|
||||
if getattr(config, "STATUS_LED_RED_ENABLED", True):
|
||||
red = gpio.GPIO(config.STATUS_LED_RED_GPIO, gpio.Mode.OUT)
|
||||
active = int(config.STATUS_LED_ACTIVE_LEVEL)
|
||||
inactive = 0 if active else 1
|
||||
if green is not None:
|
||||
green.value(inactive)
|
||||
if red is not None:
|
||||
red.value(inactive)
|
||||
last_state = None
|
||||
blink_on = False
|
||||
blink_period = max(100, int(config.STATUS_LED_CHARGING_BLINK_MS))
|
||||
poll_ms = max(100, int(config.STATUS_LED_POLL_MS))
|
||||
tick_ms = min(blink_period, poll_ms)
|
||||
sensor_elapsed = poll_ms
|
||||
blink_elapsed = blink_period
|
||||
state = "normal"
|
||||
percent = None
|
||||
charging = False
|
||||
|
||||
while self._status_led_monitor_started:
|
||||
if sensor_elapsed >= poll_ms:
|
||||
voltage = get_bus_voltage()
|
||||
percent = voltage_to_percent(voltage) if voltage > 0 else None
|
||||
charging = is_charging()
|
||||
low = percent is not None and percent <= int(config.STATUS_LED_LOW_BATTERY_PERCENT)
|
||||
full = percent is not None and percent >= int(config.STATUS_LED_FULL_BATTERY_PERCENT)
|
||||
if charging:
|
||||
state = "full" if full else "charging"
|
||||
else:
|
||||
state = "low" if low else "normal"
|
||||
sensor_elapsed = 0
|
||||
|
||||
if state == "low":
|
||||
if green is not None:
|
||||
green.value(inactive)
|
||||
if red is not None:
|
||||
red.value(active)
|
||||
elif state == "charging":
|
||||
if blink_elapsed >= blink_period:
|
||||
blink_on = not blink_on
|
||||
blink_elapsed = 0
|
||||
if green is not None:
|
||||
green.value(active if blink_on else inactive)
|
||||
if red is not None:
|
||||
red.value(inactive)
|
||||
else: # normal or full
|
||||
if green is not None:
|
||||
green.value(active)
|
||||
if red is not None:
|
||||
red.value(inactive)
|
||||
if state != last_state:
|
||||
print(f"[STATUS_LED] state={state} percent={percent} charging={charging}")
|
||||
last_state = state
|
||||
time.sleep_ms(tick_ms)
|
||||
sensor_elapsed += tick_ms
|
||||
blink_elapsed += tick_ms
|
||||
except Exception as e:
|
||||
self._status_led_monitor_started = False
|
||||
print(f"[STATUS_LED] monitor failed: {e}")
|
||||
|
||||
def start_idle_timer(self):
|
||||
self._stop_timer = False
|
||||
self._last_active_time = time.time()
|
||||
|
||||
+44
-72
@@ -31,6 +31,7 @@ class LaserManager:
|
||||
|
||||
# 私有状态
|
||||
self._serial = None # 激光串口,由 laser_manager 自己持有
|
||||
self._laser_gpio = None # A14 激光开关,低电平开启、高电平关闭
|
||||
self._calibration_active = False
|
||||
self._calibration_result = None
|
||||
self._calibration_lock = threading.Lock()
|
||||
@@ -69,10 +70,21 @@ class LaserManager:
|
||||
|
||||
# ==================== 初始化方法 ====================
|
||||
|
||||
def init_control_gpio(self):
|
||||
"""尽早初始化 A14,并拉高确保激光关闭。"""
|
||||
from maix import gpio, pinmap
|
||||
|
||||
pinmap.set_pin_function(config.LASER_CONTROL_PIN, config.LASER_CONTROL_GPIO)
|
||||
if self._laser_gpio is None:
|
||||
self._laser_gpio = gpio.GPIO(config.LASER_CONTROL_GPIO, gpio.Mode.OUT)
|
||||
self._laser_gpio.value(config.LASER_CONTROL_OFF_LEVEL)
|
||||
self._laser_turned_on = False
|
||||
print(f"[LASER] {config.LASER_CONTROL_PIN}=HIGH,激光已关闭")
|
||||
|
||||
def init(self, serial_device=None, baudrate=None):
|
||||
"""
|
||||
初始化激光模块(包括串口)
|
||||
初始化完成后主动发送关闭命令,防止 UART 初始化噪声误触发激光
|
||||
初始化激光模块(A14 开关 + 测距串口)
|
||||
初始化时先将 A14 拉高关闭激光,防止开机误触发
|
||||
|
||||
Args:
|
||||
serial_device: 串口设备路径,默认使用 config.DISTANCE_SERIAL_DEVICE
|
||||
@@ -82,23 +94,11 @@ class LaserManager:
|
||||
device = serial_device or config.DISTANCE_SERIAL_DEVICE
|
||||
baud = baudrate or config.DISTANCE_SERIAL_BAUDRATE
|
||||
|
||||
self.init_control_gpio()
|
||||
|
||||
self._serial = uart.UART(device, baud)
|
||||
print(f"[LASER] 激光串口初始化完成: device={device}, baudrate={baud}")
|
||||
|
||||
# 等待串口稳定后主动关闭激光,防止初始化噪声误触发
|
||||
time.sleep_ms(100)
|
||||
try:
|
||||
self._serial.read(-1) # 清空接收缓冲区
|
||||
except Exception:
|
||||
pass
|
||||
self._serial.write(config.LASER_OFF_CMD)
|
||||
time.sleep_ms(60)
|
||||
try:
|
||||
self._serial.read(-1) # 清空回包
|
||||
except Exception:
|
||||
pass
|
||||
print("[LASER] 已发送关闭命令(防止开机误触发)")
|
||||
|
||||
# ==================== 业务方法 ====================
|
||||
|
||||
def load_laser_point(self):
|
||||
@@ -147,66 +147,38 @@ class LaserManager:
|
||||
return False
|
||||
|
||||
def turn_on_laser(self):
|
||||
"""发送指令开启激光,并读取回包(部分模块支持)"""
|
||||
if self._serial is None:
|
||||
self.logger.error("[LASER] 激光串口未初始化,请先调用 init()")
|
||||
return None
|
||||
|
||||
# 打印调试信息
|
||||
self.logger.info(f"[LASER] 发送开启命令: {config.LASER_ON_CMD.hex()}")
|
||||
|
||||
# 清空接收缓冲区
|
||||
"""A14 输出低电平,开启激光。"""
|
||||
if self._laser_gpio is None:
|
||||
if self.logger:
|
||||
self.logger.error("[LASER] A14 GPIO 未初始化,请先调用 init()")
|
||||
return False
|
||||
try:
|
||||
self._serial.read(-1) # 清空缓冲区
|
||||
except:
|
||||
pass
|
||||
|
||||
# 发送命令
|
||||
written = self._serial.write(config.LASER_ON_CMD)
|
||||
self.logger.info(f"[LASER] 写入字节数: {written}")
|
||||
|
||||
time.sleep_ms(60)
|
||||
|
||||
# 读取回包
|
||||
resp = self._serial.read(len=20, timeout=10)
|
||||
if resp:
|
||||
self.logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
|
||||
if resp == config.LASER_ON_CMD:
|
||||
self.logger.info("✅ 激光开启指令已确认")
|
||||
else:
|
||||
self.logger.warning("🔇 无回包(可能正常或模块不支持回包)")
|
||||
self._laser_turned_on = True
|
||||
return resp
|
||||
self._laser_gpio.value(config.LASER_CONTROL_ON_LEVEL)
|
||||
self._laser_turned_on = True
|
||||
if self.logger:
|
||||
self.logger.info("[LASER] A14=LOW,激光开启")
|
||||
return True
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"[LASER] A14 开启激光失败: {e}")
|
||||
return False
|
||||
|
||||
def turn_off_laser(self):
|
||||
"""发送指令关闭激光"""
|
||||
if self._serial is None:
|
||||
self.logger.error("[LASER] 激光串口未初始化,请先调用 init()")
|
||||
return None
|
||||
|
||||
# 打印调试信息
|
||||
self.logger.info(f"[LASER] 发送关闭命令: {config.LASER_OFF_CMD.hex()}")
|
||||
|
||||
# 清空接收缓冲区
|
||||
"""A14 输出高电平,关闭激光。"""
|
||||
if self._laser_gpio is None:
|
||||
if self.logger:
|
||||
self.logger.error("[LASER] A14 GPIO 未初始化,请先调用 init()")
|
||||
return False
|
||||
try:
|
||||
self._serial.read(-1)
|
||||
except:
|
||||
pass
|
||||
|
||||
# 发送命令
|
||||
written = self._serial.write(config.LASER_OFF_CMD)
|
||||
self.logger.info(f"[LASER] 写入字节数: {written}")
|
||||
|
||||
time.sleep_ms(60)
|
||||
|
||||
# 读取回包
|
||||
resp = self._serial.read(20)
|
||||
if resp:
|
||||
self.logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
|
||||
else:
|
||||
self.logger.warning("🔇 无回包")
|
||||
self._laser_turned_on = False
|
||||
return resp
|
||||
self._laser_gpio.value(config.LASER_CONTROL_OFF_LEVEL)
|
||||
self._laser_turned_on = False
|
||||
if self.logger:
|
||||
self.logger.info("[LASER] A14=HIGH,激光关闭")
|
||||
return True
|
||||
except Exception as e:
|
||||
if self.logger:
|
||||
self.logger.error(f"[LASER] A14 关闭激光失败: {e}")
|
||||
return False
|
||||
|
||||
def flash_laser(self, duration_ms=1000):
|
||||
"""闪一下激光(非阻塞版本)"""
|
||||
|
||||
@@ -76,12 +76,14 @@ def laser_calibration_worker():
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
time.sleep_ms(1000) # 等待1秒后继续
|
||||
|
||||
def cmd_str():
|
||||
"""主程序入口"""
|
||||
# ==================== 第一阶段:硬件初始化 ====================
|
||||
# 按照 main104.py 的顺序,先完成所有硬件初始化
|
||||
|
||||
# 开机第一步先拉高 A14 关闭激光,避免其他硬件初始化期间误亮。
|
||||
laser_manager.init_control_gpio()
|
||||
|
||||
# 1. 引脚功能映射
|
||||
for pin, func in config.PIN_MAPPINGS.items():
|
||||
try:
|
||||
@@ -103,6 +105,8 @@ def cmd_str():
|
||||
print(f"[BOOT] init_ina226 开始 wall_s={_w_boot:.3f}")
|
||||
init_ina226()
|
||||
print(f"[BOOT] init_ina226 结束 wall +{int(round((wall_time.time() - _w_boot) * 1000))} ms")
|
||||
# 启动 A25 绿灯和 A23 红灯状态指示。
|
||||
hardware_manager.start_status_led_monitor()
|
||||
|
||||
# 4. 初始化显示和相机
|
||||
_w_boot = wall_time.time()
|
||||
|
||||
+1
-1
@@ -2146,7 +2146,7 @@ class NetworkManager:
|
||||
}
|
||||
self.safe_enqueue(battery_data, 2)
|
||||
self.logger.info(f"电量上报: {battery_percent}% 充电: {is_charging()}")
|
||||
if is_charging():
|
||||
if getattr(config, "CHARGING_AUTO_POWER_OFF_ENABLED", False) and is_charging():
|
||||
self.safe_enqueue(
|
||||
{
|
||||
"cmd": 700,
|
||||
|
||||
@@ -0,0 +1,603 @@
|
||||
#!/usr/bin/env python3
|
||||
# PYTHON_ARGCOMPLETE_OK
|
||||
|
||||
import sys
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import os.path
|
||||
import collections
|
||||
import uuid
|
||||
import argparse
|
||||
import tarfile
|
||||
import io
|
||||
from struct import pack, unpack
|
||||
|
||||
|
||||
PYTHON_MIN_VERSION = (3, 5, 2) # Ubuntu 16.04 LTS contains Python v3.5.2 by default
|
||||
|
||||
|
||||
if sys.version_info < PYTHON_MIN_VERSION:
|
||||
print("Python >= %r is required" % (PYTHON_MIN_VERSION,))
|
||||
sys.exit(-1)
|
||||
|
||||
|
||||
try:
|
||||
import coloredlogs
|
||||
except ImportError:
|
||||
coloredlogs = None
|
||||
|
||||
try:
|
||||
import argcomplete
|
||||
except ImportError:
|
||||
argcomplete = None
|
||||
|
||||
TOC_HEADER_NAME = 0xAA640001
|
||||
FIP_MAX_SIZE = 0xA0000
|
||||
FIP_ALIGN_SIZE = 2 * 1024
|
||||
ENTRY_SIZE = 0x28
|
||||
|
||||
IV_ZERO = b"\0" * 16
|
||||
|
||||
|
||||
class FIP_HEADER_FLAG:
|
||||
BitRange = collections.namedtuple("BitRange", "shift, bits")
|
||||
|
||||
REE_SCS = BitRange(0, 2)
|
||||
REE_ENCRYPTION = BitRange(2, 2)
|
||||
|
||||
@classmethod
|
||||
def test(cls, value, flag):
|
||||
v = value >> flag.shift
|
||||
v &= (1 << flag.bits) - 1
|
||||
return v
|
||||
|
||||
@classmethod
|
||||
def value(cls, flag):
|
||||
v = (1 << flag.bits) - 1
|
||||
v <<= flag.shift
|
||||
return v
|
||||
|
||||
|
||||
class FIP_UUID:
|
||||
# from arm-trusted-firmware/include/tools_share/firmware_image_package.h
|
||||
uuid_c_define = """
|
||||
/* ToC Entry UUIDs */
|
||||
#define UUID_LICENSE_FILE \
|
||||
{0x25360c62, 0x5151, 0x48ad, 0xb5, 0x91, {0x2d, 0x35, 0x67, 0x26, 0x85, 0xa5} }
|
||||
#define UUID_TRUSTED_UPDATE_FIRMWARE_SCP_BL2U \
|
||||
{0x03279265, 0x742f, 0x44e6, 0x8d, 0xff, {0x57, 0x9a, 0xc1, 0xff, 0x06, 0x10} }
|
||||
#define UUID_TRUSTED_UPDATE_FIRMWARE_BL2U \
|
||||
{0x37ebb360, 0xe5c1, 0x41ea, 0x9d, 0xf3, {0x19, 0xed, 0xa1, 0x1f, 0x68, 0x01} }
|
||||
#define UUID_TRUSTED_UPDATE_FIRMWARE_NS_BL2U \
|
||||
{0x111d514f, 0xe52b, 0x494e, 0xb4, 0xc5, {0x83, 0xc2, 0xf7, 0x15, 0x84, 0x0a} }
|
||||
#define UUID_TRUSTED_FWU_CERT \
|
||||
{0xb28a4071, 0xd618, 0x4c87, 0x8b, 0x2e, {0xc6, 0xdc, 0xcd, 0x50, 0xf0, 0x96} }
|
||||
#define UUID_TRUSTED_BOOT_FIRMWARE_BL2 \
|
||||
{0x0becf95f, 0x224d, 0x4d3e, 0xa5, 0x44, {0xc3, 0x9d, 0x81, 0xc7, 0x3f, 0x0a} }
|
||||
#define UUID_BLD \
|
||||
{0x3dfd6697, 0xbe89, 0x49e8, 0xae, 0x5d, {0x78, 0xa1, 0x40, 0x60, 0x82, 0x13} }
|
||||
#define UUID_EL3_RUNTIME_FIRMWARE_BL31 \
|
||||
{0x6d08d447, 0xfe4c, 0x4698, 0x9b, 0x95, {0x29, 0x50, 0xcb, 0xbd, 0x5a, 0x00} }
|
||||
#define UUID_SECURE_PAYLOAD_BL32 \
|
||||
{0x89e1d005, 0xdc53, 0x4713, 0x8d, 0x2b, {0x50, 0x0a, 0x4b, 0x7a, 0x3e, 0x38} }
|
||||
#define UUID_NON_TRUSTED_FIRMWARE_BL33 \
|
||||
{0xa7eed0d6, 0xeafc, 0x4bd5, 0x97, 0x82, {0x99, 0x34, 0xf2, 0x34, 0xb6, 0xe4} }
|
||||
/* Key certificates */
|
||||
#define UUID_ROT_KEY_CERT \
|
||||
{0x721d2d86, 0x60f8, 0x11e4, 0x92, 0x0b, {0x8b, 0xe7, 0x62, 0x16, 0x0f, 0x24} }
|
||||
#define UUID_BLD1_KEY_CERT \
|
||||
{0x90e87e82, 0x60f8, 0x11e4, 0xa1, 0xb4, {0x77, 0x7a, 0x21, 0xb4, 0xf9, 0x4c} }
|
||||
#define UUID_BLD2_KEY_CERT \
|
||||
{0xa1214202, 0x60f8, 0x11e4, 0x8d, 0x9b, {0xf3, 0x3c, 0x0e, 0x15, 0xa0, 0x14} }
|
||||
#define UUID_SOC_FW_KEY_CERT \
|
||||
{0xccbeb88a, 0x60f9, 0x11e4, 0x9a, 0xd0, {0xeb, 0x48, 0x22, 0xd8, 0xdc, 0xf8} }
|
||||
#define UUID_TRUSTED_OS_FW_KEY_CERT \
|
||||
{0x03d67794, 0x60fb, 0x11e4, 0x85, 0xdd, {0xb7, 0x10, 0x5b, 0x8c, 0xee, 0x04} }
|
||||
#define UUID_BL33_KEY_CERT \
|
||||
{0x2a83d58a, 0x60fb, 0x11e4, 0x8a, 0xaf, {0xdf, 0x30, 0xbb, 0xc4, 0x98, 0x59} }
|
||||
/* Content certificates */
|
||||
#define UUID_TRUSTED_BOOT_FW_CERT \
|
||||
{0xea69e2d6, 0x635d, 0x11e4, 0x8d, 0x8c, {0x9f, 0xba, 0xbe, 0x99, 0x56, 0xa5} }
|
||||
#define UUID_BLD_CONTENT_CERT \
|
||||
{0x046fbe44, 0x635e, 0x11e4, 0xb2, 0x8b, {0x73, 0xd8, 0xea, 0xae, 0x96, 0x56} }
|
||||
#define UUID_SOC_FW_CONTENT_CERT \
|
||||
{0x200cb2e2, 0x635e, 0x11e4, 0x9c, 0xe8, {0xab, 0xcc, 0xf9, 0x2b, 0xb6, 0x66} }
|
||||
#define UUID_TRUSTED_OS_FW_CONTENT_CERT \
|
||||
{0x11449fa4, 0x635e, 0x11e4, 0x87, 0x28, {0x3f, 0x05, 0x72, 0x2a, 0xf3, 0x3d} }
|
||||
#define UUID_BL33_CONTENT_CERT \
|
||||
{0xf3c1c48e, 0x635d, 0x11e4, 0xa7, 0xa9, {0x87, 0xee, 0x40, 0xb2, 0x3f, 0xa7} }
|
||||
/* CV keys */
|
||||
#define UUID_CV_TRUSTED_KEY_CERT \
|
||||
{0x64fbfc49, 0x4b8c, 0x4ad3, 0xb9, 0x92, {0x93, 0x55, 0x89, 0xee, 0xf0, 0x12} }
|
||||
#define UUID_CV_NON_TRUSTED_KEY_CERT \
|
||||
{0xcb48bf0d, 0x7012, 0x4201, 0xbc, 0x35, {0x8a, 0x51, 0xc4, 0x90, 0x90, 0x94} }
|
||||
|
||||
/* DDR init*/
|
||||
#define UUID_CV_DDRINIT_KEY_CERT \
|
||||
{0xa61c53c9, 0x886c, 0x484f, 0x96, 0x5d, {0xd2, 0xda, 0xd7, 0xc3, 0xeb, 0x13} }
|
||||
#define UUID_CV_DDRINIT_CONTENT_CERT \
|
||||
{0x9dfaabd2, 0x7f1b, 0x47e6, 0xa8, 0xa6, {0x6a, 0xc3, 0x10, 0xcc, 0xac, 0x91} }
|
||||
#define UUID_CV_DDRINIT \
|
||||
{0x5888a5cd, 0x38fc, 0x4f66, 0xae, 0x3d, {0x2e, 0x18, 0x6d, 0x69, 0x41, 0xfb} }
|
||||
|
||||
/* Fast boot */
|
||||
#define UUID_CV_FASTBOOT_KEY_CERT \
|
||||
{0x285df54e, 0x7b50, 0x4309, 0x9b, 0x52, {0x4b, 0xc4, 0x92, 0x82, 0x60, 0xdd} }
|
||||
#define UUID_CV_FASTBOOT_CONTENT_CERT \
|
||||
{0x61f7595b, 0x8d77, 0x4e13, 0x91, 0x2a, {0x63, 0x6e, 0x58, 0xda, 0x5b, 0x69} }
|
||||
#define UUID_CV_FASTBOOT \
|
||||
{0x43766198, 0xc363, 0x48db, 0xa9, 0x97, {0xf1, 0x0e, 0x93, 0x80, 0x4f, 0xea} }
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def cls_init(cls):
|
||||
txt = cls.uuid_c_define
|
||||
txt = txt.replace("\r\n", "\n")
|
||||
txt = txt.replace("\\\n", "\n")
|
||||
rx = r"""
|
||||
\#define\s+
|
||||
(?P<name>\S+)\s+
|
||||
{
|
||||
\s*(?P<u0>0x\S+)\s*,\s*
|
||||
\s*(?P<u1>0x\S+)\s*,\s*
|
||||
\s*(?P<u2>0x\S+)\s*,\s*
|
||||
\s*(?P<u3>0x\S+)\s*,\s*
|
||||
\s*(?P<u4>0x\S+)\s*,\s*
|
||||
{
|
||||
\s*(?P<u5>0x\S+)\s*,\s*
|
||||
\s*(?P<u6>0x\S+)\s*,\s*
|
||||
\s*(?P<u7>0x\S+)\s*,\s*
|
||||
\s*(?P<u8>0x\S+)\s*,\s*
|
||||
\s*(?P<u9>0x\S+)\s*,\s*
|
||||
\s*(?P<u10>0x\S+)\s*
|
||||
}\s*,?\s*
|
||||
}
|
||||
"""
|
||||
for m in re.finditer(rx, txt, flags=re.X):
|
||||
name = m.group("name")
|
||||
u = m.group(*["u%d" % i for i in range(11)])
|
||||
u = [int(i, 0) for i in u]
|
||||
u = pack("<IHHBBBBBBBB", *u)
|
||||
u = uuid.UUID(bytes=u)
|
||||
setattr(cls, name, u)
|
||||
|
||||
|
||||
class Entry:
|
||||
__slots__ = ["name", "loc", "uuid", "address", "flag", "content"]
|
||||
|
||||
def __init__(self):
|
||||
self.loc = 0
|
||||
self.uuid = uuid.UUID(int=0)
|
||||
self.address = 0
|
||||
self.flag = 0
|
||||
self.content = b""
|
||||
|
||||
@classmethod
|
||||
def make(cls, uuid, content):
|
||||
entry = cls()
|
||||
entry.uuid = uuid
|
||||
entry.content = content
|
||||
return entry
|
||||
|
||||
@classmethod
|
||||
def from_fip(cls, name, loc, fip_bin):
|
||||
data = fip_bin[loc : loc + ENTRY_SIZE]
|
||||
uuid_bytes, address, size, flag = unpack("<16sQQQ", data)
|
||||
content = fip_bin[address : address + size]
|
||||
|
||||
entry = cls()
|
||||
entry.name = name
|
||||
entry.loc = loc
|
||||
entry.uuid = uuid.UUID(bytes=uuid_bytes)
|
||||
entry.address = address
|
||||
entry.flag = flag
|
||||
entry.content = content
|
||||
return entry
|
||||
|
||||
def to_bytes(self):
|
||||
return pack("<16sQQQ", self.uuid.bytes, self.address, self.size, self.flag)
|
||||
|
||||
@property
|
||||
def size(self):
|
||||
return len(self.content)
|
||||
|
||||
@property
|
||||
def end(self):
|
||||
return self.address + self.size
|
||||
|
||||
def __str__(self):
|
||||
return "<%-31s loc=0x%03x U=%s a=0x%05x,0x%05x,0x%05x f=0x%x>" % (
|
||||
self.name,
|
||||
self.loc,
|
||||
self.uuid.hex[:8],
|
||||
self.address,
|
||||
self.end,
|
||||
self.size,
|
||||
self.flag,
|
||||
)
|
||||
|
||||
|
||||
class FIP:
|
||||
ENTRY_NAMES = collections.OrderedDict(
|
||||
[
|
||||
("LICENSE_FILE", "UUID_LICENSE_FILE"),
|
||||
("BL2", "UUID_TRUSTED_BOOT_FIRMWARE_BL2"),
|
||||
("BLD", "UUID_BLD"),
|
||||
("BL31", "UUID_EL3_RUNTIME_FIRMWARE_BL31"),
|
||||
("BL32", "UUID_SECURE_PAYLOAD_BL32"),
|
||||
("BL33", "UUID_NON_TRUSTED_FIRMWARE_BL33"),
|
||||
("BLD1_KEY_CERT", "UUID_BLD1_KEY_CERT"),
|
||||
("BLD2_KEY_CERT", "UUID_BLD2_KEY_CERT"),
|
||||
("CV_TRUSTED_KEY_CERT", "UUID_CV_TRUSTED_KEY_CERT"),
|
||||
("SOC_FW_KEY_CERT", "UUID_SOC_FW_KEY_CERT"),
|
||||
("TRUSTED_OS_FW_KEY_CERT", "UUID_TRUSTED_OS_FW_KEY_CERT"),
|
||||
("CV_NON_TRUSTED_KEY_CERT", "UUID_CV_NON_TRUSTED_KEY_CERT"),
|
||||
("BL33_KEY_CERT", "UUID_BL33_KEY_CERT"),
|
||||
("TRUSTED_BOOT_FW_CERT", "UUID_TRUSTED_BOOT_FW_CERT"),
|
||||
("BLD_CONTENT_CERT", "UUID_BLD_CONTENT_CERT"),
|
||||
("SOC_FW_CONTENT_CERT", "UUID_SOC_FW_CONTENT_CERT"),
|
||||
("TRUSTED_OS_FW_CONTENT_CERT", "UUID_TRUSTED_OS_FW_CONTENT_CERT"),
|
||||
("BL33_CONTENT_CERT", "UUID_BL33_CONTENT_CERT"),
|
||||
("CV_DDRINIT", "UUID_CV_DDRINIT"),
|
||||
("CV_FASTBOOT", "UUID_CV_FASTBOOT"),
|
||||
]
|
||||
)
|
||||
|
||||
TOC_Header = collections.namedtuple(
|
||||
"TOC_Header", "name, serial, flag_res, flag_plat, flag_res2"
|
||||
)
|
||||
|
||||
def __init__(self, path):
|
||||
logging.info("FIP_BIN: %s", path)
|
||||
self.path = path
|
||||
|
||||
def load(self):
|
||||
with open(self.path, "rb") as fp:
|
||||
self.binary = fp.read(FIP_MAX_SIZE)
|
||||
logging.info("%s is %d bytes", self.path, len(self.binary))
|
||||
|
||||
self.header = self.TOC_Header(*unpack("<IIIHH", self.binary[0x00:0x10]))
|
||||
if self.header.name != TOC_HEADER_NAME:
|
||||
raise ValueError(
|
||||
"FIP header is 0x%08x but should be 0x%08x"
|
||||
% (self.header[0], TOC_HEADER_NAME)
|
||||
)
|
||||
|
||||
logging.info("TOC header: flag_plat=0x%04x", self.header.flag_plat)
|
||||
logging.info(
|
||||
" REE_SCS: %r",
|
||||
FIP_HEADER_FLAG.test(self.header.flag_plat, FIP_HEADER_FLAG.REE_SCS),
|
||||
)
|
||||
logging.info(
|
||||
" REE_ENCRYPTION: %r",
|
||||
FIP_HEADER_FLAG.test(self.header.flag_plat, FIP_HEADER_FLAG.REE_ENCRYPTION),
|
||||
)
|
||||
|
||||
ents = []
|
||||
for k, v in self.ENTRY_NAMES.items():
|
||||
try:
|
||||
ents.append((k, self.find_entry(v)))
|
||||
except ValueError as err:
|
||||
logging.warning("%s", err)
|
||||
|
||||
ents.sort(key=lambda x: x[1].address)
|
||||
for n, (k, v) in enumerate(ents):
|
||||
logging.debug("%s", v)
|
||||
if n > 0:
|
||||
pk, pv = ents[n - 1]
|
||||
if v.loc != pv.loc + ENTRY_SIZE or v.address != pv.address + pv.size:
|
||||
raise Exception("Invalid FIP")
|
||||
|
||||
rest = self.binary[ents[-1][1].end :]
|
||||
loc = rest.find(b"APLB")
|
||||
if loc < 0:
|
||||
raise Exception("No BLD/DDRC")
|
||||
self.blp_ddrc_binary = rest[loc:]
|
||||
logging.debug("blp_ddrc: 0x%04x at 0x%08x", len(self.blp_ddrc_binary), loc)
|
||||
|
||||
self.ents = collections.OrderedDict(ents)
|
||||
|
||||
def make_fip(self, output_path=None):
|
||||
logging.info("New TOC header: flag_plat=0x%04x", self.header.flag_plat)
|
||||
header_bin = pack("<IIIHH", *self.header)
|
||||
fip_bin = header_bin
|
||||
|
||||
# Sort self.ents by the order of FIP.ENTRY_NAMES
|
||||
sorted_ents = collections.OrderedDict()
|
||||
for name in self.ENTRY_NAMES:
|
||||
try:
|
||||
sorted_ents[name] = self.ents[name]
|
||||
except KeyError:
|
||||
pass
|
||||
self.ents = sorted_ents
|
||||
|
||||
offset = (len(self.ents) + 1) * ENTRY_SIZE + 0x10
|
||||
|
||||
for k, v in self.ents.items():
|
||||
v.address = offset
|
||||
fip_bin += v.to_bytes()
|
||||
offset += v.size
|
||||
|
||||
null_entry = Entry()
|
||||
null_entry.address = offset
|
||||
fip_bin += null_entry.to_bytes()
|
||||
|
||||
for k, v in self.ents.items():
|
||||
fip_bin += v.content
|
||||
|
||||
if (len(fip_bin) % FIP_ALIGN_SIZE) > 0:
|
||||
fip_bin += b"\x00" * (FIP_ALIGN_SIZE - len(fip_bin) % FIP_ALIGN_SIZE)
|
||||
fip_bin += self.blp_ddrc_binary
|
||||
|
||||
if output_path:
|
||||
path = output_path
|
||||
else:
|
||||
path = os.path.splitext(self.path)
|
||||
path = path[0] + "_signed_encrypted" + path[1]
|
||||
logging.info("Save new FIP image to %s", path)
|
||||
with open(path, "wb") as fp:
|
||||
fp.write(fip_bin)
|
||||
|
||||
def dump_uuids(self):
|
||||
for k, v in vars(FIP_UUID).items():
|
||||
if k.startswith("UUID_"):
|
||||
print("%-38s" % k, v.hex)
|
||||
|
||||
def find_entry(self, name):
|
||||
# UUID=0, offset=any, size=0, flags=0
|
||||
nullm = re.search(rb"\0{16}.{8}\0{16}", self.binary, flags=re.DOTALL)
|
||||
if nullm is None:
|
||||
raise Exception("NULL TOC entry is not found")
|
||||
|
||||
max_toc_size = nullm.start(0)
|
||||
uuid = getattr(FIP_UUID, name)
|
||||
loc = self.binary.find(uuid.bytes, 0, max_toc_size)
|
||||
if loc < 0:
|
||||
raise ValueError("%s is not found" % name)
|
||||
return Entry.from_fip(name, loc, self.binary)
|
||||
|
||||
|
||||
def entry(args):
|
||||
logging.debug("cmd_fip")
|
||||
|
||||
|
||||
def init_logging(log_file=None, file_level="DEBUG", stdout_level="WARNING"):
|
||||
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.NOTSET)
|
||||
|
||||
fmt = "%(asctime)s %(levelname)8s:%(name)s:%(message)s"
|
||||
|
||||
if log_file is not None:
|
||||
file_handler = logging.FileHandler(log_file, encoding="utf-8")
|
||||
file_handler.setFormatter(logging.Formatter(fmt))
|
||||
file_handler.setLevel(file_level)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
if coloredlogs:
|
||||
os.environ["COLOREDLOGS_DATE_FORMAT"] = "%H:%M:%S"
|
||||
|
||||
field_styles = {
|
||||
"asctime": {"color": "green"},
|
||||
"hostname": {"color": "magenta"},
|
||||
"levelname": {"color": "black", "bold": True},
|
||||
"name": {"color": "blue"},
|
||||
"programname": {"color": "cyan"},
|
||||
}
|
||||
|
||||
level_styles = coloredlogs.DEFAULT_LEVEL_STYLES
|
||||
level_styles["debug"]["color"] = "cyan"
|
||||
|
||||
coloredlogs.install(
|
||||
level=stdout_level,
|
||||
fmt=fmt,
|
||||
field_styles=field_styles,
|
||||
level_styles=level_styles,
|
||||
milliseconds=True,
|
||||
)
|
||||
|
||||
|
||||
def parse_fip(fip_path):
|
||||
logging.debug("parse_fip: %s", fip_path)
|
||||
fip = FIP(fip_path)
|
||||
fip.load()
|
||||
|
||||
|
||||
def unpack_fip(fip_path):
|
||||
logging.debug("unpack_fip: %s", fip_path)
|
||||
fip = FIP(fip_path)
|
||||
fip.load()
|
||||
|
||||
def save(name, content):
|
||||
fn = os.path.splitext(fip_path)
|
||||
fn = "%s_%s%s" % (fn[0], name, fn[1])
|
||||
logging.info("Save %s", fn)
|
||||
with open(fn, "wb") as fp:
|
||||
fp.write(content)
|
||||
|
||||
for k, v in fip.ents.items():
|
||||
save(k, v.content)
|
||||
|
||||
save("BLP_DDRC", fip.blp_ddrc_binary)
|
||||
|
||||
|
||||
def tar_bld(fip_path, output_path, multibin):
|
||||
logging.debug("tar_bld: %s multibin=%r", fip_path, multibin)
|
||||
fip = FIP(fip_path)
|
||||
fip.load()
|
||||
|
||||
members = [
|
||||
"BLD_CONTENT_CERT",
|
||||
"BLD2_KEY_CERT",
|
||||
"BLD1_KEY_CERT",
|
||||
"CV_DDRINIT" if multibin else "BLD",
|
||||
]
|
||||
|
||||
if not output_path:
|
||||
output_path = os.path.join(os.path.dirname(fip_path), "bld.tar")
|
||||
|
||||
logging.info("bld_tar_path=%s", output_path)
|
||||
|
||||
with tarfile.open(output_path, "w") as tf:
|
||||
for m in members:
|
||||
logging.debug("Tar %s", m)
|
||||
try:
|
||||
fp = io.BytesIO(fip.ents[m].content)
|
||||
except KeyError:
|
||||
logging.warning("%s doesn't exist", m)
|
||||
continue
|
||||
info = tarfile.TarInfo(name=m + ".bin")
|
||||
info.size = len(fp.getbuffer())
|
||||
tf.addfile(tarinfo=info, fileobj=fp)
|
||||
|
||||
|
||||
def merge_fip(fip_path, inputs, output_path):
|
||||
logging.debug("merge_fip: %s", fip_path)
|
||||
fip = FIP(fip_path)
|
||||
fip.load()
|
||||
|
||||
for name in FIP.ENTRY_NAMES:
|
||||
binary = inputs.get(name)
|
||||
if not binary:
|
||||
continue
|
||||
|
||||
logging.debug("merge %s", name)
|
||||
ent = fip.ents.get(name)
|
||||
if ent:
|
||||
ent.content = binary
|
||||
else:
|
||||
ent = Entry.make(getattr(FIP_UUID, "UUID_" + name), binary)
|
||||
fip.ents[name] = ent
|
||||
|
||||
binary = inputs.get("BLP_DDRC")
|
||||
if binary:
|
||||
fip.blp_ddrc_binary = binary
|
||||
|
||||
if not output_path:
|
||||
fn = os.path.splitext(fip_path)
|
||||
fn = "%s_%s%s" % (fn[0], "merged", fn[1])
|
||||
output_path = fn
|
||||
|
||||
fip.make_fip(output_path)
|
||||
|
||||
|
||||
def round_up(n, k):
|
||||
return (n + k - 1) // k * k
|
||||
|
||||
|
||||
def read_blp_and_ddrc(inputs, blp_path, ddrc_path):
|
||||
logging.info("Open %s and %s", blp_path, ddrc_path)
|
||||
with open(blp_path, "rb") as fp:
|
||||
blp_bin = fp.read()
|
||||
|
||||
logging.info("Open %s", ddrc_path)
|
||||
with open(ddrc_path, "rb") as fp:
|
||||
ddrc_bin = fp.read()
|
||||
|
||||
blp_bin += b"\0" * (round_up(len(blp_bin), FIP_ALIGN_SIZE) - len(blp_bin))
|
||||
ddrc_bin += b"\0" * (round_up(len(ddrc_bin), FIP_ALIGN_SIZE) - len(ddrc_bin))
|
||||
|
||||
inputs["BLP_DDRC"] = blp_bin + ddrc_bin
|
||||
|
||||
|
||||
def read_bld_tar(inputs, bld_tar_path, multibin):
|
||||
logging.info("Open %s multibin=%r", bld_tar_path, multibin)
|
||||
members = [
|
||||
"BLD_CONTENT_CERT.bin",
|
||||
"BLD2_KEY_CERT.bin",
|
||||
"BLD1_KEY_CERT.bin",
|
||||
"CV_DDRINIT.bin" if multibin else "BLD.bin",
|
||||
]
|
||||
|
||||
with tarfile.open(bld_tar_path, "r") as tf:
|
||||
for member in members:
|
||||
try:
|
||||
fp = tf.extractfile(member)
|
||||
inputs[os.path.splitext(member)[0]] = fp.read()
|
||||
except KeyError:
|
||||
logging.warning("%s does not exist", member)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="FIP packer")
|
||||
|
||||
for name in FIP.ENTRY_NAMES:
|
||||
parser.add_argument(
|
||||
"--add-%s" % name.lower(),
|
||||
dest=name,
|
||||
type=str,
|
||||
help="Merge %s into FIP" % name,
|
||||
)
|
||||
parser.add_argument(
|
||||
"--add-blp-ddrc", dest="BLP_DDRC", type=str, help="Merge BLP+DDRC into FIP"
|
||||
)
|
||||
parser.add_argument("--add-blp", dest="BLP", type=str, help="Merge BLP into FIP")
|
||||
parser.add_argument("--add-ddrc", dest="DDRC", type=str, help="Merge DDRC into FIP")
|
||||
parser.add_argument(
|
||||
"--add-bld-tar", dest="BLD_TAR", type=str, help="Merge BLD.tar into FIP"
|
||||
)
|
||||
|
||||
parser.add_argument("--multibin", action="store_true", help="Use multibin")
|
||||
|
||||
parser.add_argument("FIP_BIN", type=str, nargs=1, help="Input FIP binary")
|
||||
parser.add_argument("--output", type=str, help="Output filename")
|
||||
|
||||
parser.add_argument(
|
||||
"--version", action="store_true", help="Output version information and exit"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
help="Increase output verbosity",
|
||||
action="store_const",
|
||||
const=logging.DEBUG,
|
||||
default=logging.DEBUG,
|
||||
)
|
||||
|
||||
parser.add_argument("--unpack", action="store_true", help="Unpack FIP.bin")
|
||||
parser.add_argument("--parse", action="store_true", help="Parse FIP.bin")
|
||||
parser.add_argument(
|
||||
"--tar-bld", action="store_true", help="Extrace BLD.bin and tar"
|
||||
)
|
||||
|
||||
if argcomplete:
|
||||
argcomplete.autocomplete(parser)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
init_logging(stdout_level=args.verbose)
|
||||
|
||||
logging.debug("args=%r", args)
|
||||
|
||||
FIP_UUID.cls_init()
|
||||
|
||||
if args.parse:
|
||||
parse_fip(args.FIP_BIN[0])
|
||||
|
||||
if args.unpack:
|
||||
unpack_fip(args.FIP_BIN[0])
|
||||
|
||||
if args.tar_bld:
|
||||
tar_bld(args.FIP_BIN[0], args.output, args.multibin)
|
||||
|
||||
inputs = collections.OrderedDict()
|
||||
for name in list(FIP.ENTRY_NAMES) + ["BLP_DDRC"]:
|
||||
fn = getattr(args, name)
|
||||
if not fn:
|
||||
continue
|
||||
logging.info("Open %s", fn)
|
||||
with open(fn, "rb") as fp:
|
||||
inputs[name] = fp.read()
|
||||
|
||||
if args.BLP or args.DDRC:
|
||||
read_blp_and_ddrc(inputs, args.BLP, args.DDRC)
|
||||
|
||||
if args.BLD_TAR:
|
||||
read_bld_tar(inputs, args.BLD_TAR, args.multibin)
|
||||
|
||||
if len(inputs):
|
||||
merge_fip(args.FIP_BIN[0], inputs, args.output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -7,10 +7,12 @@
|
||||
import config
|
||||
import os
|
||||
import subprocess
|
||||
import _thread
|
||||
from logger_manager import logger_manager
|
||||
from maix import time as maix_time
|
||||
|
||||
_INA226_PRESENT = None
|
||||
_INA226_LOCK = _thread.allocate_lock()
|
||||
|
||||
|
||||
def _ina226_ready() -> bool:
|
||||
@@ -32,7 +34,11 @@ def write_register(reg, value):
|
||||
data = [(value >> 8) & 0xFF, value & 0xFF]
|
||||
# 某些底层驱动在失败时只打印 “write failed” 并返回 -1,而不是抛异常;
|
||||
# 为避免误判“初始化成功”导致后续 readfrom_mem SIGSEGV,这里把失败显式转成异常。
|
||||
ret = hardware_manager.bus.writeto_mem(config.INA226_ADDR, reg, bytes(data))
|
||||
_INA226_LOCK.acquire()
|
||||
try:
|
||||
ret = hardware_manager.bus.writeto_mem(config.INA226_ADDR, reg, bytes(data))
|
||||
finally:
|
||||
_INA226_LOCK.release()
|
||||
if isinstance(ret, int) and ret < 0:
|
||||
if logger:
|
||||
logger.error(f"[INA226] writeto_mem 失败: addr=0x{config.INA226_ADDR:02X} reg=0x{reg:02X} ret={ret}")
|
||||
@@ -42,7 +48,11 @@ def write_register(reg, value):
|
||||
def read_register(reg):
|
||||
"""读取INA226寄存器"""
|
||||
from hardware import hardware_manager
|
||||
data = hardware_manager.bus.readfrom_mem(config.INA226_ADDR, reg, 2)
|
||||
_INA226_LOCK.acquire()
|
||||
try:
|
||||
data = hardware_manager.bus.readfrom_mem(config.INA226_ADDR, reg, 2)
|
||||
finally:
|
||||
_INA226_LOCK.release()
|
||||
return (data[0] << 8) | data[1]
|
||||
|
||||
|
||||
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,62 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Interactive GPIO test for physical pin A14."""
|
||||
|
||||
import sys
|
||||
|
||||
from maix import gpio, pinmap
|
||||
|
||||
|
||||
PIN = "A14"
|
||||
GPIO_NAME = "GPIOA14"
|
||||
|
||||
|
||||
def set_level(output, command):
|
||||
if command == "1":
|
||||
output.value(1)
|
||||
print("A14 = HIGH, laser OFF")
|
||||
return True
|
||||
if command == "0":
|
||||
output.value(0)
|
||||
print("A14 = LOW, laser ON")
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
pinmap.set_pin_function(PIN, GPIO_NAME)
|
||||
output = gpio.GPIO(GPIO_NAME, gpio.Mode.OUT)
|
||||
|
||||
# One-shot mode for SSH/serial shells: python3 test_gpio_a14.py 1|0
|
||||
if len(sys.argv) > 1:
|
||||
command = sys.argv[1].strip()
|
||||
if not set_level(output, command):
|
||||
print("Invalid argument. Use 1 or 0.")
|
||||
return
|
||||
return
|
||||
|
||||
output.value(1)
|
||||
print("A14 laser test: input 0 for ON, 1 for OFF, q to quit.")
|
||||
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
command = input("A14> ").strip().lower()
|
||||
except EOFError:
|
||||
print("This runner has no stdin. Run from an SSH/serial shell with argument 1 or 0.")
|
||||
return
|
||||
if set_level(output, command):
|
||||
continue
|
||||
elif command in ("q", "quit", "exit"):
|
||||
break
|
||||
elif command:
|
||||
print("Invalid input. Use 1, 0, or q.")
|
||||
except KeyboardInterrupt:
|
||||
print()
|
||||
finally:
|
||||
output.value(1)
|
||||
print("A14 = HIGH, laser OFF, test stopped.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,122 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Standalone WiFi/GPIO/INA226 isolation test for the official MaixPy tool.
|
||||
|
||||
This file intentionally does not import project modules or start project
|
||||
threads. Select TEST_MODE below, then run the file directly.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from maix import gpio, i2c, network, pinmap
|
||||
|
||||
|
||||
# Change only this value before each run.
|
||||
# wifi WiFi only
|
||||
# a25 A25 only
|
||||
# a23 A23 only
|
||||
# gpio A23/A26 only
|
||||
# ina INA226 only
|
||||
# gpio_ina GPIOs, then INA226
|
||||
# a25_wifi A25, then WiFi
|
||||
# a23_wifi A23, then WiFi
|
||||
# all GPIOs, INA226, then WiFi
|
||||
TEST_MODE = "a25_wifi"
|
||||
|
||||
WIFI_SSID = "sheling4b02-5G"
|
||||
WIFI_PASSWORD = "Aa12345678"
|
||||
WIFI_TIMEOUT_S = 20
|
||||
I2C_BUS_NUM = 5
|
||||
INA226_ADDR = 0x40
|
||||
|
||||
|
||||
def init_leds():
|
||||
return init_selected_leds(True, True)
|
||||
|
||||
|
||||
def init_selected_leds(use_a26, use_a23):
|
||||
outputs = []
|
||||
if use_a26:
|
||||
print("Initializing A25 -> GPIOA25")
|
||||
pinmap.set_pin_function("A25", "GPIOA25")
|
||||
green = gpio.GPIO("GPIOA25", gpio.Mode.OUT)
|
||||
green.value(0)
|
||||
outputs.append(("GPIOA25", green))
|
||||
print("GPIOA25 initialized LOW")
|
||||
if use_a23:
|
||||
print("Initializing A23 -> GPIOA23")
|
||||
pinmap.set_pin_function("A23", "GPIOA23")
|
||||
red = gpio.GPIO("GPIOA23", gpio.Mode.OUT)
|
||||
red.value(0)
|
||||
outputs.append(("GPIOA23", red))
|
||||
print("GPIOA23 initialized LOW")
|
||||
return outputs
|
||||
|
||||
|
||||
def test_ina226():
|
||||
# Match the board mapping used by the application before opening I2C5.
|
||||
pinmap.set_pin_function("A15", "I2C5_SCL")
|
||||
pinmap.set_pin_function("A27", "I2C5_SDA")
|
||||
print("A15/A27 configured for I2C5")
|
||||
print("Initializing I2C bus", I2C_BUS_NUM)
|
||||
bus = i2c.I2C(I2C_BUS_NUM, i2c.Mode.MASTER)
|
||||
print("Reading INA226 at 0x%02X" % INA226_ADDR)
|
||||
config = bus.readfrom_mem(INA226_ADDR, 0x00, 2)
|
||||
voltage_raw = bus.readfrom_mem(INA226_ADDR, 0x02, 2)
|
||||
voltage = ((voltage_raw[0] << 8) | voltage_raw[1]) * 1.25 / 1000
|
||||
print("INA226 config=0x%02X%02X voltage=%.3fV" % (config[0], config[1], voltage))
|
||||
return bus
|
||||
|
||||
|
||||
def test_wifi():
|
||||
print("Starting MaixPy WiFi connection...")
|
||||
wifi = network.wifi.Wifi()
|
||||
result = wifi.connect(WIFI_SSID, WIFI_PASSWORD, wait=True, timeout=WIFI_TIMEOUT_S)
|
||||
print("WiFi connect result:", result)
|
||||
print("WiFi connected:", wifi.is_connected())
|
||||
try:
|
||||
print("WiFi IP:", wifi.get_ip())
|
||||
except Exception as exc:
|
||||
print("WiFi status query failed:", exc)
|
||||
|
||||
|
||||
def main():
|
||||
valid = ("wifi", "a25", "a23", "gpio", "ina", "gpio_ina", "a25_wifi", "a23_wifi", "all")
|
||||
mode = TEST_MODE.lower()
|
||||
if mode not in valid:
|
||||
print("TEST_MODE must be one of:", ", ".join(valid))
|
||||
return 1
|
||||
|
||||
leds = []
|
||||
try:
|
||||
print("=== Standalone WiFi/GPIO/INA226 isolation ===")
|
||||
print("mode:", mode)
|
||||
if mode in ("a25", "a25_wifi"):
|
||||
leds = init_selected_leds(True, False)
|
||||
time.sleep(1)
|
||||
elif mode in ("a23", "a23_wifi"):
|
||||
leds = init_selected_leds(False, True)
|
||||
time.sleep(1)
|
||||
elif mode in ("gpio", "gpio_ina", "all"):
|
||||
leds = init_leds()
|
||||
time.sleep(1)
|
||||
if mode in ("ina", "gpio_ina", "all"):
|
||||
test_ina226()
|
||||
time.sleep(1)
|
||||
if mode in ("wifi", "a25_wifi", "a23_wifi", "all"):
|
||||
test_wifi()
|
||||
print("TEST COMPLETE")
|
||||
return 0
|
||||
except Exception as exc:
|
||||
print("TEST FAILED:", repr(exc))
|
||||
return 1
|
||||
finally:
|
||||
for name, led in leds:
|
||||
try:
|
||||
led.value(0)
|
||||
print(name, "LOW")
|
||||
except Exception as exc:
|
||||
print(name, "cleanup failed:", exc)
|
||||
|
||||
|
||||
main()
|
||||
@@ -0,0 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Standalone WiFi/GPIO isolation test.
|
||||
|
||||
This script intentionally does not import any project module. It only tests
|
||||
MaixPy WiFi startup with optional A23/A26 GPIO initialization.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from maix import gpio, network, pinmap
|
||||
|
||||
|
||||
GREEN_PIN = "A26"
|
||||
GREEN_GPIO = "GPIOA26"
|
||||
RED_PIN = "A23"
|
||||
RED_GPIO = "GPIOA23"
|
||||
|
||||
# Run this file directly from the official MaixPy tool.
|
||||
# Change only TEST_MODE between runs: none -> a26 -> a23 -> both.
|
||||
TEST_MODE = "none"
|
||||
WIFI_SSID = "sheling4b02-5G"
|
||||
WIFI_PASSWORD = "Aa12345678"
|
||||
WIFI_TIMEOUT_S = 20
|
||||
|
||||
|
||||
def init_gpio(mode):
|
||||
outputs = []
|
||||
if mode in ("a26", "both"):
|
||||
pinmap.set_pin_function(GREEN_PIN, GREEN_GPIO)
|
||||
green = gpio.GPIO(GREEN_GPIO, gpio.Mode.OUT)
|
||||
green.value(1)
|
||||
outputs.append((GREEN_GPIO, green))
|
||||
print("GPIOA26 initialized HIGH")
|
||||
if mode in ("a23", "both"):
|
||||
pinmap.set_pin_function(RED_PIN, RED_GPIO)
|
||||
red = gpio.GPIO(RED_GPIO, gpio.Mode.OUT)
|
||||
red.value(1)
|
||||
outputs.append((RED_GPIO, red))
|
||||
print("GPIOA23 initialized HIGH")
|
||||
return outputs
|
||||
|
||||
|
||||
def connect_wifi(ssid, password, timeout_s):
|
||||
print("Starting MaixPy WiFi connection...")
|
||||
wifi = network.wifi.Wifi()
|
||||
result = wifi.connect(ssid, password, wait=True, timeout=timeout_s)
|
||||
print("WiFi connect result:", result)
|
||||
try:
|
||||
print("WiFi connected:", wifi.is_connected())
|
||||
print("WiFi IP:", wifi.get_ip())
|
||||
except Exception as exc:
|
||||
print("WiFi status query failed:", exc)
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
mode = TEST_MODE.lower()
|
||||
if mode not in ("none", "a26", "a23", "both"):
|
||||
print("TEST_MODE must be none, a26, a23, or both")
|
||||
return 1
|
||||
ssid = WIFI_SSID
|
||||
password = WIFI_PASSWORD
|
||||
timeout_s = WIFI_TIMEOUT_S
|
||||
|
||||
print("=== Standalone WiFi/GPIO isolation ===")
|
||||
print("mode:", mode)
|
||||
print("ssid:", ssid)
|
||||
outputs = []
|
||||
try:
|
||||
outputs = init_gpio(mode)
|
||||
time.sleep(1)
|
||||
connect_wifi(ssid, password, timeout_s)
|
||||
return 0
|
||||
except Exception as exc:
|
||||
print("TEST FAILED:", repr(exc))
|
||||
return 1
|
||||
finally:
|
||||
for gpio_name, output in outputs:
|
||||
try:
|
||||
output.value(0)
|
||||
print(gpio_name, "LOW")
|
||||
except Exception as exc:
|
||||
print(gpio_name, "cleanup failed:", exc)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
+1
-1
@@ -4,6 +4,6 @@
|
||||
应用版本号
|
||||
每次 OTA 更新时,只需要更新这个文件中的版本号
|
||||
"""
|
||||
VERSION = '2.17.0'
|
||||
VERSION = '2.18.0'
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user