#!/usr/bin/env python3 """ Soleprint Build Tool Generates soleprint instances from source + room config. Usage: python build.py # Build gen/standalone/ python build.py --cfg # Build gen// python build.py --all # Build all rooms python build.py --output /path/ # Build to custom path python build.py --models # Only regenerate models Generated structure for standalone rooms: gen/standalone/ artery/, atlas/, station/, main.py, ... Generated structure for managed rooms: gen// / # Copied repos + ctrl link/ # Bridge code soleprint/ # Soleprint instance """ import argparse import importlib.util import json import logging import shutil import stat import subprocess import sys from pathlib import Path SPR_ROOT = Path(__file__).resolve().parent logging.basicConfig(level=logging.INFO, format="%(message)s") log = logging.getLogger(__name__) def load_config(cfg_name: str | None) -> dict: """Load room config.json.""" room = cfg_name or "standalone" config_path = SPR_ROOT / "cfg" / room / "config.json" if config_path.exists(): return json.loads(config_path.read_text()) return {} def ensure_dir(path: Path): path.mkdir(parents=True, exist_ok=True) def _rmtree_resilient(path: Path): """Remove path tree, tolerating root-owned files written by containers. Docker containers that mount gen/ as a volume sometimes write files as root (e.g. __pycache__). A plain shutil.rmtree then fails with EACCES. We first try shutil.rmtree; if that hits a PermissionError we fall back to deleting the offending files from inside an ephemeral alpine container. """ def _chmod_and_retry(func, target, exc_info): try: Path(target).chmod(stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR) func(target) except Exception: raise try: shutil.rmtree(path, onerror=_chmod_and_retry) return except PermissionError: pass log.info(" (falling back to docker-based cleanup)") subprocess.run( ["docker", "run", "--rm", "-v", f"{path.parent}:/work", "alpine:3", "sh", "-c", f"rm -rf /work/{path.name}"], check=True, ) def copy_path(source: Path, target: Path, quiet: bool = False): """Copy file or directory, resolving symlinks.""" if target.is_symlink(): target.unlink() elif target.exists(): if target.is_dir(): shutil.rmtree(target) else: target.unlink() if source.is_dir(): shutil.copytree(source, target, symlinks=False) if not quiet: log.info(f" {target.name}/") else: shutil.copy2(source, target) if not quiet: log.info(f" {target.name}") def count_files(path: Path) -> int: return sum(1 for _ in path.rglob("*") if _.is_file()) def merge_into(source: Path, target: Path): """Merge source directory into target (additive, overwrites conflicts).""" if not source.exists(): return for item in source.rglob("*"): if item.is_file(): rel = item.relative_to(source) dest = target / rel dest.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(item, dest) def parse_gitignore(gitignore_path: Path) -> set[str]: """Parse .gitignore and return set of patterns to ignore.""" patterns = set() if not gitignore_path.exists(): return patterns for line in gitignore_path.read_text().splitlines(): line = line.strip() # Skip comments and empty lines if not line or line.startswith("#"): continue # Remove trailing slashes (directory indicators) pattern = line.rstrip("/") # Skip negation patterns (we don't support them) if pattern.startswith("!"): continue patterns.add(pattern) return patterns def copy_repo(source: Path, target: Path): """Copy a repo directory, excluding .git and gitignored files.""" if not source.exists(): log.warning(f"Repo not found: {source}") return False # Always ignore these always_ignore = {".git", "__pycache__", "node_modules", ".venv", "venv", ".env"} # Parse .gitignore from repo root gitignore_patterns = parse_gitignore(source / ".gitignore") def ignore_patterns(directory, files): ignored = set() rel_dir = Path(directory).relative_to(source) for f in files: # Always ignore these if f in always_ignore: ignored.add(f) continue # Check gitignore patterns rel_path = rel_dir / f if str(rel_dir) != "." else Path(f) for pattern in gitignore_patterns: # Simple pattern matching (filename or extension) if pattern.startswith("*."): # Extension pattern like *.pyc if f.endswith(pattern[1:]): ignored.add(f) break elif pattern == f or pattern == str(rel_path): # Exact match ignored.add(f) break elif "/" not in pattern and f == pattern: # Simple name match anywhere ignored.add(f) break return ignored shutil.copytree(source, target, ignore=ignore_patterns, symlinks=False) return True def build_managed(output_dir: Path, cfg_name: str, config: dict): """Build managed folder with repos + ctrl.""" managed = config.get("managed", {}) managed_name = managed.get("name", cfg_name) repos = managed.get("repos", {}) managed_dir = output_dir / managed_name ensure_dir(managed_dir) log.info(f"Building managed ({managed_name})...") # Copy repos (relative paths resolve from SPR_ROOT) for repo_name, repo_path in repos.items(): source = Path(repo_path) if not source.is_absolute(): source = SPR_ROOT / source target = managed_dir / repo_name if copy_repo(source, target): log.info(f" {repo_name}/") room_cfg = SPR_ROOT / "cfg" / cfg_name # Docker files from room root -> managed root for item in room_cfg.iterdir(): if item.is_file() and ( item.name.startswith("Dockerfile") or item.name.startswith("docker-compose") ): copy_path(item, managed_dir / item.name) # Copy managed app config from cfg/// (e.g., .env, dumps/) room_managed_cfg = room_cfg / managed_name if room_managed_cfg.exists(): log.info(f" Copying {managed_name} config...") for item in room_managed_cfg.iterdir(): if item.is_file(): copy_path(item, managed_dir / item.name, quiet=True) elif item.is_dir(): target = managed_dir / item.name if target.exists(): # Merge into existing repo directory merge_into(item, target) else: copy_path(item, target) # Scripts from ctrl/ -> output_dir/ctrl/ (sibling of managed, link, soleprint) room_ctrl = room_cfg / "ctrl" if room_ctrl.exists(): ctrl_dir = output_dir / "ctrl" ensure_dir(ctrl_dir) for item in room_ctrl.iterdir(): if item.is_file(): copy_path(item, ctrl_dir / item.name) def build_link(output_dir: Path, cfg_name: str): """Build link folder.""" room_cfg = SPR_ROOT / "cfg" / cfg_name link_source = room_cfg / "link" if not link_source.exists(): return log.info("Building link...") link_dir = output_dir / "link" copy_path(link_source, link_dir) def generate_models(output_dir: Path, room: str): """Generate models using modelgen tool.""" from soleprint.station.tools.modelgen import ModelGenerator, load_config config_path = SPR_ROOT / "cfg" / room / "config.json" if not config_path.exists(): log.warning(f"Config not found: {config_path}") return False models_file = output_dir / "models" / "pydantic" / "__init__.py" models_file.parent.mkdir(parents=True, exist_ok=True) try: config = load_config(config_path) generator = ModelGenerator( config=config, output_path=models_file, output_format="pydantic", ) generator.generate() return True except Exception as e: log.error(f"Model generation failed: {e}") return False def copy_cfg(output_dir: Path, room: str): """Copy room configuration files to output directory.""" room_cfg = SPR_ROOT / "cfg" / room if not room_cfg.exists(): log.warning(f"Room config not found: {room}") return log.info(f"Copying {room} config...") # config.json -> cfg/ cfg_dir = output_dir / "cfg" ensure_dir(cfg_dir) if (room_cfg / "config.json").exists(): copy_path(room_cfg / "config.json", cfg_dir / "config.json") # data/ -> data/ if (room_cfg / "data").exists(): copy_path(room_cfg / "data", output_dir / "data") # .env.example if (room_cfg / ".env.example").exists(): copy_path(room_cfg / ".env.example", output_dir / ".env.example") # Room-specific soleprint config (docker-compose.yml, etc) # Now in cfg//soleprint/ room_soleprint = room_cfg / "soleprint" if room_soleprint.exists(): systems = {"artery", "atlas", "station"} for item in room_soleprint.iterdir(): if item.name in systems: # Merge system dirs into already-copied framework code log.info(f" Merging {room} {item.name}...") merge_into(item, output_dir / item.name) elif item.is_file(): copy_path(item, output_dir / item.name) elif item.is_dir(): # Copy non-system dirs as-is (nginx/, etc.) copy_path(item, output_dir / item.name) def load_cabinets(room: str) -> list[dict]: """The dependency containers a room asked for, in the order it listed them. Read from cfg//data/cabinets.json — the same shape and the same place as its sibling data/*.json files, so nothing new has to know about it. Entries are {"name": "postgres"} and may carry an "env" override. """ path = SPR_ROOT / "cfg" / room / "data" / "cabinets.json" if not path.exists(): return [] try: entries = json.loads(path.read_text()) except ValueError as e: log.warning(f" cabinets.json is not valid JSON, ignoring: {e}") return [] out = [] for entry in entries if isinstance(entries, list) else []: if isinstance(entry, str): entry = {"name": entry} if isinstance(entry, dict) and entry.get("name"): out.append(entry) return out def resolve_cabinets(requested: list[dict]) -> list[dict]: """Expand each request into its definition, pulling in what it depends on. Airflow without postgres is a container that exits on boot, so a cabinet's depends_on is added for you rather than left as something to remember. """ cabinets_dir = SPR_ROOT / "soleprint" / "station" / "cabinets" resolved: dict[str, dict] = {} def add(name: str, overrides: dict) -> None: if name in resolved: # Already pulled in as somebody's dependency. The room asking for it # by name is the more specific statement, so its env still applies — # otherwise declaring airflow before postgres would silently drop # postgres's settings. if overrides.get("env"): resolved[name]["env"] = { **resolved[name].get("env", {}), **overrides["env"], } return definition_path = cabinets_dir / name / "cabinet.json" if not definition_path.exists(): available = sorted( p.name for p in cabinets_dir.iterdir() if p.is_dir() ) if cabinets_dir.exists() else [] log.warning(f" no such cabinet: {name} (available: {', '.join(available) or 'none'})") return try: definition = json.loads(definition_path.read_text()) except ValueError as e: log.warning(f" cabinet {name} has invalid cabinet.json: {e}") return # Mark it claimed before recursing, so a dependency cycle terminates. resolved[name] = definition for dependency in definition.get("depends_on", []) or []: add(dependency, {}) definition["env"] = {**definition.get("env", {}), **overrides.get("env", {})} for entry in requested: add(entry["name"], entry) # Dependencies first, so compose reads in the order things start. ordered: list[dict] = [] seen: set[str] = set() def emit(name: str) -> None: if name in seen or name not in resolved: return seen.add(name) for dependency in resolved[name].get("depends_on", []) or []: emit(dependency) ordered.append(resolved[name]) for name in resolved: emit(name) return ordered def compose_cabinets(output_dir: Path, room: str): """Merge the room's cabinets into its docker-compose.yml and .env.example. This is the compile step for dependencies: a room declares postgres, and the built instance comes out with postgres in its compose file rather than with instructions for adding it. """ requested = load_cabinets(room) if not requested: return try: import yaml except ImportError: log.warning( " cabinets need PyYAML to merge into docker-compose.yml " "(pip install pyyaml) — skipping" ) return cabinets = resolve_cabinets(requested) if not cabinets: return compose_path = output_dir / "docker-compose.yml" if not compose_path.exists(): log.warning( f" no docker-compose.yml in {output_dir.name}, " f"so there is nothing to merge {len(cabinets)} cabinet(s) into" ) return original = compose_path.read_text() # A YAML round-trip drops every comment, and the room's compose file leads # with the one that says how to run it. Keep the header block; the rest is # generated anyway. header = [] for line in original.splitlines(): if line.startswith("#") or not line.strip(): header.append(line) else: break while header and not header[-1].strip(): header.pop() compose = yaml.safe_load(original) or {} services = compose.setdefault("services", {}) volumes = compose.setdefault("volumes", {}) or {} cabinets_dir = SPR_ROOT / "soleprint" / "station" / "cabinets" added, skipped = [], [] for cabinet in cabinets: name = cabinet["name"] service_name = cabinet.get("service", name) # The room's own compose file is the authority. A room that already # declares `db` has arranged it deliberately, and silently replacing it # would be the worst possible outcome of switching a cabinet on. if service_name in services: skipped.append(service_name) continue fragment_path = cabinets_dir / name / "service.yml" if not fragment_path.exists(): log.warning(f" cabinet {name} has no service.yml") continue fragment = yaml.safe_load(fragment_path.read_text()) or {} for key, value in fragment.items(): if key in services: skipped.append(key) continue services[key] = value added.append(key) for volume in cabinet.get("volumes", []) or []: volumes.setdefault(volume, None) if volumes: compose["volumes"] = volumes rendered = yaml.safe_dump(compose, sort_keys=False, default_flow_style=False) banner = f"# Cabinets merged in by build.py: {', '.join(c['name'] for c in cabinets)}.\n" preamble = ("\n".join(header) + "\n" + banner + "\n") if header else banner + "\n" compose_path.write_text(preamble + rendered) if added: log.info(f" cabinets: {', '.join(added)}") if skipped: log.info(f" cabinets already declared by the room, left alone: {', '.join(skipped)}") _append_cabinet_env(output_dir, cabinets) def _append_cabinet_env(output_dir: Path, cabinets: list[dict]): """Add each cabinet's settings to .env.example, without touching .env.""" example = output_dir / ".env.example" existing = example.read_text() if example.exists() else "" # Match whole settings, not substrings: `POSTGRES_DB=` appears inside # `MY_POSTGRES_DB=`, and a substring test would decide the setting was # already there and skip it. declared = { line.split("=", 1)[0].strip() for line in existing.splitlines() if "=" in line and not line.lstrip().startswith("#") } lines = [] for cabinet in cabinets: env = cabinet.get("env", {}) if not env: continue block = [f"\n# ── {cabinet.get('title', cabinet['name'])} (cabinet) ──"] for note in cabinet.get("notes", []) or []: block.append(f"# {note}") wrote = False for key, value in env.items(): if key in declared: continue block.append(f"{key}={value}") declared.add(key) wrote = True if wrote: lines.extend(block) if lines: example.write_text(existing.rstrip("\n") + "\n" + "\n".join(lines) + "\n") def load_plexuses(room: str) -> list[dict]: """The plexuses a room asked for. Same shape as its sibling data/*.json.""" path = SPR_ROOT / "cfg" / room / "data" / "plexuses.json" if not path.exists(): return [] try: raw = json.loads(path.read_text()) except ValueError as e: log.warning(f" plexuses.json is not valid JSON, ignoring: {e}") return [] entries = raw.get("items", raw) if isinstance(raw, dict) else raw out = [] for entry in entries if isinstance(entries, list) else []: if isinstance(entry, str): entry = {"name": entry} if isinstance(entry, dict) and entry.get("name"): out.append(entry) return out def _theme_css(theme: str) -> str: """The token contract plus one theme, flattened for inlining. Only the named theme ships alongside the others it can switch to, because the export has to work with no server: there is no /theme.css to fetch. """ theme_dir = SPR_ROOT / "soleprint" / "common" / "theme" parts = [] tokens = theme_dir / "tokens.css" if tokens.exists(): parts.append(tokens.read_text()) # Every theme, so the switcher in the page has something to switch to. for sheet in sorted((theme_dir / "themes").glob("*.css")): parts.append(sheet.read_text()) return "\n".join(parts) def _inline_svg(name: str, theme: str) -> str: """A rendered graph, stripped of its XML prolog so it can sit in HTML. Inlined rather than -linked so the page's CSS can recolour it when the theme switches — graphviz writes class="node accent" into the SVG, and CSS outranks the presentation attributes it bakes in. """ graphs = SPR_ROOT / "docs" / "graphs" for candidate in (graphs / f"{name}.{theme}.svg", graphs / f"{name}.svg"): if candidate.exists(): svg = candidate.read_text() start = svg.find("= 0 else svg log.warning(f" no rendered graph '{name}' — run docs/graphs/render.sh") return "

