mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-11-08 13:33:42 +08:00
firestar5683
8cb60d82fe
Provide laptop and device build entry points, cache compatibility checks and developer tooling. Keep distribution assets and model packaging available before runtime consumers.
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
"""Read-only local manager admission check for an existing Params namespace."""
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from openpilot.common.params import Params
|
|
from openpilot.starpilot.state_migration import MigrationRequired, prepare_manager_start
|
|
|
|
|
|
class ExistingParams:
|
|
def __init__(self, namespace: Path, keys):
|
|
self.namespace = namespace
|
|
self.keys = keys
|
|
|
|
def get_param_path(self):
|
|
return str(self.namespace)
|
|
|
|
def all_keys(self):
|
|
return self.keys
|
|
|
|
|
|
def check_startup(params_root: Path, storage: Path) -> None:
|
|
if not params_root.is_dir():
|
|
raise ValueError("Existing Params namespace required")
|
|
standard = params_root / "d"
|
|
if standard.is_symlink() and standard.is_dir():
|
|
namespace = standard
|
|
elif params_root.is_symlink() or (not any(path.is_symlink() for path in params_root.iterdir()) and
|
|
any(path.is_file() and path.name != ".lock" for path in params_root.iterdir())):
|
|
namespace = params_root
|
|
else:
|
|
raise ValueError("Supply an existing Params namespace or a root with d; empty or named-link roots are ambiguous")
|
|
if not storage.is_dir():
|
|
raise ValueError("Existing recovery storage directory required")
|
|
with TemporaryDirectory(prefix="startup-preflight-") as temporary:
|
|
keys = Params(temporary).all_keys()
|
|
prepare_manager_start(ExistingParams(namespace, keys), storage, dry_run=True)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("params_root", type=Path)
|
|
parser.add_argument("storage", type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
check_startup(args.params_root, args.storage)
|
|
except (MigrationRequired, ValueError, OSError) as error:
|
|
print(f"Startup admission rejected: {error}")
|
|
return 1
|
|
print("Startup admission accepted (read-only point-in-time check)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|