ADAM October 2026 source release: PixelRow, INRFlow, Wan Video, Oasis player and field guide
f8c73f9 verified | from __future__ import annotations | |
| from pathlib import Path | |
| import os | |
| import tempfile | |
| import threading | |
| import time | |
| from typing import Any | |
| from urllib.parse import parse_qs | |
| from adam.assets import Asset, AssetRegistry | |
| from adam.commands import CommandValidationError, TrainingCommand | |
| from adam.dataset_registry import DatasetRecord, DatasetRegistry | |
| from adam.dataset_lab import scan_dataset | |
| from adam.generations import build_generation_plan, generation_tools | |
| from adam.remote_api import ( | |
| RemoteApiError, | |
| RemoteResponse, | |
| bounded_float, | |
| bounded_int, | |
| bounded_text, | |
| coerce_json_object, | |
| error_response, | |
| json_response, | |
| media_response, | |
| parse_pagination, | |
| sanitized_arguments, | |
| ) | |
| from adam.remote_dispatcher import RemoteCommandDispatcher | |
| from adam.remote_media import OpaqueIdCodec, RemoteMediaStore | |
| from adam.studio import caption_path, image_files, StudioStore | |
| from adam.training_assistant import append_preflight_summary | |
| class RemoteV1Service: | |
| """Versioned ADAM Remote API built around existing ADAM services.""" | |
| def __init__( | |
| self, | |
| *, | |
| root: Path, | |
| config: Any, | |
| jobs: Any, | |
| planner: Any, | |
| dispatcher: RemoteCommandDispatcher, | |
| codec: OpaqueIdCodec, | |
| media: RemoteMediaStore, | |
| auto_approve_training: Any, | |
| ) -> None: | |
| self.root = root.resolve() | |
| self.config = config | |
| self.jobs = jobs | |
| self.planner = planner | |
| self.dispatcher = dispatcher | |
| self.codec = codec | |
| self.media = media | |
| self.auto_approve_training = auto_approve_training | |
| self._asset_fallback: AssetRegistry | None = None | |
| self._studio: StudioStore | None = None | |
| # The dashboard requests both schemas at once. They share an asset | |
| # registry, so build one short-lived catalog instead of asking two | |
| # request threads to rediscover and rewrite it concurrently. | |
| self._model_catalog_lock = threading.RLock() | |
| self._model_catalog: list[dict[str, Any]] = [] | |
| self._model_catalog_at = 0.0 | |
| def route( | |
| self, | |
| method: str, | |
| path: str, | |
| query: str = "", | |
| payload: dict[str, Any] | None = None, | |
| ) -> RemoteResponse | None: | |
| if not path.startswith("/api/v1/"): | |
| return None | |
| parts = [part for part in path.removeprefix("/api/v1/").split("/") if part] | |
| parsed_query = parse_qs(query) | |
| try: | |
| if method == "GET": | |
| return self._get(parts, parsed_query) | |
| if method == "POST": | |
| return self._post(parts, coerce_json_object(payload or {})) | |
| except RemoteApiError as exc: | |
| return error_response(str(exc), status=exc.status) | |
| except (CommandValidationError, ValueError) as exc: | |
| return error_response(str(exc), status=400) | |
| except Exception as exc: | |
| return error_response(f"ADAM could not finish that remote request: {exc}", status=500) | |
| return error_response("Unsupported Remote API method.", status=405) | |
| def _get(self, parts: list[str], query: dict[str, list[str]]) -> RemoteResponse: | |
| if parts == ["datasets"]: | |
| return json_response({"ok": True, "datasets": self.datasets()}) | |
| if parts == ["datasets", "locations"]: | |
| return json_response({"ok": True, "locations": self.dataset_locations()}) | |
| if parts == ["training", "presets"]: | |
| return json_response({"ok": True, "presets": self.training_presets()}) | |
| if len(parts) == 2 and parts[0] == "datasets": | |
| return json_response({"ok": True, "dataset": self.dataset_detail(parts[1])}) | |
| if len(parts) == 3 and parts[0] == "datasets" and parts[2] == "thumbnail": | |
| return self.dataset_thumbnail(parts[1], query) | |
| if len(parts) == 3 and parts[0] == "datasets" and parts[2] == "items": | |
| return json_response({"ok": True, **self.dataset_items(parts[1], query)}) | |
| if len(parts) == 2 and parts[0] == "media": | |
| return self.media_file(parts[1], query) | |
| if parts == ["models"]: | |
| return json_response({"ok": True, "models": self.models()}) | |
| if parts == ["training", "schema"]: | |
| return json_response({"ok": True, **self.training_schema()}) | |
| if parts == ["generation", "schema"]: | |
| return json_response({"ok": True, **self.generation_schema()}) | |
| if len(parts) == 2 and parts[0] == "jobs": | |
| return json_response({"ok": True, "job": self.job_detail(parts[1])}) | |
| raise RemoteApiError("Unknown Remote API endpoint.", status=404) | |
| def _post(self, parts: list[str], payload: dict[str, Any]) -> RemoteResponse: | |
| if len(parts) == 5 and parts[0] == "datasets" and parts[2] == "items" and parts[4] == "caption": | |
| return json_response({"ok": True, **self.update_caption(parts[1], parts[3], payload)}) | |
| if len(parts) == 5 and parts[0] == "datasets" and parts[2] == "items" and parts[4] == "decision": | |
| return json_response({"ok": True, **self.update_decision(parts[1], parts[3], payload)}) | |
| if len(parts) == 3 and parts[0] == "datasets" and parts[2] == "favorite": | |
| return json_response({"ok": True, **self.favorite_dataset(parts[1], payload)}) | |
| if len(parts) == 3 and parts[0] == "datasets" and parts[2] == "use": | |
| return json_response({"ok": True, "dataset": self.use_dataset(parts[1])}) | |
| if parts == ["training", "plan"]: | |
| return json_response({"ok": True, "plan": self.training_plan(payload)}) | |
| if parts == ["training", "start"]: | |
| return json_response({"ok": True, **self.start_training(payload)}) | |
| if parts == ["generation", "start"]: | |
| return json_response({"ok": True, **self.start_generation(payload)}) | |
| raise RemoteApiError("Unknown Remote API endpoint.", status=404) | |
| def _assets(self) -> AssetRegistry: | |
| assets = getattr(self.planner, "assets", None) | |
| if assets is None: | |
| if self._asset_fallback is None: | |
| self._asset_fallback = AssetRegistry(self.root) | |
| assets = self._asset_fallback | |
| if hasattr(assets, "discover"): | |
| assets.discover(self.config) | |
| return assets | |
| def _registry(self) -> Any: | |
| registry = getattr(self.planner, "registry", None) | |
| if registry is None: | |
| raise RemoteApiError("The ADAM tool registry is not available.", status=503) | |
| return registry | |
| def _studio_store(self) -> StudioStore: | |
| if self._studio is None: | |
| self._studio = StudioStore(self.root) | |
| return self._studio | |
| def _dataset_registry(self) -> DatasetRegistry: | |
| return DatasetRegistry(self.root, self.config) | |
| def _asset_public_id(self, asset: Asset) -> str: | |
| return self.codec.encode({"kind": asset.kind, "asset_id": asset.id}) | |
| def _asset_from_id(self, public_id: str, *, kind: str = "") -> Asset: | |
| payload = self.codec.decode(public_id) | |
| asset_id = str(payload.get("asset_id", "")) | |
| expected_kind = kind or str(payload.get("kind", "")) | |
| for asset in self._assets().assets: | |
| if asset.id == asset_id and (not expected_kind or asset.kind == expected_kind): | |
| return asset | |
| raise RemoteApiError("Unknown resource id.", status=404) | |
| def _item_path(self, item_id: str, *, dataset: Asset | None = None) -> tuple[Asset, Path, int]: | |
| payload = self.codec.decode(item_id) | |
| if payload.get("kind") != "dataset_image": | |
| raise RemoteApiError("Unknown media id.", status=404) | |
| asset_id = str(payload.get("asset_id", "")) | |
| index = bounded_int(payload.get("index"), minimum=0, maximum=100_000, default=0, label="Image index") | |
| if dataset is not None and dataset.id != asset_id: | |
| raise RemoteApiError("Dataset image does not belong to that dataset.", status=403) | |
| dataset_asset = dataset or next((item for item in self._assets().assets if item.id == asset_id and item.kind == "dataset"), None) | |
| if dataset_asset is None: | |
| raise RemoteApiError("Unknown dataset image.", status=404) | |
| paths = image_files(dataset_asset.path, limit=100_000) | |
| if index >= len(paths): | |
| raise RemoteApiError("Dataset image is no longer available.", status=404) | |
| path = paths[index].resolve() | |
| try: | |
| path.relative_to(Path(dataset_asset.path).expanduser().resolve()) | |
| except ValueError as exc: | |
| raise RemoteApiError("Dataset image is outside its dataset.", status=403) from exc | |
| return dataset_asset, path, index | |
| def _dataset_path(asset: Asset, path: Path) -> Path: | |
| resolved = path.expanduser().resolve() | |
| if not resolved.is_relative_to(Path(asset.path).expanduser().resolve()): | |
| raise RemoteApiError("File is outside its dataset.", status=403) | |
| return resolved | |
| def _dataset_counts(self, asset: Asset) -> dict[str, int]: | |
| review = self._studio_store().review(asset.path) | |
| images = image_files(asset.path, limit=100_000) | |
| resolved = {str(path.resolve()) for path in images} | |
| keep = sum(1 for path, decision in review.decisions.items() if decision == "keep" and path in resolved) | |
| reject = sum(1 for path, decision in review.decisions.items() if decision == "reject" and path in resolved) | |
| return { | |
| "accepted": keep, | |
| "rejected": reject, | |
| "unreviewed": max(0, len(images) - keep - reject), | |
| } | |
| def _dimensions(self, path: Path) -> dict[str, int]: | |
| try: | |
| from PIL import Image | |
| with Image.open(path) as image: | |
| return {"width": int(image.width), "height": int(image.height)} | |
| except Exception: | |
| return {"width": 0, "height": 0} | |
| def datasets(self) -> list[dict[str, Any]]: | |
| assets = self._assets() | |
| records = self._dataset_registry().discover(asset_registry=assets) | |
| by_path = {str(Path(asset.path).expanduser().resolve()): asset for asset in assets.assets if asset.kind == "dataset"} | |
| rows = [] | |
| for record in records[:300]: | |
| asset = by_path.get(str(Path(record.path).expanduser().resolve())) | |
| if asset is None and record.exists: | |
| asset = assets.register( | |
| kind="dataset", | |
| name=record.name, | |
| path=record.path, | |
| metadata={ | |
| "dataset_registry_source": record.source, | |
| "dataset_location_id": record.location_id, | |
| }, | |
| persist=False, | |
| ) | |
| if asset is None: | |
| continue | |
| rows.append(self._dataset_summary(asset, record)) | |
| if rows: | |
| assets.save() | |
| return rows | |
| def _dataset_summary(self, asset: Asset, record: DatasetRecord | None = None) -> dict[str, Any]: | |
| if record is None: | |
| record = self._dataset_registry().record_for_path(asset.path) | |
| exists = Path(asset.path).is_dir() | |
| public_id = self._asset_public_id(asset) | |
| if exists: | |
| review_state = self._studio_store().review(asset.path) | |
| accepted = sum(1 for decision in review_state.decisions.values() if decision == "keep") | |
| rejected = sum(1 for decision in review_state.decisions.values() if decision == "reject") | |
| review = { | |
| "accepted": accepted, | |
| "rejected": rejected, | |
| "unreviewed": max(0, record.image_count - accepted - rejected), | |
| } | |
| else: | |
| review = {"accepted": 0, "rejected": 0, "unreviewed": 0} | |
| warnings = list(record.warnings) | |
| if not exists and "Dataset folder is unavailable." not in warnings: | |
| warnings.append("Dataset folder is unavailable.") | |
| return { | |
| "id": public_id, | |
| "name": asset.name or record.name, | |
| "exists": exists, | |
| "available": exists, | |
| "source": record.source, | |
| "location_id": record.location_id, | |
| "favorite": record.favorite, | |
| "last_used_at": record.last_used_at, | |
| "item_count": record.item_count, | |
| "image_count": record.image_count, | |
| "video_count": record.video_count, | |
| "caption_count": record.caption_count, | |
| "missing_caption_count": record.missing_caption_count, | |
| "dataset_format": record.dataset_format, | |
| "warnings": warnings[:6], | |
| "thumbnail_url": f"/api/v1/datasets/{public_id}/thumbnail" if exists else "", | |
| "review": review, | |
| } | |
| def dataset_locations(self) -> list[dict[str, Any]]: | |
| registry = self._dataset_registry() | |
| return [ | |
| { | |
| "id": item.id, | |
| "name": item.name, | |
| "source": item.source, | |
| "exists": Path(item.path).is_dir(), | |
| "available": Path(item.path).is_dir(), | |
| "last_seen_at": item.last_seen_at, | |
| } | |
| for item in registry.known_locations() | |
| ][:200] | |
| def _first_image(self, folder: str | Path) -> Path | None: | |
| root = Path(folder).expanduser() | |
| if not root.is_dir(): | |
| return None | |
| try: | |
| for path in root.rglob("*"): | |
| if path.is_file() and path.suffix.casefold() in {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif"}: | |
| return path | |
| except OSError: | |
| return None | |
| return None | |
| def dataset_detail(self, dataset_id: str) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| registry = self._dataset_registry() | |
| record = registry.record_for_path(asset.path) | |
| if not Path(asset.path).is_dir(): | |
| return self._dataset_summary(asset, record) | |
| report = scan_dataset(asset.path, limit=300) | |
| registry.refresh_async(asset.path, source=record.source, location_id=record.location_id) | |
| return { | |
| "id": self._asset_public_id(asset), | |
| "name": asset.name, | |
| "exists": True, | |
| "available": True, | |
| "source": record.source, | |
| "location_id": record.location_id, | |
| "favorite": record.favorite, | |
| "last_used_at": record.last_used_at, | |
| "image_count": report.image_count, | |
| "video_count": report.video_count, | |
| "item_count": report.image_count + report.video_count, | |
| "caption_count": report.caption_count, | |
| "missing_caption_count": report.missing_caption_count, | |
| "duplicate_groups": report.duplicate_groups, | |
| "dimensions": dict(list(report.dimensions.items())[:20]), | |
| "extensions": report.extensions, | |
| "warnings": report.warnings, | |
| "dataset_format": record.dataset_format, | |
| "thumbnail_url": f"/api/v1/datasets/{dataset_id}/thumbnail", | |
| "review": self._dataset_counts(asset), | |
| } | |
| def dataset_thumbnail(self, dataset_id: str, query: dict[str, list[str]]) -> RemoteResponse: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| source = self._first_image(asset.path) | |
| if source is None: | |
| raise RemoteApiError("Dataset thumbnail is not available.", status=404) | |
| source = self._dataset_path(asset, source) | |
| size = bounded_int((query.get("size") or ["320"])[0], minimum=64, maximum=640, default=320, label="Media size") | |
| media = self.media.thumbnail(source, size=size) | |
| return media_response(media.path.read_bytes(), media.content_type, cache_seconds=86400) | |
| def dataset_items(self, dataset_id: str, query: dict[str, list[str]]) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| page = parse_pagination(query, default_size=24, max_size=60) | |
| paths = image_files(asset.path, limit=100_000) | |
| total = len(paths) | |
| review = self._studio_store().review(asset.path) | |
| rows = [] | |
| for index, path in enumerate(paths[page["offset"]: page["offset"] + page["limit"]], page["offset"]): | |
| path = self._dataset_path(asset, path) | |
| item_id = self.media.media_id(kind="dataset_image", asset_id=asset.id, index=index) | |
| caption_file = self._dataset_path(asset, caption_path(path)) | |
| try: | |
| caption = "" | |
| if caption_file.is_file(): | |
| with caption_file.open(encoding="utf-8") as handle: | |
| caption = handle.read(4000) | |
| except (OSError, UnicodeError): | |
| caption = "" | |
| rows.append({ | |
| "id": item_id, | |
| "display_name": path.name, | |
| "dimensions": self._dimensions(path), | |
| "caption": caption, | |
| "has_caption": bool(caption.strip()), | |
| "decision": review.decisions.get(str(path.resolve()), "unreviewed"), | |
| "thumbnail_url": f"/api/v1/media/{item_id}?size=320", | |
| "preview_url": f"/api/v1/media/{item_id}?size=960", | |
| }) | |
| return { | |
| "dataset": self.dataset_detail(dataset_id), | |
| "items": rows, | |
| "pagination": { | |
| "page": page["page"], | |
| "page_size": page["page_size"], | |
| "total": total, | |
| "has_next": page["offset"] + page["limit"] < total, | |
| }, | |
| } | |
| def media_file(self, media_id: str, query: dict[str, list[str]]) -> RemoteResponse: | |
| _dataset, source, _index = self._item_path(media_id) | |
| size = bounded_int((query.get("size") or ["320"])[0], minimum=64, maximum=1200, default=320, label="Media size") | |
| media = self.media.thumbnail(source, size=size) | |
| return media_response(media.path.read_bytes(), media.content_type, cache_seconds=86400) | |
| def update_caption(self, dataset_id: str, item_id: str, payload: dict[str, Any]) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| _asset, image, _index = self._item_path(item_id, dataset=asset) | |
| caption = bounded_text(payload.get("caption", ""), max_length=4000, label="Caption") | |
| target = self._dataset_path(asset, caption_path(image)) | |
| # Replace the file instead of writing through a possible hard link. | |
| temporary = None | |
| try: | |
| with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=target.parent, delete=False) as handle: | |
| temporary = Path(handle.name) | |
| handle.write(caption.rstrip() + ("\n" if caption else "")) | |
| os.replace(temporary, target) | |
| finally: | |
| if temporary is not None: | |
| temporary.unlink(missing_ok=True) | |
| return {"message": "Caption saved.", "item": {"id": item_id, "caption": caption, "has_caption": bool(caption)}} | |
| def update_decision(self, dataset_id: str, item_id: str, payload: dict[str, Any]) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| _asset, image, _index = self._item_path(item_id, dataset=asset) | |
| decision = bounded_text(payload.get("decision", "unreviewed"), max_length=32, label="Decision") | |
| if decision not in {"keep", "reject", "unreviewed"}: | |
| raise RemoteApiError("Decision must be keep, reject, or unreviewed.") | |
| self._studio_store().set_decision(asset.path, str(image), decision) | |
| return {"message": "Review decision saved.", "item": {"id": item_id, "decision": decision}} | |
| def favorite_dataset(self, dataset_id: str, payload: dict[str, Any]) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| enabled = bool(payload.get("favorite", True)) | |
| record = self._dataset_registry().favorite(asset.path, enabled) | |
| return { | |
| "message": "Dataset favorite updated.", | |
| "dataset": self._dataset_summary(asset, record), | |
| } | |
| def use_dataset(self, dataset_id: str) -> dict[str, Any]: | |
| asset = self._asset_from_id(dataset_id, kind="dataset") | |
| record = self._dataset_registry().touch(asset.path) | |
| return self._dataset_summary(asset, record) | |
| def models(self) -> list[dict[str, Any]]: | |
| with self._model_catalog_lock: | |
| if time.monotonic() - self._model_catalog_at < 1.0: | |
| return list(self._model_catalog) | |
| assets = self._assets() | |
| experiment_by_model: dict[str, Any] = {} | |
| try: | |
| for run in getattr(getattr(self.jobs, "experiments", None), "list_runs", lambda limit=100: [])(limit=100): | |
| experiment_by_model.setdefault(run.model_name, run) | |
| except Exception: | |
| experiment_by_model = {} | |
| rows = [] | |
| for asset in assets.assets: | |
| if asset.kind not in {"model", "base_model"} or not Path(asset.path).exists(): | |
| continue | |
| dataset = next((item for item in assets.assets if item.id == asset.dataset_id), None) | |
| metadata = dict(asset.metadata or {}) | |
| trigger_word = str(metadata.get("trigger_word") or (asset.name if asset.trainer == "lora" else "")) | |
| latest = experiment_by_model.get(asset.name) | |
| rows.append({ | |
| "id": self._asset_public_id(asset), | |
| "name": asset.name, | |
| "kind": asset.kind, | |
| "architecture": asset.trainer or ("stable_diffusion" if asset.kind == "base_model" else ""), | |
| "trainer": asset.trainer, | |
| "checkpoint_name": Path(asset.checkpoint or asset.path).name, | |
| "dataset": None if dataset is None else {"id": self._asset_public_id(dataset), "name": dataset.name}, | |
| "epochs": asset.epochs, | |
| "trigger_word": trigger_word, | |
| "latest_experiment": None if latest is None else { | |
| "id": latest.id, | |
| "status": latest.status, | |
| "resolution": latest.resolution, | |
| "dataset_name": latest.dataset_name, | |
| "trigger_word": getattr(latest, "trigger_word", ""), | |
| }, | |
| }) | |
| self._model_catalog = rows[:300] | |
| self._model_catalog_at = time.monotonic() | |
| return list(self._model_catalog) | |
| def training_schema(self) -> dict[str, Any]: | |
| registry = self._registry() | |
| trainers = [] | |
| for tool in registry.enabled(): | |
| if not tool.id.endswith("_trainer"): | |
| continue | |
| trainer = tool.id.removesuffix("_trainer") | |
| schema = registry.model_plugins.training_schema(trainer) | |
| trainers.append({ | |
| "id": trainer, | |
| "tool_id": tool.id, | |
| "name": tool.name, | |
| "requires_confirmation": tool.requires_confirmation, | |
| "settings": schema, | |
| "arguments": [arg for arg in tool.arguments if arg not in {"dataset_dir", "output_dir"}], | |
| "required_arguments": [arg for arg in tool.required_arguments if arg not in {"dataset_dir", "output_dir"}], | |
| }) | |
| return { | |
| "trainers": trainers, | |
| "datasets": self.datasets(), | |
| "base_models": [item for item in self.models() if item["kind"] == "base_model"], | |
| "presets": self.training_presets(), | |
| } | |
| def training_presets(self) -> list[dict[str, Any]]: | |
| presets = [ | |
| { | |
| "id": "standard_ddpm", | |
| "name": "Standard DDPM", | |
| "trainer": "ddpm", | |
| "epochs": 100, | |
| "settings": { | |
| "resolution": 128, | |
| "batch_size": 1, | |
| "learning_rate": 0.0001, | |
| "save_every": 10, | |
| "preview_enabled": True, | |
| "preview_every": 5, | |
| }, | |
| }, | |
| { | |
| "id": "quick_ddpm_test", | |
| "name": "Quick DDPM Test", | |
| "trainer": "ddpm", | |
| "epochs": 3, | |
| "settings": { | |
| "resolution": 64, | |
| "batch_size": 1, | |
| "learning_rate": 0.0001, | |
| "save_every": 1, | |
| "preview_enabled": True, | |
| "preview_every": 1, | |
| "training_intensity": 25, | |
| }, | |
| }, | |
| { | |
| "id": "oasis_training", | |
| "name": "Oasis Training", | |
| "trainer": "oasis", | |
| "epochs": 25, | |
| "settings": { | |
| "resolution": "256x144", | |
| "batch_size": 2, | |
| "learning_rate": 0.00002, | |
| "workers": 2, | |
| "preview_enabled": True, | |
| "preview_every": 5, | |
| }, | |
| }, | |
| ] | |
| for recipe in self._studio_store().recipes: | |
| presets.append({ | |
| "id": f"recipe_{recipe.id}", | |
| "name": recipe.name, | |
| "trainer": recipe.trainer, | |
| "epochs": recipe.epochs, | |
| "settings": { | |
| "preview_prompt": recipe.preview_prompt, | |
| **({"base_model": recipe.base_model} if recipe.base_model else {}), | |
| }, | |
| "user": True, | |
| }) | |
| available = {trainer["id"] for trainer in self.training_schema_no_presets()} | |
| return [preset for preset in presets if preset["trainer"] in available] | |
| def training_schema_no_presets(self) -> list[dict[str, Any]]: | |
| registry = self._registry() | |
| trainers = [] | |
| for tool in registry.enabled(): | |
| if not tool.id.endswith("_trainer"): | |
| continue | |
| trainer = tool.id.removesuffix("_trainer") | |
| trainers.append({"id": trainer}) | |
| return trainers | |
| def _build_training_plan(self, payload: dict[str, Any]) -> Any: | |
| if self.planner is None: | |
| raise RemoteApiError("Planning is not available in this ADAM session.", status=503) | |
| trainer = bounded_text(payload.get("trainer", "lora"), max_length=64, label="Trainer", required=True).casefold() | |
| dataset = self._asset_from_id(bounded_text(payload.get("dataset_id"), max_length=4000, label="Dataset", required=True), kind="dataset") | |
| if not Path(dataset.path).is_dir(): | |
| raise RemoteApiError("That dataset is unavailable on the PC.", status=404) | |
| model_name = bounded_text(payload.get("model_name") or dataset.name, max_length=96, label="Model name", required=True) | |
| epochs = bounded_int(payload.get("epochs"), minimum=1, maximum=100_000, default=10, label="Epochs") | |
| options = dict(payload.get("settings") or payload.get("training_options") or {}) | |
| if not isinstance(options, dict): | |
| raise RemoteApiError("Training settings must be an object.") | |
| trigger_word = bounded_text(payload.get("trigger_word") or options.get("trigger_word") or "", max_length=128, label="Trigger word") | |
| if trigger_word: | |
| options["trigger_word"] = trigger_word | |
| base_model = "" | |
| if trainer == "lora": | |
| base_id = bounded_text(payload.get("base_model_id", ""), max_length=4000, label="Base model") | |
| if base_id: | |
| base_model = self._asset_from_id(base_id, kind="base_model").path | |
| else: | |
| base_model = str(options.get("base_model") or "") | |
| if not base_model and hasattr(self.planner, "_lora_base_model"): | |
| base_model = str(self.planner._lora_base_model()) | |
| if base_model: | |
| options["base_model"] = base_model | |
| output = "" | |
| if hasattr(self.planner, "_training_output"): | |
| output_path = self.planner._training_output(trainer, model_name) | |
| output = str(output_path) if output_path else "" | |
| command = TrainingCommand.from_dict({ | |
| "action": "train", | |
| "trainer": trainer, | |
| "dataset": dataset.path, | |
| "model_name": model_name, | |
| "epochs": epochs, | |
| "output": output, | |
| "base_model": base_model, | |
| "trigger_word": trigger_word, | |
| "training_options": options, | |
| }) | |
| if not hasattr(self.planner, "_plan_training_command"): | |
| raise RemoteApiError("Structured training validation is not available.", status=503) | |
| plan = self.planner._plan_training_command("Remote structured training", command) | |
| append_preflight_summary(plan, self.config) | |
| return plan | |
| def training_plan(self, payload: dict[str, Any]) -> dict[str, Any]: | |
| plan = self._build_training_plan(payload) | |
| return self._plan_payload(plan) | |
| def start_training(self, payload: dict[str, Any]) -> dict[str, Any]: | |
| if self.jobs is None: | |
| raise RemoteApiError("Jobs are not available in this ADAM session.", status=503) | |
| plan = self._build_training_plan(payload) | |
| if not plan.steps: | |
| raise RemoteApiError(plan.summary or "ADAM could not build a runnable training plan.") | |
| job = self.dispatcher.submit_job(self.jobs, plan) | |
| auto_approved = False | |
| if self.auto_approve_training(plan): | |
| self.dispatcher.confirm_job(self.jobs, job.id) | |
| auto_approved = True | |
| try: | |
| dataset = self._asset_from_id(bounded_text(payload.get("dataset_id"), max_length=4000, label="Dataset"), kind="dataset") | |
| self._dataset_registry().touch(dataset.path) | |
| except Exception: | |
| pass | |
| return { | |
| "message": f"Queued {job.plan.project_name}.", | |
| "job_id": job.id, | |
| "requires_approval": bool(plan.requires_confirmation and not auto_approved), | |
| "auto_approved": auto_approved, | |
| "plan": self._plan_payload(plan), | |
| } | |
| def generation_schema(self) -> dict[str, Any]: | |
| registry = self._registry() | |
| providers = [] | |
| for tool in generation_tools(registry): | |
| providers.append({ | |
| "id": tool.id, | |
| "name": tool.name, | |
| "model_trainers": tool.model_trainers, | |
| "requires_confirmation": tool.requires_confirmation, | |
| "options": tool.generation_options, | |
| "settings": registry.model_plugins.generation_schema_for_tool(tool.id), | |
| }) | |
| models = self.models() | |
| return { | |
| "providers": providers, | |
| "models": [item for item in models if item["kind"] == "model"], | |
| "base_models": [item for item in models if item["kind"] == "base_model"], | |
| } | |
| def start_generation(self, payload: dict[str, Any]) -> dict[str, Any]: | |
| if self.jobs is None: | |
| raise RemoteApiError("Jobs are not available in this ADAM session.", status=503) | |
| registry = self._registry() | |
| provider_id = bounded_text(payload.get("provider_id", ""), max_length=96, label="Provider", required=True) | |
| tool = next((item for item in generation_tools(registry) if item.id == provider_id), None) | |
| if tool is None: | |
| raise RemoteApiError("Unknown generation provider.", status=404) | |
| model = self._asset_from_id(bounded_text(payload.get("model_id", ""), max_length=4000, label="Model", required=True), kind="model") | |
| if model.trainer not in tool.model_trainers: | |
| raise RemoteApiError("That model is not compatible with the selected provider.") | |
| prompt = bounded_text(payload.get("prompt", ""), max_length=2000, label="Prompt") | |
| count = bounded_int(payload.get("image_count"), minimum=1, maximum=32, default=1, label="Image count") | |
| if provider_id == "lora_generator": | |
| count = min(count, 8) | |
| options = tool.generation_options | |
| step_max = int(options.get("step_max", 500) or 500) | |
| steps = bounded_int(payload.get("steps"), minimum=1, maximum=step_max, default=int(options.get("step_default", 30) or 30), label="Steps") | |
| seed = bounded_int(payload.get("seed"), minimum=0, maximum=2_147_483_647, default=0, label="Seed") | |
| sampler_options = [str(item) for item in options.get("samplers", [])] | |
| sampler = bounded_text(payload.get("sampler") or (sampler_options[0] if sampler_options else "DDIM"), max_length=64, label="Sampler") | |
| if sampler_options and sampler not in sampler_options: | |
| raise RemoteApiError("Sampler is not supported by that provider.") | |
| aspect_options = [str(item) for item in options.get("aspect_ratios", [])] | |
| aspect = bounded_text(payload.get("aspect_ratio") or (aspect_options[0] if aspect_options else "1:1 (Square)"), max_length=64, label="Aspect ratio") | |
| if aspect_options and aspect not in aspect_options: | |
| raise RemoteApiError("Aspect ratio is not supported by that provider.") | |
| extra: dict[str, Any] = {} | |
| if provider_id == "lora_generator": | |
| base_id = bounded_text(payload.get("base_model_id", ""), max_length=4000, label="Base model") | |
| base_model_path = self._asset_from_id(base_id, kind="base_model").path if base_id else "" | |
| extra.update({ | |
| "negative_prompt": bounded_text(payload.get("negative_prompt", ""), max_length=3000, label="Negative prompt"), | |
| "base_model_path": base_model_path, | |
| "width": bounded_int(payload.get("width"), minimum=0, maximum=2048, default=0, label="Width"), | |
| "height": bounded_int(payload.get("height"), minimum=0, maximum=2048, default=0, label="Height"), | |
| "cfg_scale": bounded_float(payload.get("cfg_scale"), minimum=0.0, maximum=30.0, default=0.0, label="CFG scale"), | |
| "lora_strength": bounded_float(payload.get("lora_strength"), minimum=0.0, maximum=3.0, default=1.0, label="LoRA strength"), | |
| "denoise_strength": bounded_float(payload.get("denoise_strength"), minimum=0.0, maximum=1.0, default=0.0, label="Denoise strength"), | |
| "prompt_weighting": bool(payload.get("prompt_weighting", True)), | |
| }) | |
| elif provider_id == "ddpm_generator": | |
| extra.update({ | |
| "reference_strength": bounded_int(payload.get("reference_strength"), minimum=0, maximum=100, default=65, label="Reference strength"), | |
| "width": bounded_int(payload.get("width"), minimum=0, maximum=2048, default=0, label="Width"), | |
| "height": bounded_int(payload.get("height"), minimum=0, maximum=2048, default=0, label="Height"), | |
| }) | |
| plugin_settings = payload.get("settings") or {} | |
| if isinstance(plugin_settings, dict): | |
| extra.update(plugin_settings) | |
| plan = build_generation_plan( | |
| tool, | |
| model_name=model.name, | |
| model_path=model.path, | |
| prompt=prompt, | |
| image_count=count, | |
| steps=steps, | |
| seed=seed, | |
| sampler=sampler, | |
| aspect_ratio=aspect, | |
| extra_arguments=extra, | |
| ) | |
| job = self.dispatcher.submit_job(self.jobs, plan) | |
| return { | |
| "message": f"Queued {job.plan.project_name}.", | |
| "job_id": job.id, | |
| "requires_approval": bool(plan.requires_confirmation), | |
| "plan": self._plan_payload(plan), | |
| } | |
| def job_detail(self, job_id: str) -> dict[str, Any]: | |
| if self.jobs is None or not hasattr(self.jobs, "get"): | |
| raise RemoteApiError("Jobs are not available in this ADAM session.", status=503) | |
| job = self.jobs.get(job_id) | |
| if job is None: | |
| raise RemoteApiError("ADAM could not find that job.", status=404) | |
| return { | |
| "id": job.id, | |
| "project": job.plan.project_name, | |
| "status": job.status.value, | |
| "progress": job.progress, | |
| "current_step": job.current_step, | |
| "logs": list(job.logs)[-40:], | |
| "steps": [ | |
| { | |
| "tool_id": step.tool_id, | |
| "title": step.title, | |
| "description": step.description, | |
| "status": step.status.value, | |
| "arguments": sanitized_arguments(dict(step.arguments)), | |
| } | |
| for step in job.plan.steps | |
| ], | |
| } | |
| def _plan_payload(plan: Any) -> dict[str, Any]: | |
| return { | |
| "id": getattr(plan, "id", ""), | |
| "summary": getattr(plan, "summary", ""), | |
| "project_name": getattr(plan, "project_name", ""), | |
| "requires_confirmation": bool(getattr(plan, "requires_confirmation", False)), | |
| "confirmation_reason": getattr(plan, "confirmation_reason", ""), | |
| "steps": [ | |
| { | |
| "tool_id": step.tool_id, | |
| "title": step.title, | |
| "description": step.description, | |
| "arguments": sanitized_arguments(dict(step.arguments)), | |
| } | |
| for step in getattr(plan, "steps", []) | |
| ], | |
| } | |