#! /usr/bin/env python3
|
from dotenv import load_dotenv
|
load_dotenv()
|
|
import os
|
import struct
|
import zipfile
|
import oss2
|
import json
|
import requests
|
from requests.exceptions import RequestException
|
|
# 切换到项目根目录
|
os.chdir(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
def get_chip_id_string(chip_id):
|
return {
|
0x0000: "esp32",
|
0x0002: "esp32s2",
|
0x0005: "esp32c3",
|
0x0009: "esp32s3",
|
0x000C: "esp32c2",
|
0x000D: "esp32c6",
|
0x0010: "esp32h2",
|
0x0011: "esp32c5",
|
0x0012: "esp32p4",
|
0x0017: "esp32c5",
|
}[chip_id]
|
|
def get_flash_size(flash_size):
|
MB = 1024 * 1024
|
return {
|
0x00: 1 * MB,
|
0x01: 2 * MB,
|
0x02: 4 * MB,
|
0x03: 8 * MB,
|
0x04: 16 * MB,
|
0x05: 32 * MB,
|
0x06: 64 * MB,
|
0x07: 128 * MB,
|
}[flash_size]
|
|
def get_app_desc(data):
|
magic = struct.unpack("<I", data[0x00:0x04])[0]
|
if magic != 0xabcd5432:
|
raise Exception("Invalid app desc magic")
|
version = data[0x10:0x30].decode("utf-8").strip('\0')
|
project_name = data[0x30:0x50].decode("utf-8").strip('\0')
|
time = data[0x50:0x60].decode("utf-8").strip('\0')
|
date = data[0x60:0x70].decode("utf-8").strip('\0')
|
idf_ver = data[0x70:0x90].decode("utf-8").strip('\0')
|
elf_sha256 = data[0x90:0xb0].hex()
|
return {
|
"name": project_name,
|
"version": version,
|
"compile_time": date + "T" + time,
|
"idf_version": idf_ver,
|
"elf_sha256": elf_sha256,
|
}
|
|
def get_board_name(folder):
|
basename = os.path.basename(folder)
|
if basename.startswith("v0.2"):
|
return "bread-simple"
|
if basename.startswith("v0.3") or basename.startswith("v0.4") or basename.startswith("v0.5") or basename.startswith("v0.6"):
|
if "ML307" in basename:
|
return "bread-compact-ml307"
|
elif "WiFi" in basename:
|
return "bread-compact-wifi"
|
elif "KevinBox1" in basename:
|
return "kevin-box-1"
|
if basename.startswith("v0.7") or basename.startswith("v0.8") or basename.startswith("v0.9") or basename.startswith("v1.") or basename.startswith("v2."):
|
return basename.split("_")[1]
|
raise Exception(f"Unknown board name: {basename}")
|
|
def find_app_partition(data):
|
partition_begin = 0x8000
|
partition_end = partition_begin + 0x4000
|
# find the first parition with type 0x00
|
for i in range(partition_begin, partition_end, 0x20):
|
# magic is aa 50
|
if data[i] == 0xaa and data[i + 1] == 0x50:
|
# type is app
|
if data[i + 2] == 0x00:
|
# read offset and size
|
offset = struct.unpack("<I", data[i + 4:i + 8])[0]
|
size = struct.unpack("<I", data[i + 8:i + 12])[0]
|
# then 16 bytes is label
|
label = data[i + 12:i + 28].decode("utf-8").strip('\0')
|
print(f"found app partition at 0x{i:08x}, offset: 0x{offset:08x}, size: 0x{size:08x}, label: {label}")
|
return {
|
"offset": offset,
|
"size": size,
|
"label": label,
|
}
|
return None
|
|
def read_binary(dir_path):
|
merged_bin_path = os.path.join(dir_path, "merged-binary.bin")
|
merged_bin_data = open(merged_bin_path, "rb").read()
|
|
# find app partition
|
app_partition = find_app_partition(merged_bin_data)
|
if app_partition is None:
|
print("no app partition found")
|
return
|
app_data = merged_bin_data[app_partition["offset"]:app_partition["offset"] + app_partition["size"]]
|
# check magic
|
if app_data[0] != 0xE9:
|
print("not a valid image")
|
return
|
# get flash size
|
flash_size = get_flash_size(app_data[0x3] >> 4)
|
chip_id = get_chip_id_string(app_data[0xC])
|
# get segments
|
segment_count = app_data[0x1]
|
segments = []
|
offset = 0x18
|
image_size = 0x18
|
for i in range(segment_count):
|
segment_size = struct.unpack("<I", app_data[offset + 4:offset + 8])[0]
|
image_size += 8 + segment_size
|
offset += 8
|
segment_data = app_data[offset:offset + segment_size]
|
offset += segment_size
|
segments.append(segment_data)
|
assert offset < len(app_data), "offset is out of bounds"
|
|
# add checksum size
|
image_size += 1
|
image_size = (image_size + 15) & ~15
|
# hash appended
|
if app_data[0x17] == 1:
|
image_size += 32
|
print(f"image size: {image_size}")
|
|
# verify the remaining data are all 0xFF
|
for i in range(image_size, len(app_data)):
|
if app_data[i] != 0xFF:
|
print(f"Failed to verify image, data at 0x{i:08x} is not 0xFF")
|
return
|
|
image_data = app_data[:image_size]
|
|
# extract bin file
|
bin_path = os.path.join(dir_path, "xiaozhi.bin")
|
if not os.path.exists(bin_path):
|
print("extract bin file to", bin_path)
|
open(bin_path, "wb").write(image_data)
|
|
# The app desc is in the first segment
|
desc = get_app_desc(segments[0])
|
return {
|
"chip_id": chip_id,
|
"flash_size": flash_size,
|
"board": get_board_name(dir_path),
|
"application": desc,
|
"firmware_size": image_size,
|
}
|
|
def extract_zip(zip_path, extract_path):
|
if not os.path.exists(extract_path):
|
os.makedirs(extract_path)
|
print(f"Extracting {zip_path} to {extract_path}")
|
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
|
zip_ref.extractall(extract_path)
|
|
def upload_dir_to_oss(source_dir, target_dir):
|
auth = oss2.Auth(os.environ['OSS_ACCESS_KEY_ID'], os.environ['OSS_ACCESS_KEY_SECRET'])
|
bucket = oss2.Bucket(auth, os.environ['OSS_ENDPOINT'], os.environ['OSS_BUCKET_NAME'])
|
for filename in os.listdir(source_dir):
|
oss_key = os.path.join(target_dir, filename)
|
# check if is file
|
if not os.path.isfile(os.path.join(source_dir, filename)):
|
continue
|
print('uploading', oss_key)
|
bucket.put_object(oss_key, open(os.path.join(source_dir, filename), 'rb'))
|
|
def post_info_to_server(info):
|
"""
|
将固件信息发送到服务器
|
|
Args:
|
info: 包含固件信息的字典
|
"""
|
try:
|
# 从环境变量获取服务器URL和token
|
server_url = os.environ.get('VERSIONS_SERVER_URL')
|
server_token = os.environ.get('VERSIONS_TOKEN')
|
|
if not server_url or not server_token:
|
raise Exception("Missing SERVER_URL or TOKEN in environment variables")
|
|
# 准备请求头和数据
|
headers = {
|
'Authorization': f'Bearer {server_token}',
|
'Content-Type': 'application/json'
|
}
|
|
# 发送POST请求
|
response = requests.post(
|
server_url,
|
headers=headers,
|
json={'jsonData': json.dumps(info)}
|
)
|
|
# 检查响应状态
|
response.raise_for_status()
|
|
print(f"Successfully uploaded version info for tag: {info['tag']}")
|
|
except RequestException as e:
|
if hasattr(e.response, 'json'):
|
error_msg = e.response.json().get('error', str(e))
|
else:
|
error_msg = str(e)
|
print(f"Failed to upload version info: {error_msg}")
|
raise
|
except Exception as e:
|
print(f"Error uploading version info: {str(e)}")
|
raise
|
|
def main():
|
release_dir = "releases"
|
# look for zip files startswith "v"
|
for name in os.listdir(release_dir):
|
if name.startswith("v") and name.endswith(".zip"):
|
tag = name[:-4]
|
folder = os.path.join(release_dir, tag)
|
info_path = os.path.join(folder, "info.json")
|
if not os.path.exists(info_path):
|
if not os.path.exists(folder):
|
os.makedirs(folder)
|
extract_zip(os.path.join(release_dir, name), folder)
|
info = read_binary(folder)
|
target_dir = os.path.join("firmwares", tag)
|
info["tag"] = tag
|
info["url"] = os.path.join(os.environ['OSS_BUCKET_URL'], target_dir, "xiaozhi.bin")
|
open(info_path, "w", encoding="utf-8").write(json.dumps(info, indent=4))
|
# upload all file to oss
|
upload_dir_to_oss(folder, target_dir)
|
# read info.json
|
info = json.load(open(info_path, encoding="utf-8"))
|
# post info.json to server
|
post_info_to_server(info)
|
|
|
|
if __name__ == "__main__":
|
main()
|
