from pathlib import Path from typing import Dict import shutil from PIL import Image import glob import os from sorghum_pipeline.pipeline import SorghumPipeline from sorghum_pipeline.config import Config, Paths def run_pipeline_on_image(input_image_path: str, work_dir: str, save_artifacts: bool = True) -> Dict[str, str]: """ Run sorghum pipeline on a single image (no instance segmentation). Returns dict[label -> image_path] for gallery display. """ work = Path(work_dir) work.mkdir(parents=True, exist_ok=True) # Use input path directly (already in work_dir from app.py) input_path = Path(input_image_path) # Ensure demo env vars are set before pipeline construction os.environ['MINIMAL_DEMO'] = '1' os.environ['FAST_OUTPUT'] = '1' # Build in-memory config pointing input/output to the working directory cfg = Config() cfg.paths = Paths( input_folder=str(work), output_folder=str(work), boundingbox_dir=str(work) ) pipeline = SorghumPipeline(config=cfg) # Run the pipeline (single image minimal demo) results = pipeline.run(single_image_path=str(input_path)) # Collect outputs outputs: Dict[str, str] = {} # Collect desired vegetation indices (replace ARI with SAVI) wanted = [ work / 'Vegetation_indices_images/ndvi.png', work / 'Vegetation_indices_images/gndvi.png', work / 'Vegetation_indices_images/savi.png', ] labels = [ 'NDVI', 'GNDVI', 'SAVI', ] for label, path in zip(labels, wanted): if path.exists(): outputs[label] = str(path) # Also include overlay and mask if present overlay_path = work / 'results/overlay.png' mask_path = work / 'results/mask.png' composite_path = work / 'results/composite.png' if overlay_path.exists(): outputs['Overlay'] = str(overlay_path) if mask_path.exists(): outputs['Mask'] = str(mask_path) if composite_path.exists(): outputs['Composite'] = str(composite_path) # Extract simple stats for display if present in pipeline results try: plants = results.get('plants', {}) if isinstance(results, dict) else {} if plants: _, pdata = next(iter(plants.items())) veg = pdata.get('vegetation_indices', {}) stats_lines = [] for name in ['NDVI', 'GNDVI', 'SAVI']: entry = veg.get(name, {}) st = entry.get('statistics', {}) if isinstance(entry, dict) else {} if st: stats_lines.append(f"{name}: mean={st.get('mean', 0):.3f}, std={st.get('std', 0):.3f}") if stats_lines: outputs['StatsText'] = "\n".join(stats_lines) except Exception: pass return outputs