diagram not rendered

" def build_plexuses(output_dir: Path, room: str): """Export each plexus the room declared to a single self-contained file. A plexus is exported, not served. The output is one index.html carrying its theme, its data and its diagram, so it survives a locked-down machine, a zip attachment and a double-click — which is the whole point of the format. """ requested = load_plexuses(room) if not requested: return source_root = SPR_ROOT / "soleprint" / "artery" / "plexuses" built = [] for entry in requested: name = entry["name"] source = source_root / name manifest_path = source / "plexus.json" if not manifest_path.exists(): available = sorted( p.name for p in source_root.iterdir() if p.is_dir() ) if source_root.exists() else [] log.warning( f" no such plexus: {name} (available: {', '.join(available) or 'none'})" ) continue try: manifest = json.loads(manifest_path.read_text()) except ValueError as e: log.warning(f" plexus {name} has invalid plexus.json: {e}") continue # The room may override anything the plexus declares — theme first. manifest.update({k: v for k, v in entry.items() if k != "name"}) template_path = source / "app" / "index.html" if not template_path.exists(): log.warning(f" plexus {name} has no app/index.html") continue theme = manifest.get("theme", "soleprint") data = {k: v for k, v in manifest.items() if not k.startswith("_")} # A plexus may ship a showcase.py exposing collect(): anything it returns # is merged into the page's data. The bundle uses it to run the real # tools over the real fixtures at build time, so what the page shows # cannot drift from what the tools do. collector = source / "showcase.py" if collector.exists(): try: spec = importlib.util.spec_from_file_location( f"plexus_{name}_showcase", collector ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) data["showcase"] = module.collect() except Exception as e: log.warning(f" {name}: showcase.py failed ({type(e).__name__}: {e})") page = template_path.read_text() for token, value in ( ("%%TITLE%%", manifest.get("title", name)), ("%%DESCRIPTION%%", manifest.get("description", "")), ("%%DEFAULT_THEME%%", theme), ("%%BUILT%%", f"{room} · built by soleprint build.py"), ("%%THEME_CSS%%", _theme_css(theme)), ("%%GRAPH%%", _inline_svg(manifest.get("graph", "system_overview"), theme)), ("%%BUNDLE%%", json.dumps(data, indent=2)), ): page = page.replace(token, value) target = output_dir / "plexuses" / name ensure_dir(target) (target / "index.html").write_text(page) # Anything else in app/ rides along, for a plexus that outgrows one file. for extra in (source / "app").iterdir(): if extra.name != "index.html": copy_path(extra, target / extra.name, quiet=True) built.append(f"{name} ({theme})") if built: log.info(f" plexuses: {', '.join(built)}") def build_soleprint(output_dir: Path, room: str): """Build soleprint folder with core + room config merged.""" soleprint = SPR_ROOT / "soleprint" # Soleprint core files log.info("Copying soleprint core...") for name in [ "main.py", "run.py", "index.html", "requirements.txt", "Dockerfile", ]: if (soleprint / name).exists(): copy_path(soleprint / name, output_dir / name) copy_path(soleprint / "dataloader", output_dir / "dataloader") # System directories log.info("Copying systems...") for system in ["artery", "atlas", "station"]: source = soleprint / system if source.exists(): copy_path(source, output_dir / system) # Common modules (auth, etc) if (soleprint / "common").exists(): copy_path(soleprint / "common", output_dir / "common") # Room config (includes merging room-specific artery/atlas/station) copy_cfg(output_dir, room) # Dependency containers the room asked for, merged into its compose file. # After copy_cfg, because the compose file being merged into is the room's. log.info("Composing cabinets...") compose_cabinets(output_dir, room) # Plexuses are exported rather than served, so this is a compile step like # the cabinet merge above — not something run.py does at request time. log.info("Exporting plexuses...") build_plexuses(output_dir, room) # Generate models log.info("Generating models...") if not generate_models(output_dir, room): log.warning("Model generation failed") def build(output_dir: Path, cfg_name: str | None = None, clean: bool = True): """Build complete room instance.""" room = cfg_name or "standalone" config = load_config(cfg_name) managed = config.get("managed") log.info(f"\n=== Building {room} ===") # Clean output directory first if clean and output_dir.exists(): log.info(f"Cleaning {output_dir}...") _rmtree_resilient(output_dir) ensure_dir(output_dir) if managed: # 3-folder structure: /, link/, soleprint/ build_managed(output_dir, room, config) build_link(output_dir, room) soleprint_name = config.get("framework", {}).get("name", "soleprint") soleprint_dir = output_dir / soleprint_name ensure_dir(soleprint_dir) build_soleprint(soleprint_dir, room) else: # Standalone: everything in output_dir build_soleprint(output_dir, room) # Layer 7 (optional): render kind-cluster manifests try: from soleprint.ctrl.k8s import render_k8s from soleprint.ctrl.k8s.render import k8s_enabled if k8s_enabled(config): render_k8s(room=room, config=config, gen_dir=output_dir) except ImportError as e: log.warning(f"k8s rendering unavailable: {e}") log.info(f"\n✓ Built: {output_dir}") def build_models_only(): """Only regenerate models.""" log.info("Generating models...") if generate_models(SPR_ROOT / "gen"): log.info("✓ Models generated") else: log.error("Model generation failed") sys.exit(1) def build_plexuses_only(room: str): """Compile just the plexus UIs, without rebuilding the room around them. The equivalent of `vite build` for this repo: the iteration loop when you are working on the UI itself is edit, compile, reopen the file — and a full room build to see a CSS change is a slow way to do that. """ output_dir = SPR_ROOT / "gen" / room if not output_dir.exists(): log.error(f"Room '{room}' is not built — run: python build.py --cfg {room}") sys.exit(1) log.info(f"Compiling plexus UIs for {room}...") build_plexuses(output_dir, room) built = sorted((output_dir / "plexuses").glob("*/index.html")) if not built: log.warning( f" nothing compiled — does cfg/{room}/data/plexuses.json list one?" ) return for page in built: log.info(f" {page.relative_to(SPR_ROOT)} ({page.stat().st_size // 1024} KB)") log.info("\n✓ Open directly — no server needed:") log.info(f" xdg-open {built[0]}") def main(): parser = argparse.ArgumentParser(description="Soleprint Build Tool") parser.add_argument("--output", "-o", type=Path, help="Output directory") parser.add_argument("--cfg", "-c", type=str, help="Room config name") parser.add_argument("--all", action="store_true", help="Build all rooms") parser.add_argument("--models", action="store_true", help="Only regenerate models") parser.add_argument( "--plexuses", action="store_true", help="Only compile the plexus UIs into an already-built room", ) args = parser.parse_args() if args.plexuses: build_plexuses_only(args.cfg or "standalone") elif args.models: build_models_only() elif args.all: build(SPR_ROOT / "gen" / "standalone", None) for room in (SPR_ROOT / "cfg").iterdir(): # cfg/ is itself a git repo and rooms may carry dot-dirs — skip them, # or --all tries to build ".git" as a room. if room.name.startswith(".") or room.name == "__pycache__": continue if room.is_dir() and room.name != "standalone": build(SPR_ROOT / "gen" / room.name, room.name) else: if args.output: output_dir = args.output.resolve() elif args.cfg: output_dir = SPR_ROOT / "gen" / args.cfg else: output_dir = SPR_ROOT / "gen" / "standalone" build(output_dir, args.cfg) if __name__ == "__main__": main()