File size: 3,193 Bytes
61b3c2b dd1d7f5 61b3c2b dd1d7f5 61b3c2b 7613956 61b3c2b e768711 dd1d7f5 61b3c2b dd1d7f5 61b3c2b dd1d7f5 7613956 61b3c2b 31ddfa7 61b3c2b c170961 dd1d7f5 c170961 dd1d7f5 c170961 dd1d7f5 61b3c2b e768711 5f6c42c e768711 5f6c42c e768711 c170961 e768711 61b3c2b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
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] = {}
try:
# Log immediate output directory contents for debugging
for sub in ['results', 'Vegetation_indices_images', 'texture_output']:
p = work / sub
if p.exists():
files = sorted([str(x.name) for x in p.iterdir() if x.is_file()])
print(f"Artifacts in {sub}: {files}")
except Exception:
pass
# 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
|