goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases 1 Wiki Activity

khh
Some checks failed
smoke-mm / Install & smoke test mm --help (push) Has been cancelled

Browse Source
This commit is contained in:
nose
2025-12-24 02:13:21 -08:00
parent 8bf04c6b71
commit 24dd18de7e
20 changed files with 1792 additions and 636 deletions
Download Patch File Download Diff File Expand all files Collapse all files

455
TUI/pipeline_runner.py
View File

@@ -1,18 +1,16 @@
"""Pipeline execution utilities for the Textual UI.
This module mirrors the CLI pipeline behaviour while exposing a class-based
interface that the TUI can call. It keeps all pipeline/cmdlet integration in
one place so the interface layer stays focused on presentation.
The TUI is a frontend to the CLI, so it must use the same pipeline executor
implementation as the CLI (`CLI.PipelineExecutor`).
"""
from __future__ import annotations
import contextlib
import io
import shlex
import uuid
from dataclasses import dataclass, field
import sys
from pathlib import Path
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Sequence
BASE_DIR = Path(__file__).resolve().parent
@@ -23,11 +21,10 @@ for path in (ROOT_DIR, BASE_DIR):
sys.path.insert(0, str_path)
import pipeline as ctx
from cmdlet import REGISTRY
from config import get_local_storage_path, load_config
from SYS.worker_manager import WorkerManager
from CLI import MedeiaCLI
from CLI import ConfigLoader, PipelineExecutor as CLIPipelineExecutor, WorkerManagerRegistry
from SYS.logger import set_debug
from rich_display import capture_rich_output
from result_table import ResultTable
@dataclass(slots=True)
@@ -73,24 +70,16 @@ class PipelineRunResult:
}
class PipelineExecutor:
"""Thin wrapper over the cmdlet registry + pipeline context."""
class PipelineRunner:
"""TUI wrapper that delegates to the canonical CLI pipeline executor."""
def __init__(
self,
*,
config: Optional[Dict[str, Any]] = None,
worker_manager: Optional[WorkerManager] = None,
) -> None:
self._config = config or load_config()
self._worker_manager = worker_manager
if self._worker_manager is None:
self._worker_manager = self._ensure_worker_manager()
if self._worker_manager:
self._config["_worker_manager"] = self._worker_manager
def __init__(self) -> None:
self._config_loader = ConfigLoader(root=ROOT_DIR)
self._executor = CLIPipelineExecutor(config_loader=self._config_loader)
self._worker_manager = None
@property
def worker_manager(self) -> Optional[WorkerManager]:
def worker_manager(self):
return self._worker_manager
def run_pipeline(
@@ -98,290 +87,214 @@ class PipelineExecutor:
pipeline_text: str,
*,
seeds: Optional[Any] = None,
isolate: bool = False,
on_log: Optional[Callable[[str], None]] = None,
) -> PipelineRunResult:
"""Execute a pipeline string and return structured results.
snapshot: Optional[Dict[str, Any]] = None
if isolate:
snapshot = self._snapshot_ctx_state()
Args:
pipeline_text: Raw pipeline text entered by the user.
on_log: Optional callback that receives human-readable log lines.
"""
normalized = pipeline_text.strip()
normalized = str(pipeline_text or "").strip()
result = PipelineRunResult(pipeline=normalized, success=False)
if not normalized:
result.error = "Pipeline is empty"
return result
tokens = self._tokenize(normalized)
stages = self._split_stages(tokens)
if not stages:
result.error = "Pipeline contains no stages"
try:
from cli_syntax import validate_pipeline_text
syntax_error = validate_pipeline_text(normalized)
if syntax_error:
result.error = syntax_error.message
result.stderr = syntax_error.message
return result
except Exception:
pass
try:
tokens = shlex.split(normalized)
except Exception as exc:
result.error = f"Syntax error: {exc}"
result.stderr = result.error
return result
if not tokens:
result.error = "Pipeline contains no tokens"
return result
config = self._config_loader.load()
try:
set_debug(bool(config.get("debug", False)))
except Exception:
pass
try:
self._worker_manager = WorkerManagerRegistry.ensure(config)
except Exception:
self._worker_manager = None
ctx.reset()
ctx.set_current_command_text(normalized)
if seeds is not None:
try:
# Mirror CLI behavior: treat seeds as output of a virtual previous stage.
if not isinstance(seeds, list):
seeds = [seeds]
setter = getattr(ctx, "set_last_result_items_only", None)
if callable(setter):
setter(seeds)
else:
ctx.set_last_items(list(seeds))
ctx.set_last_result_items_only(list(seeds))
except Exception:
pass
stdout_buffer = io.StringIO()
stderr_buffer = io.StringIO()
piped_result: Any = None
worker_session = self._start_worker_session(normalized)
try:
with contextlib.redirect_stdout(stdout_buffer), contextlib.redirect_stderr(
stderr_buffer
):
for index, stage_tokens in enumerate(stages):
stage = self._execute_stage(
index=index,
total=len(stages),
stage_tokens=stage_tokens,
piped_input=piped_result,
on_log=on_log,
)
result.stages.append(stage)
if stage.status != "completed":
result.error = stage.error or f"Stage {stage.name} failed"
return result
if index == len(stages) - 1:
result.emitted = stage.emitted
result.result_table = stage.result_table
else:
piped_result = stage.emitted
result.success = True
return result
with capture_rich_output(stdout=stdout_buffer, stderr=stderr_buffer):
with contextlib.redirect_stdout(stdout_buffer), contextlib.redirect_stderr(stderr_buffer):
if on_log:
on_log("Executing pipeline via CLI executor...")
self._executor.execute_tokens(list(tokens))
except Exception as exc:
result.error = f"{type(exc).__name__}: {exc}"
finally:
try:
ctx.clear_current_command_text()
except Exception:
pass
result.stdout = stdout_buffer.getvalue()
result.stderr = stderr_buffer.getvalue()
ctx.clear_current_command_text()
if worker_session is not None:
status = "completed" if result.success else "error"
worker_session.finish(status=status, message=result.error or "")
# ------------------------------------------------------------------
# Stage execution helpers
# ------------------------------------------------------------------
def _execute_stage(
self,
*,
index: int,
total: int,
stage_tokens: Sequence[str],
piped_input: Any,
on_log: Optional[Callable[[str], None]],
) -> PipelineStageResult:
if not stage_tokens:
return PipelineStageResult(name="(empty)", args=[], status="skipped")
cmd_name = stage_tokens[0].replace("_", "-").lower()
stage_args = stage_tokens[1:]
stage = PipelineStageResult(name=cmd_name, args=stage_args)
if cmd_name.startswith("@"):
return self._apply_selection_stage(
token=cmd_name,
stage=stage,
piped_input=piped_input,
on_log=on_log,
)
cmd_fn = REGISTRY.get(cmd_name)
if not cmd_fn:
stage.status = "failed"
stage.error = f"Unknown command: {cmd_name}"
return stage
pipeline_ctx = ctx.PipelineStageContext(stage_index=index, total_stages=total, pipe_index=index)
ctx.set_stage_context(pipeline_ctx)
# Pull the canonical state out of pipeline context.
table = None
try:
return_code = cmd_fn(piped_input, list(stage_args), self._config)
except Exception as exc: # pragma: no cover - surfaced in UI
stage.status = "failed"
stage.error = f"{type(exc).__name__}: {exc}"
if on_log:
on_log(stage.error)
return stage
finally:
ctx.set_stage_context(None)
emitted = list(getattr(pipeline_ctx, "emits", []) or [])
stage.emitted = emitted
# Capture the ResultTable if the cmdlet set one
# Check display table first (overlay), then last result table
stage.result_table = ctx.get_display_table() or ctx.get_last_result_table()
if return_code != 0:
stage.status = "failed"
stage.error = f"Exit code {return_code}"
else:
stage.status = "completed"
stage.error = None
worker_id = self._current_worker_id()
if self._worker_manager and worker_id:
label = f"[Stage {index + 1}/{total}] {cmd_name} {stage.status}"
self._worker_manager.log_step(worker_id, label)
# Don't clear the table if we just captured it, but ensure items are set for next stage
# If we have a table, we should probably keep it in ctx for history if needed
# But for pipeline execution, we mainly care about passing items to next stage
# ctx.set_last_result_table(None, emitted) <-- This was clearing it
# Ensure items are available for next stage
ctx.set_last_items(emitted)
return stage
def _apply_selection_stage(
self,
*,
token: str,
stage: PipelineStageResult,
piped_input: Any,
on_log: Optional[Callable[[str], None]],
) -> PipelineStageResult:
# Bare '@' means use the subject associated with the current result table (e.g., the file shown in a tag/URL view)
if token == "@":
subject = ctx.get_last_result_subject()
if subject is None:
stage.status = "failed"
stage.error = "Selection requested (@) but there is no current result context"
return stage
stage.emitted = subject if isinstance(subject, list) else [subject]
ctx.set_last_items(stage.emitted)
stage.status = "completed"
if on_log:
on_log("Selected current table subject via @")
return stage
selection = self._parse_selection(token)
items = piped_input or []
if not isinstance(items, list):
items = list(items if isinstance(items, Sequence) else [items])
if not items:
stage.status = "failed"
stage.error = "Selection requested but there is no upstream data"
return stage
if selection is None:
stage.emitted = list(items)
else:
zero_based = sorted(i - 1 for i in selection if i > 0)
stage.emitted = [items[i] for i in zero_based if 0 <= i < len(items)]
if not stage.emitted:
stage.status = "failed"
stage.error = "Selection matched no rows"
return stage
ctx.set_last_items(stage.emitted)
ctx.set_last_result_table(None, stage.emitted)
stage.status = "completed"
if on_log:
on_log(f"Selected {len(stage.emitted)} item(s) via {token}")
return stage
# ------------------------------------------------------------------
# Worker/session helpers
# ------------------------------------------------------------------
def _start_worker_session(self, pipeline_text: str) -> Optional[_WorkerSession]:
manager = self._ensure_worker_manager()
if manager is None:
return None
worker_id = f"tui_pipeline_{uuid.uuid4().hex[:8]}"
tracked = manager.track_worker(
worker_id,
worker_type="pipeline",
title="Pipeline run",
description=pipeline_text,
pipe=pipeline_text,
)
if not tracked:
return None
manager.log_step(worker_id, "Pipeline started")
self._config["_current_worker_id"] = worker_id
return _WorkerSession(manager=manager, worker_id=worker_id, config=self._config)
def _ensure_worker_manager(self) -> Optional[WorkerManager]:
if self._worker_manager:
return self._worker_manager
library_root = get_local_storage_path(self._config)
if not library_root:
return None
try:
self._worker_manager = WorkerManager(Path(library_root), auto_refresh_interval=0)
self._config["_worker_manager"] = self._worker_manager
table = ctx.get_display_table() or ctx.get_current_stage_table() or ctx.get_last_result_table()
except Exception:
self._worker_manager = None
return self._worker_manager
table = None
def _current_worker_id(self) -> Optional[str]:
worker_id = self._config.get("_current_worker_id")
return str(worker_id) if worker_id else None
# ------------------------------------------------------------------
# Parsing helpers
# ------------------------------------------------------------------
@staticmethod
def _tokenize(pipeline_text: str) -> List[str]:
items: List[Any] = []
try:
return shlex.split(pipeline_text)
except ValueError:
return pipeline_text.split()
items = list(ctx.get_last_result_items() or [])
except Exception:
items = []
if table is None and items:
try:
synth = ResultTable("Results")
for item in items:
synth.add_result(item)
table = synth
except Exception:
table = None
result.emitted = items
result.result_table = table
combined = (result.stdout + "\n" + result.stderr).strip().lower()
failure_markers = (
"unknown command:",
"pipeline order error:",
"invalid selection:",
"invalid pipeline syntax",
"failed to execute pipeline",
"[error]",
)
if result.error:
result.success = False
elif any(m in combined for m in failure_markers):
result.success = False
if not result.error:
result.error = "Pipeline failed"
else:
result.success = True
if isolate and snapshot is not None:
try:
self._restore_ctx_state(snapshot)
except Exception:
# Best-effort; isolation should never break normal operation.
pass
return result
@staticmethod
def _split_stages(tokens: Sequence[str]) -> List[List[str]]:
stages: List[List[str]] = []
current: List[str] = []
for token in tokens:
if token == "|":
if current:
stages.append(current)
current = []
else:
current.append(token)
if current:
stages.append(current)
return stages
def _snapshot_ctx_state() -> Dict[str, Any]:
"""Best-effort snapshot of pipeline context so TUI popups don't clobber UI state."""
@staticmethod
def _parse_selection(token: str) -> Optional[Sequence[int]]:
parsed = MedeiaCLI.parse_selection_syntax(token)
return sorted(parsed) if parsed else None
def _copy(val: Any) -> Any:
if isinstance(val, list):
return val.copy()
if isinstance(val, dict):
return val.copy()
return val
snap: Dict[str, Any] = {}
keys = [
"_LIVE_PROGRESS",
"_CURRENT_CONTEXT",
"_LAST_SEARCH_QUERY",
"_PIPELINE_REFRESHED",
"_PIPELINE_LAST_ITEMS",
"_LAST_RESULT_TABLE",
"_LAST_RESULT_ITEMS",
"_LAST_RESULT_SUBJECT",
"_RESULT_TABLE_HISTORY",
"_RESULT_TABLE_FORWARD",
"_CURRENT_STAGE_TABLE",
"_DISPLAY_ITEMS",
"_DISPLAY_TABLE",
"_DISPLAY_SUBJECT",
"_PIPELINE_LAST_SELECTION",
"_PIPELINE_COMMAND_TEXT",
"_CURRENT_CMDLET_NAME",
"_CURRENT_STAGE_TEXT",
"_PIPELINE_VALUES",
"_PENDING_PIPELINE_TAIL",
"_PENDING_PIPELINE_SOURCE",
"_UI_LIBRARY_REFRESH_CALLBACK",
]
class _WorkerSession:
"""Minimal worker session wrapper for the TUI executor."""
for k in keys:
snap[k] = _copy(getattr(ctx, k, None))
def __init__(self, *, manager: WorkerManager, worker_id: str, config: Optional[Dict[str, Any]] = None) -> None:
self._manager = manager
self.worker_id = worker_id
self._config = config
def finish(self, *, status: str, message: str) -> None:
# Deepen copies where nested lists are common.
try:
self._manager.finish_worker(self.worker_id, result=status, error_msg=message)
self._manager.log_step(self.worker_id, f"Pipeline {status}")
hist = list(getattr(ctx, "_RESULT_TABLE_HISTORY", []) or [])
snap["_RESULT_TABLE_HISTORY"] = [
(t, (items.copy() if isinstance(items, list) else list(items) if items else []), subj)
for (t, items, subj) in hist
if isinstance((t, items, subj), tuple)
]
except Exception:
pass
if self._config and self._config.get("_current_worker_id") == self.worker_id:
self._config.pop("_current_worker_id", None)
try:
fwd = list(getattr(ctx, "_RESULT_TABLE_FORWARD", []) or [])
snap["_RESULT_TABLE_FORWARD"] = [
(t, (items.copy() if isinstance(items, list) else list(items) if items else []), subj)
for (t, items, subj) in fwd
if isinstance((t, items, subj), tuple)
]
except Exception:
pass
try:
tail = list(getattr(ctx, "_PENDING_PIPELINE_TAIL", []) or [])
snap["_PENDING_PIPELINE_TAIL"] = [list(stage) for stage in tail if isinstance(stage, list)]
except Exception:
pass
try:
values = getattr(ctx, "_PIPELINE_VALUES", None)
if isinstance(values, dict):
snap["_PIPELINE_VALUES"] = values.copy()
except Exception:
pass
return snap
@staticmethod
def _restore_ctx_state(snapshot: Dict[str, Any]) -> None:
for k, v in (snapshot or {}).items():
try:
setattr(ctx, k, v)
except Exception:
pass