317 lines
9.6 KiB
Python
Executable File
317 lines
9.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import sys
|
|
import shutil
|
|
import subprocess
|
|
import glob
|
|
import importlib.util
|
|
import argparse
|
|
import inspect
|
|
|
|
REPO_URL = "https://github.com/crosspoint-reader/crosspoint-reader.git"
|
|
REPO_DIR = "crosspoint-reader"
|
|
PLUGINS_DIR = "plugins"
|
|
|
|
ENVIRONMENTS = {
|
|
"default": "Debug logging enabled, version from current git branch (recommended)",
|
|
"gh_release": "Info logging only, version hardcoded to release tag",
|
|
"slim": "No serial logging, smallest binary size",
|
|
}
|
|
|
|
|
|
def run(cmd, cwd=None, check=True):
|
|
print(f" $ {cmd}")
|
|
return subprocess.run(cmd, shell=True, cwd=cwd, check=check)
|
|
|
|
|
|
def remove_readonly(func, path, _):
|
|
os.chmod(path, 0o777)
|
|
func(path)
|
|
|
|
|
|
def rmtree(path):
|
|
if sys.version_info >= (3, 12):
|
|
shutil.rmtree(path, onexc=remove_readonly)
|
|
else:
|
|
shutil.rmtree(path, onerror=remove_readonly)
|
|
|
|
|
|
def get_remote_head_sha():
|
|
try:
|
|
result = subprocess.run(
|
|
f"git ls-remote {REPO_URL} HEAD",
|
|
shell=True, capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0 and result.stdout:
|
|
return result.stdout.split()[0]
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_local_head_sha():
|
|
try:
|
|
result = subprocess.run(
|
|
"git rev-parse HEAD",
|
|
shell=True, capture_output=True, text=True, cwd=REPO_DIR
|
|
)
|
|
if result.returncode == 0:
|
|
return result.stdout.strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def clone_repo(force: bool = True):
|
|
print("[1/3] Preparing CrossPoint repository...")
|
|
|
|
if not force and os.path.exists(REPO_DIR):
|
|
print(" Existing repo found, checking remote for updates...")
|
|
remote_sha = get_remote_head_sha()
|
|
local_sha = get_local_head_sha()
|
|
|
|
if remote_sha and local_sha and remote_sha == local_sha:
|
|
print(f" Already up to date ({local_sha[:8]}), skipping clone.")
|
|
return
|
|
elif remote_sha:
|
|
print(f" Remote has changed ({remote_sha[:8]} vs local {(local_sha or 'unknown')[:8]}), re-cloning...")
|
|
else:
|
|
print(" Could not reach remote, re-using existing repo.")
|
|
return
|
|
|
|
if os.path.exists(REPO_DIR):
|
|
print(f" Removing existing '{REPO_DIR}'...")
|
|
rmtree(REPO_DIR)
|
|
|
|
run(f"git clone --recursive {REPO_URL} {REPO_DIR}")
|
|
|
|
|
|
def run_plugin_patch(plugin_dir: str, yes_all: bool = False) -> bool:
|
|
patch_path = os.path.join(plugin_dir, "patch.py")
|
|
if not os.path.exists(patch_path):
|
|
print(f" WARNING: No patch.py found in {plugin_dir}, skipping.")
|
|
return False
|
|
|
|
spec = importlib.util.spec_from_file_location("patch", patch_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
if hasattr(module, "patch"):
|
|
sig = inspect.signature(module.patch)
|
|
if "yes_all" in sig.parameters:
|
|
module.patch(os.path.abspath(REPO_DIR), yes_all=yes_all)
|
|
else:
|
|
module.patch(os.path.abspath(REPO_DIR))
|
|
return True
|
|
except Exception as exc:
|
|
print(f" ERROR while running patch.py for {plugin_dir}: {exc}")
|
|
return False
|
|
|
|
|
|
def apply_plugins(yes_all: bool = False):
|
|
print("[2/3] Applying plugins...")
|
|
|
|
if not os.path.exists(PLUGINS_DIR):
|
|
print(f" No '{PLUGINS_DIR}' directory found — skipping plugin step.")
|
|
return
|
|
|
|
plugin_dirs = sorted(
|
|
d for d in os.listdir(PLUGINS_DIR)
|
|
if os.path.isdir(os.path.join(PLUGINS_DIR, d))
|
|
)
|
|
|
|
if not plugin_dirs:
|
|
print(" No plugin directories found.")
|
|
return
|
|
|
|
for plugin_name in plugin_dirs:
|
|
plugin_path = os.path.join(PLUGINS_DIR, plugin_name)
|
|
|
|
print(f"\n Plugin: {plugin_name}")
|
|
if yes_all:
|
|
answer = "y"
|
|
print(f" Install '{plugin_name}'? [Y/n]: Y (--yes)")
|
|
else:
|
|
answer = input(f" Install '{plugin_name}'? [Y/n]: ").strip().lower()
|
|
|
|
if answer in ("", "y", "yes"):
|
|
print(f" Installing {plugin_name}...")
|
|
success = run_plugin_patch(plugin_path, yes_all=yes_all)
|
|
if success:
|
|
print(f" ✓ {plugin_name} installed.")
|
|
else:
|
|
print(f" ✗ {plugin_name} failed or was skipped.")
|
|
else:
|
|
print(f" Skipping {plugin_name}.")
|
|
|
|
|
|
def list_likely_serial_ports():
|
|
patterns = [
|
|
"/dev/cu.usb*",
|
|
"/dev/tty.usb*",
|
|
"/dev/cu.wchusb*",
|
|
"/dev/tty.wchusb*",
|
|
"/dev/cu.SLAB*",
|
|
"/dev/tty.SLAB*",
|
|
"/dev/ttyACM*",
|
|
"/dev/ttyUSB*",
|
|
]
|
|
ports = []
|
|
for pattern in patterns:
|
|
ports.extend(glob.glob(pattern))
|
|
return sorted(set(ports))
|
|
|
|
|
|
def detect_serial_ports():
|
|
port_map: dict[str, str] = {}
|
|
|
|
try:
|
|
import serial.tools.list_ports
|
|
|
|
usb_keywords = {
|
|
"usb", "cp210", "ftdi", "ch340", "ch9102", "esp",
|
|
"serial", "uart", "prolific", "silabs",
|
|
}
|
|
skip_keywords = {"bluetooth", "bt-", "btle", "airpods", "handsfree"}
|
|
|
|
for port in serial.tools.list_ports.comports():
|
|
combined = (port.device + " " + (port.description or "") +
|
|
" " + (port.hwid or "")).lower()
|
|
|
|
if any(s in combined for s in skip_keywords):
|
|
continue
|
|
if any(k in combined for k in usb_keywords):
|
|
port_map[port.device] = port.description or "USB Serial Device"
|
|
|
|
except ImportError:
|
|
pass
|
|
|
|
for device in list_likely_serial_ports():
|
|
if device not in port_map:
|
|
port_map[device] = "Detected via glob pattern"
|
|
|
|
return sorted(port_map.items())
|
|
|
|
|
|
def prompt_for_upload_port():
|
|
print("\n[3/3] Select serial port for flashing...")
|
|
|
|
max_retries = 3
|
|
|
|
for attempt in range(1, max_retries + 1):
|
|
print(f"\n Attempt {attempt}/{max_retries}")
|
|
print(" Ensure your xteink device is connected via USB and awake.")
|
|
input(" Press Enter when ready... ")
|
|
|
|
ports = detect_serial_ports()
|
|
|
|
if ports:
|
|
cu_ports = [(d, desc) for d, desc in ports if d.startswith("/dev/cu.")]
|
|
recommended = cu_ports[0][0] if cu_ports else ports[0][0]
|
|
|
|
print(f"\n Found {len(ports)} serial port(s):")
|
|
for i, (device, desc) in enumerate(ports, 1):
|
|
tag = " ← recommended" if device == recommended else ""
|
|
print(f" {i}. {device} — {desc}{tag}")
|
|
|
|
choice = input(f"\n Press ENTER to use {recommended}, or enter 1-{len(ports)}: ").strip()
|
|
|
|
if not choice:
|
|
print(f" Selected: {recommended}")
|
|
return recommended
|
|
|
|
if choice.isdigit():
|
|
idx = int(choice) - 1
|
|
if 0 <= idx < len(ports):
|
|
print(f" Selected: {ports[idx][0]}")
|
|
return ports[idx][0]
|
|
|
|
print(" Invalid choice, please try again.")
|
|
|
|
else:
|
|
print(" No USB serial ports detected.")
|
|
if attempt < max_retries:
|
|
print(" Make sure the device is connected, powered on, and awake.")
|
|
|
|
print("\n Could not auto-detect a port.")
|
|
print(" Common names: macOS: /dev/cu.usb* Linux: /dev/ttyUSB* Windows: COMx")
|
|
|
|
while True:
|
|
port = input("\n Enter port manually (or 'quit' to exit): ").strip()
|
|
if port.lower() == "quit":
|
|
sys.exit(1)
|
|
if port:
|
|
return port
|
|
print(" Please enter a valid port name.")
|
|
|
|
|
|
def build_and_flash(environment: str):
|
|
print(f"\nBuilding firmware with PlatformIO (environment: {environment})...")
|
|
run(f"pio run --environment {environment}", cwd=REPO_DIR)
|
|
|
|
port = prompt_for_upload_port()
|
|
|
|
print(f"\n Flashing to {port}...")
|
|
run(
|
|
f"pio run --target upload --environment {environment} --upload-port {port}",
|
|
cwd=REPO_DIR,
|
|
)
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(
|
|
description="xteink Plugin Installer",
|
|
formatter_class=argparse.RawTextHelpFormatter,
|
|
)
|
|
env_help = "\n".join(f" {name}: {desc}" for name, desc in ENVIRONMENTS.items())
|
|
parser.add_argument(
|
|
"-e", "--environment",
|
|
default="default",
|
|
choices=ENVIRONMENTS.keys(),
|
|
metavar="ENV",
|
|
help=f"PlatformIO build environment (default: default)\n{env_help}",
|
|
)
|
|
parser.add_argument(
|
|
"-y", "--yes",
|
|
action="store_true",
|
|
help="Auto-answer Y to all plugin install prompts",
|
|
)
|
|
parser.add_argument(
|
|
"--no-reclone",
|
|
action="store_true",
|
|
dest="no_reclone",
|
|
help="Skip re-cloning if CrossPoint repo already exists and matches remote",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
|
|
print("=" * 60)
|
|
print(" xteink Plugin Installer")
|
|
print("=" * 60)
|
|
print()
|
|
print(" This script modifies and flashes custom firmware to your")
|
|
print(" device. The author of this script (Justin Oros) accepts")
|
|
print(" no responsibility for any damage that may occur to your")
|
|
print(" device as a result of using this installer.")
|
|
print()
|
|
print(f" Environment : {args.environment}")
|
|
print(f" {ENVIRONMENTS[args.environment]}")
|
|
print()
|
|
answer = input(" Do you wish to proceed? [Y/n]: ").strip().lower() if not args.yes else "y"
|
|
if answer not in ("", "y", "yes"):
|
|
print(" Aborted.")
|
|
sys.exit(0)
|
|
print()
|
|
clone_repo(force=not args.no_reclone)
|
|
apply_plugins(yes_all=args.yes)
|
|
build_and_flash(args.environment)
|
|
print("\nDone.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|