203 lines
7.3 KiB
Python
203 lines
7.3 KiB
Python
import os
|
|
import shutil
|
|
import hashlib
|
|
import json
|
|
from typing import Iterable, List, Optional
|
|
import datetime
|
|
|
|
def copy_folder_to_destination(src_path: str, dest_path: str, skip_dirs: Optional[Iterable[str]] = None, skip_files: Optional[Iterable[str]] = None) -> None:
|
|
# Ensure the destination directory exists
|
|
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
|
|
|
|
# Remove existing folder if present
|
|
if os.path.exists(dest_path):
|
|
shutil.rmtree(dest_path)
|
|
|
|
# Create destination directory
|
|
os.makedirs(dest_path)
|
|
|
|
# Walk through source directory
|
|
for root, dirs, files in os.walk(src_path):
|
|
# Remove directories to skip from dirs list
|
|
if skip_dirs:
|
|
dirs[:] = [d for d in dirs if d not in skip_dirs]
|
|
|
|
# Calculate relative path
|
|
rel_path = os.path.relpath(root, src_path)
|
|
dest_dir = os.path.join(dest_path, rel_path)
|
|
|
|
# Create corresponding destination directory
|
|
os.makedirs(dest_dir, exist_ok=True)
|
|
|
|
# Copy files that aren't in skip_files
|
|
for file in files:
|
|
if skip_files and file in skip_files:
|
|
continue
|
|
src_file = os.path.join(root, file)
|
|
dest_file = os.path.join(dest_dir, file)
|
|
shutil.copy2(src_file, dest_file)
|
|
|
|
def calculate_md5(file_path: str) -> str:
|
|
hash_md5 = hashlib.md5()
|
|
with open(file_path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(4096), b""):
|
|
hash_md5.update(chunk)
|
|
return hash_md5.hexdigest()
|
|
|
|
def get_file_size(file_path: str) -> int:
|
|
return os.path.getsize(file_path)
|
|
|
|
def update_json_file(json_array: List[dict], folder_path: str) -> None:
|
|
# Create new data array for files
|
|
file_array = []
|
|
|
|
# Walk through the copied folder and collect file details
|
|
for root, _, files in os.walk(folder_path):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
relative_path = os.path.relpath(file_path, folder_path)
|
|
|
|
# Replace backslashes with forward slashes
|
|
relative_path = relative_path.replace('\\', '/')
|
|
|
|
# Build remote/local using forward slashes only
|
|
file_entry = {
|
|
"remote": f"data/{relative_path}",
|
|
"local": f"/{relative_path}",
|
|
"md5": calculate_md5(file_path),
|
|
"size": get_file_size(file_path)
|
|
}
|
|
file_array.append(file_entry)
|
|
|
|
# Replace the contents of the input json_array with new data
|
|
json_array.clear()
|
|
json_array.extend(file_array)
|
|
|
|
|
|
def perform_data_copy(src_path: str, dest_path: str, skip_dirs: Optional[Iterable[str]] = None, skip_files: Optional[Iterable[str]] = None) -> bool:
|
|
# Check if the source folder exists
|
|
if not os.path.isdir(src_path):
|
|
print(f"Source folder does not exist: {src_path}")
|
|
return False
|
|
copy_folder_to_destination(src_path, dest_path, skip_dirs, skip_files)
|
|
print("Data folder copied successfully.")
|
|
return True
|
|
|
|
|
|
def main():
|
|
|
|
here_path = os.path.dirname(os.path.abspath(__file__))
|
|
project_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Path of the folder to copy (you can modify this)
|
|
src_folder_name = "data"
|
|
src_path = os.path.join(project_path, src_folder_name)
|
|
#print(f"source path: {src_path}")
|
|
|
|
# Path of the destination folder
|
|
dest_folder_name = "firmware_update\\latest\\data"
|
|
dest_path = os.path.join(project_path, dest_folder_name)
|
|
#print(f"destination path: {dest_path}")
|
|
|
|
# Path of the firmware binary file
|
|
bin_name = ".pio\\build\\esp32s3dev\\firmware.bin"
|
|
|
|
# Skip these directories
|
|
skip_dirs = ["boards", "booths"]
|
|
|
|
# Skip these files
|
|
skip_files = ["wifi.json", "system.json", "luma-stiks.json", ]
|
|
|
|
|
|
# Allow non-interactive mode via environment variable (set NON_INTERACTIVE=y to auto 'y')
|
|
non_interactive = os.environ.get("NON_INTERACTIVE", "n").lower() in ("1", "y", "yes", "true")
|
|
|
|
file_choice = os.environ.get("UPDATE_FILES") if non_interactive else input("Do you want to update the files? (y/n): ")
|
|
if file_choice is None:
|
|
file_choice = "n"
|
|
file_choice = file_choice.strip().lower()
|
|
|
|
|
|
# *********************** Copy Data Files ***********************
|
|
copied = False
|
|
if file_choice == "y":
|
|
copied = perform_data_copy(src_path, dest_path, skip_dirs, skip_files)
|
|
|
|
|
|
# *********************** Copy Binary file ***********************
|
|
fw_choice = os.environ.get("UPDATE_FIRMWARE") if non_interactive else input("Do you want to update the firmware? (y/n): ")
|
|
if fw_choice is None:
|
|
fw_choice = "n"
|
|
fw_choice = fw_choice.strip().lower()
|
|
if fw_choice == "y":
|
|
# Copy firmware.bin to the destination
|
|
bin_path = os.path.join(project_path, bin_name)
|
|
if not os.path.isfile(bin_path):
|
|
print(f"Firmware binary not found: {bin_path}")
|
|
else:
|
|
shutil.copy(bin_path, here_path)
|
|
print("firmware.bin copied successfully.")
|
|
|
|
|
|
|
|
# *********************** Process update.json ***********************
|
|
# Update the JSON file
|
|
#json_path = os.path.join(here_path, "update.json")
|
|
json_path = os.path.join(here_path, "latest", "update.json")
|
|
print(f"json path: {json_path}")
|
|
|
|
# Read existing JSON
|
|
if not os.path.isfile(json_path):
|
|
print(f"update.json not found at {json_path}; creating a new one.")
|
|
json_doc = {"files": [], "firmware": {"md5": "", "size": 0}}
|
|
else:
|
|
with open(json_path, "r") as f:
|
|
try:
|
|
json_doc = json.load(f)
|
|
except json.JSONDecodeError:
|
|
print("Invalid JSON file! Aborting.")
|
|
return
|
|
|
|
# Ensure required keys exist
|
|
json_doc.setdefault("files", [])
|
|
json_doc.setdefault("firmware", {"md5": "", "size": 0})
|
|
|
|
|
|
# process the files array
|
|
#if update_files.lower() == "y":
|
|
if file_choice == "y":
|
|
if not copied:
|
|
print("Skipping file list update because data copy failed.")
|
|
else:
|
|
json_files_array = json_doc["files"]
|
|
update_json_file(json_files_array, dest_path)
|
|
#print(f"Folder {os.path.basename(src_path)} processed successfully.")
|
|
|
|
|
|
# *********************** Process firmwware.bin in update.json ***********************
|
|
# process the firmware
|
|
if fw_choice == "y":
|
|
json_firmware = json_doc["firmware"]
|
|
firmware_path = os.path.join(here_path, "firmware.bin")
|
|
if os.path.isfile(firmware_path):
|
|
json_firmware["md5"] = calculate_md5(firmware_path)
|
|
json_firmware["size"] = get_file_size(firmware_path)
|
|
else:
|
|
print("Firmware file missing; firmware section not updated.")
|
|
|
|
|
|
# *********************** Update release date, time in update.json ***********************
|
|
now = datetime.datetime.now()
|
|
json_doc["release_date"] = now.strftime("%Y-%m-%d")
|
|
json_doc["release_time"] = now.strftime("%H:%M:%S")
|
|
|
|
|
|
# Write updated JSON
|
|
with open(json_path, "w") as f:
|
|
json.dump(json_doc, f, indent=4)
|
|
|
|
print("update.json updated successfully.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|