| from __future__ import annotations |
|
|
| import html |
| import json |
| import re |
| from dataclasses import asdict |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
| from adam.executor import ToolContext |
| from adam.monitoring import SystemMonitor |
|
|
|
|
| def _slug(value: str) -> str: |
| cleaned = re.sub(r"[^A-Za-z0-9._ -]+", "", value).strip(" .") |
| cleaned = re.sub(r"\s+", "_", cleaned) |
| return (cleaned or "adam_project")[:80] |
|
|
|
|
| def _project_folder(context: ToolContext, project_name: str) -> Path: |
| base = (context.root / "data" / "projects").resolve() |
| folder = (base / _slug(project_name)).resolve() |
| if base != folder and base not in folder.parents: |
| raise ValueError("Project folder resolved outside ADAM's data directory.") |
| folder.mkdir(parents=True, exist_ok=True) |
| return folder |
|
|
|
|
| def _write_json(path: Path, payload: dict[str, Any]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| temporary = path.with_suffix(path.suffix + ".tmp") |
| temporary.write_text(json.dumps(payload, indent=2), encoding="utf-8") |
| temporary.replace(path) |
|
|
|
|
| def _simulate( |
| context: ToolContext, |
| updates: list[tuple[int, str]], |
| *, |
| multiplier: float = 1.0, |
| ) -> None: |
| for percent, message in updates: |
| context.log(message) |
| context.progress(percent, message) |
| context.wait(multiplier) |
|
|
|
|
| def collect_dataset( |
| context: ToolContext, |
| subject: str, |
| image_count: int, |
| project_name: str, |
| ) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| dataset = folder / "dataset" |
| dataset.mkdir(exist_ok=True) |
| _simulate( |
| context, |
| [ |
| (8, f"Preparing collection query for {subject}"), |
| (28, "Checking collector configuration"), |
| (52, "Creating candidate image manifest"), |
| (76, "Recording source and license review fields"), |
| (100, "Dataset collection manifest is ready"), |
| ], |
| ) |
| manifest = { |
| "mode": "demo", |
| "notice": ( |
| "No images were downloaded. Connect your Dataset Collector backend in " |
| "config/tools.json to perform real collection." |
| ), |
| "subject": subject, |
| "requested_images": int(image_count), |
| "candidates": [], |
| "created_at": datetime.now(timezone.utc).isoformat(), |
| } |
| _write_json(dataset / "collection_manifest.json", manifest) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def prepare_dataset(context: ToolContext, project_name: str) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| dataset = folder / "dataset" |
| dataset.mkdir(exist_ok=True) |
| _simulate( |
| context, |
| [ |
| (12, "Validating dataset manifest"), |
| (34, "Checking file integrity and dimensions"), |
| (58, "Running duplicate analysis"), |
| (81, "Preparing normalized dataset layout"), |
| (100, "Dataset preparation report is ready"), |
| ], |
| ) |
| report = { |
| "mode": "demo", |
| "valid_images": 0, |
| "duplicates_removed": 0, |
| "rejected_images": 0, |
| "ready_for_captioning": False, |
| "notice": "Connect a real preparation backend to process collected files.", |
| } |
| _write_json(dataset / "preparation_report.json", report) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def generate_captions( |
| context: ToolContext, |
| subject: str, |
| project_name: str, |
| ) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| captions = folder / "captions" |
| captions.mkdir(exist_ok=True) |
| _simulate( |
| context, |
| [ |
| (15, "Loading prepared dataset report"), |
| (39, "Preparing captioning policy"), |
| (67, "Creating editable caption template"), |
| (88, "Checking caption consistency"), |
| (100, "Caption review file is ready"), |
| ], |
| ) |
| (captions / "captions_demo.txt").write_text( |
| "# ADAM demo caption template\n" |
| f"# Subject: {subject}\n" |
| "# No image captions were generated because no real backend is connected.\n", |
| encoding="utf-8", |
| ) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def train_lora( |
| context: ToolContext, |
| subject: str, |
| project_name: str, |
| epochs: int, |
| ) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| training = folder / "training" |
| training.mkdir(exist_ok=True) |
| updates = [(4, "Validating trainer configuration")] |
| for epoch in range(1, max(1, int(epochs)) + 1): |
| percent = 8 + int(epoch / max(1, int(epochs)) * 84) |
| updates.append((percent, f"Simulating epoch {epoch}/{epochs}")) |
| updates.extend( |
| [(96, "Writing transparent demo summary"), (100, "Training simulation complete")] |
| ) |
| _simulate(context, updates, multiplier=0.65) |
| _write_json( |
| training / "training_summary.json", |
| { |
| "mode": "demo", |
| "subject": subject, |
| "epochs_requested": int(epochs), |
| "model_created": False, |
| "notice": ( |
| "This was a workflow simulation. No GPU training ran and no model " |
| "weights were created." |
| ), |
| }, |
| ) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def generate_previews( |
| context: ToolContext, |
| subject: str, |
| project_name: str, |
| preview_count: int, |
| model_name: str = "", |
| checkpoint: str = "", |
| prompt: str = "", |
| seed: int = 0, |
| ) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| previews = folder / "previews" |
| previews.mkdir(exist_ok=True) |
| count = max(1, min(int(preview_count), 100)) |
| _simulate( |
| context, |
| [ |
| (10, "Loading preview generator configuration"), |
| (30, f"Preparing {count} preview tasks"), |
| (62, "Rendering demo preview cards"), |
| (86, "Writing preview manifest"), |
| (100, "Preview outputs are ready"), |
| ], |
| ) |
| safe_subject = html.escape(subject) |
| for index in range(1, count + 1): |
| svg = f"""<svg xmlns="http://www.w3.org/2000/svg" width="768" height="512"> |
| <defs><linearGradient id="g" x1="0" y1="0" x2="1" y2="1"> |
| <stop offset="0" stop-color="#07111c"/><stop offset="1" stop-color="#0b2131"/></linearGradient> |
| </defs><rect width="768" height="512" fill="url(#g)"/> |
| <circle cx="384" cy="210" r="72" fill="#139cff" opacity=".13"/> |
| <circle cx="384" cy="210" r="38" fill="#eff8ff"/> |
| <g fill="none" stroke-width="6" opacity=".85"> |
| <ellipse cx="384" cy="210" rx="150" ry="55" stroke="#55e75b"/> |
| <ellipse cx="384" cy="210" rx="150" ry="55" stroke="#168cff" transform="rotate(60 384 210)"/> |
| <ellipse cx="384" cy="210" rx="150" ry="55" stroke="#ff3948" transform="rotate(120 384 210)"/> |
| </g> |
| <text x="384" y="385" fill="#f3f8fd" font-size="30" text-anchor="middle" |
| font-family="Segoe UI, sans-serif">{safe_subject}</text> |
| <text x="384" y="425" fill="#6d8294" font-size="18" text-anchor="middle" |
| font-family="Segoe UI, sans-serif">ADAM DEMO PREVIEW {index:02d}</text></svg>""" |
| (previews / f"preview_{index:02d}.svg").write_text(svg, encoding="utf-8") |
| _write_json( |
| previews / "preview_manifest.json", |
| { |
| "mode": "demo", |
| "subject": subject, |
| "model_name": model_name, |
| "checkpoint": checkpoint, |
| "prompt": prompt, |
| "seed": int(seed), |
| "preview_count": count, |
| "notice": "These are branded placeholders, not model-generated images.", |
| }, |
| ) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def notify_complete(context: ToolContext, project_name: str) -> dict[str, Any]: |
| folder = _project_folder(context, project_name) |
| _simulate( |
| context, |
| [ |
| (30, "Collecting workflow results"), |
| (70, "Recording completion status"), |
| (100, "Workflow complete"), |
| ], |
| multiplier=0.5, |
| ) |
| _write_json( |
| folder / "completion.json", |
| { |
| "job_id": context.job_id, |
| "project_name": project_name, |
| "completed_at": datetime.now(timezone.utc).isoformat(), |
| }, |
| ) |
| return {"output_folder": str(folder)} |
|
|
|
|
| def inspect_system(context: ToolContext, project_name: str) -> dict[str, Any]: |
| del project_name |
| context.progress(20, "Reading system sensors") |
| monitor = SystemMonitor(context.root) |
| try: |
| snapshot = monitor.snapshot() |
| finally: |
| monitor.close() |
| context.log( |
| f"CPU {snapshot.cpu_percent:.0f}% 路 RAM {snapshot.memory_percent:.0f}% 路 " |
| f"GPU {snapshot.gpu_percent:.0f}% 路 {snapshot.gpu_name}" |
| ) |
| context.progress(100, "System snapshot complete") |
| return {"system_snapshot": asdict(snapshot)} |
|
|
