Spaces:
Sleeping
Sleeping
import streamlit as st
import sys
import os
from datetime import datetime
from PIL import Image
import json
import tempfile
import time
import re
import hashlib
import platform
# Add utils to path
utils_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'utils')
if utils_path not in sys.path:
sys.path.append(utils_path)
# Import utilities with error handling and robust fallbacks
UPLOAD_IMPORT_ERROR = None
try:
from utils.ocr_utils import extract_data_from_image
from utils.parsing_utils import _parse_raw_text_to_structured_json
from utils.analysis_engine import validate_soil_data, validate_leaf_data
from utils.parameter_standardizer import parameter_standardizer
from utils.config_manager import get_ui_config
except Exception:
try:
from ocr_utils import extract_data_from_image, validate_soil_data, validate_leaf_data
from parsing_utils import _parse_raw_text_to_structured_json
from parameter_standardizer import parameter_standardizer
from config_manager import get_ui_config
except Exception as e:
UPLOAD_IMPORT_ERROR = str(e)
extract_data_from_image = None
_parse_raw_text_to_structured_json = None
validate_soil_data = None
validate_leaf_data = None
parameter_standardizer = None
def get_ui_config():
return None
# Import translations
try:
from utils.translations import t
except ImportError:
try:
from translations import t
except ImportError:
# Fallback translation function
def t(key, default=None):
return default or key
def get_machine_id():
"""Generate a persistent machine identifier that persists across sessions"""
if 'machine_id' not in st.session_state:
machine_id_file = os.path.join(os.path.expanduser('~'), '.cropdrive_ai_machine_id')
machine_id = None
# Try to read existing machine ID from file
try:
if os.path.exists(machine_id_file):
with open(machine_id_file, 'r') as f:
machine_id = f.read().strip()
if machine_id and len(machine_id) == 16:
# Valid existing ID
st.session_state.machine_id = machine_id
return machine_id
except Exception:
pass
# Generate new machine ID based on system information
machine_info = {
'platform': platform.platform(),
'system': platform.system(),
'processor': platform.processor(),
'node': platform.node(), # Network name
}
# Create a hash from machine info
machine_string = json.dumps(machine_info, sort_keys=True)
machine_id = hashlib.md5(machine_string.encode()).hexdigest()[:16] # 16 char ID
# Save to file for persistence
try:
with open(machine_id_file, 'w') as f:
f.write(machine_id)
except Exception:
# If file write fails, still use the generated ID
pass
# Store in session state
st.session_state.machine_id = machine_id
return st.session_state.machine_id
def show_upload_page():
"""Main upload page - focused only on file upload and preview"""
# Import translations
try:
from utils.translations import t
except ImportError:
from translations import t
if UPLOAD_IMPORT_ERROR:
st.error(f"Upload module failed to load required dependencies: {UPLOAD_IMPORT_ERROR}")
st.info("Please check Space build logs and requirements.txt, then restart the Space.")
return
# CRITICAL: Check upload limit BEFORE showing upload page
try:
from utils.cropdrive_integration import can_start_analysis, get_upload_limit_info
# Check if user can start analysis
if not can_start_analysis():
# Upload limit reached - show message and redirect to history
limit_info = get_upload_limit_info()
uploads_used = limit_info.get('uploads_used', 0)
uploads_limit = limit_info.get('uploads_limit', 0)
upload_limit_exceeded = limit_info.get('upload_limit_exceeded', False)
st.error("🚫 **Upload Limit Reached**")
if upload_limit_exceeded or (uploads_limit > 0 and uploads_used >= uploads_limit):
st.warning(f"⚠️ You have reached your upload limit: **{uploads_used}/{uploads_limit}** analyses used.")
st.info("💡 **What you can do:**")
st.markdown("""
- 📋 **View your past analyses** - Access your analysis history to review previous reports
- 🔄 **Upgrade your plan** - Contact support to increase your analysis limit
- 📊 **Review existing reports** - All your past analyses are still accessible
""")
# Show button to view history (if analysisId is available)
st.markdown("---")
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.info("📋 **To view your analysis history, go to the CropDrive website and click 'Open Recent Analysis' on any report.**")
if st.button("🏠 Back to Home", width="stretch", type="primary"):
st.session_state.current_page = 'home'
st.rerun()
# Stop execution - don't show upload page
st.stop()
except ImportError:
# If cropdrive_integration is not available, allow upload (fallback)
pass
except Exception as e:
# Log error but allow upload (fallback)
import logging
logging.warning(f"Error checking upload limit: {e}")
st.markdown(f'<h1 style="color: #2E8B57; text-align: center; font-size: 3rem; font-weight: 700; margin: 1.5rem 0 1rem 0;">📤 {t("upload_title")}</h1>', unsafe_allow_html=True)
st.markdown(f"### {t('upload_desc')}")
# Main upload section
upload_section()
def display_structured_soil_data(soil_data: dict) -> None:
"""Display structured soil data in a clean, organized format"""
samples = soil_data.get('samples', [])
if not samples:
st.warning("No soil samples found in extracted data")
return
# Display sample count
st.success(f"✅ **{len(samples)} soil samples extracted successfully**")
# Create tabs for different views
tab1, tab2, tab3 = st.tabs(["📊 Structured Data", "🔍 Sample Details", "📋 Raw JSON"])
with tab1:
st.markdown("**📊 Soil Analysis Summary**")
# Create a summary table
if samples:
# Get all unique parameters from the data structure
all_params = set()
for sample in samples:
# Handle both flat and nested structures
if 'data' in sample:
all_params.update(sample['data'].keys())
all_params.update(sample.keys())
# Display parameter summary
st.info(f"**Parameters detected:** {', '.join(sorted(all_params))}")
# Display first few samples in a table format
display_samples = samples[:5] # Show first 5 samples
for i, sample in enumerate(display_samples, 1):
sample_id = sample.get('sample_id', sample.get('Lab No.', f'Sample {i}'))
with st.expander(f"🧪 Sample {i} - {sample_id}", expanded=i==1):
col1, col2 = st.columns(2)
# Get data from nested structure if it exists
sample_data = sample.get('data', {}) if 'data' in sample else sample
with col1:
st.markdown("**Basic Parameters:**")
# Map actual parameter names to display names
basic_mapping = {
'sample_id': 'Sample ID',
'Lab No.': 'Lab Number',
'pH': 'pH',
'Org. C (%)': 'Organic Carbon (%)',
'N (%)': 'Nitrogen (%)',
'Total N (%)': 'Total Nitrogen (%)'
}
# Show sample ID first
if 'sample_id' in sample:
st.write(f"• **Sample ID:** {sample['sample_id']}")
# Show other basic parameters
for param, display_name in basic_mapping.items():
if param in sample_data:
st.write(f"• **{display_name}:** {sample_data[param]}")
with col2:
st.markdown("**Nutrient Parameters:**")
# Use standardized parameter names from parameter_standardizer
nutrient_mapping = parameter_standardizer.get_display_name_mapping('soil')
# Filter to show only nutrient parameters (exclude pH, N, Org.C)
nutrient_params = {
'Total P (mg/kg)': 'Total P (mg/kg)',
'Avail P (mg/kg)': 'Available P (mg/kg)',
'Exch. K (meq%)': 'Exchangeable K (meq%)',
'Exch. Ca (meq%)': 'Exchangeable Ca (meq%)',
'Exch. Mg (meq%)': 'Exchangeable Mg (meq%)',
'CEC (meq%)': 'CEC (meq%)'
}
for param, display_name in nutrient_params.items():
if param in sample_data:
st.write(f"• **{display_name}:** {sample_data[param]}")
if len(samples) > 5:
st.info(f"Showing first 5 samples. Total samples: {len(samples)}")
with tab2:
st.markdown("**🔍 Detailed Sample Analysis**")
# Sample selector
sample_options = [f"Sample {i+1}: {sample.get('sample_id', sample.get('Lab No.', f'Sample {i+1}'))}"
for i, sample in enumerate(samples)]
selected_sample_idx = st.selectbox("Select sample to view:",
range(len(sample_options)),
format_func=lambda x: sample_options[x])
if selected_sample_idx is not None:
selected_sample = samples[selected_sample_idx]
sample_data = selected_sample.get('data', {}) if 'data' in selected_sample else selected_sample
st.markdown(f"**Sample Details: {sample_options[selected_sample_idx]}**")
# Group parameters by category with actual parameter names
categories = {
'Basic Info': {
'sample_id': 'Sample ID',
'Lab No.': 'Lab Number'
},
'Soil Chemistry': {
'pH': 'pH',
'Org. C (%)': 'Organic Carbon (%)',
'N (%)': 'Nitrogen (%)',
'Total N (%)': 'Total Nitrogen (%)'
},
'Phosphorus': {
'Avail P (mg/kg)': 'Available P (mg/kg)',
'Total P (mg/kg)': 'Total P (mg/kg)'
},
'Exchangeable Cations': {
'Exch. K (meq%)': 'Exchangeable K (meq%)',
'Exch. Ca (meq%)': 'Exchangeable Ca (meq%)',
'Exch. Mg (meq%)': 'Exchangeable Mg (meq%)'
},
'Other': {
'CEC (meq%)': 'CEC (meq%)'
}
}
for category, param_mapping in categories.items():
category_data = {}
# Check both sample and sample_data for parameters
for param, display_name in param_mapping.items():
if param in selected_sample:
category_data[display_name] = selected_sample[param]
elif param in sample_data:
category_data[display_name] = sample_data[param]
if category_data:
st.markdown(f"**{category}:**")
for display_name, value in category_data.items():
st.write(f" • {display_name}: {value}")
st.write("")
with tab3:
st.markdown("**📋 Complete Soil Data (JSON)**")
st.code(json.dumps(soil_data, indent=2, ensure_ascii=False), language='json')
def display_structured_leaf_data(leaf_data: dict) -> None:
"""Display structured leaf data in a clean, organized format"""
samples = leaf_data.get('samples', [])
if not samples:
st.warning("No leaf samples found in extracted data")
return
# Display sample count
st.success(f"✅ **{len(samples)} leaf samples extracted successfully**")
# Create tabs for different views
tab1, tab2, tab3 = st.tabs(["📊 Structured Data", "🔍 Sample Details", "📋 Raw JSON"])
with tab1:
st.markdown("**📊 Leaf Analysis Summary**")
if samples:
# Display first few samples in organized format
display_samples = samples[:5] # Show first 5 samples
for i, sample in enumerate(display_samples, 1):
sample_id = sample.get('sample_id', sample.get('Lab No.', f'Sample {i}'))
with st.expander(f"🍃 Sample {i} - {sample_id}", expanded=i==1):
col1, col2 = st.columns(2)
with col1:
st.markdown("**% Dry Matter Content:**")
dry_matter = sample.get('% Dry Matter', {})
for nutrient, value in dry_matter.items():
st.write(f"• **{nutrient}:** {value}%")
# Basic info
basic_info = {k: v for k, v in sample.items()
if k not in ['% Dry Matter', 'mg/kg Dry Matter']}
if basic_info:
st.markdown("**Sample Info:**")
for key, value in basic_info.items():
st.write(f"• **{key}:** {value}")
with col2:
st.markdown("**mg/kg Dry Matter Content:**")
mg_kg = sample.get('mg/kg Dry Matter', {})
for nutrient, value in mg_kg.items():
st.write(f"• **{nutrient}:** {value} mg/kg")
if len(samples) > 5:
st.info(f"Showing first 5 samples. Total samples: {len(samples)}")
with tab2:
st.markdown("**🔍 Detailed Sample Analysis**")
# Sample selector
sample_options = [f"Sample {i+1}: {sample.get('sample_id', sample.get('Lab No.', f'Sample {i+1}'))}"
for i, sample in enumerate(samples)]
selected_sample_idx = st.selectbox("Select sample to view:",
range(len(sample_options)),
format_func=lambda x: sample_options[x])
if selected_sample_idx is not None:
selected_sample = samples[selected_sample_idx]
st.markdown(f"**Sample Details: {sample_options[selected_sample_idx]}**")
# Display organized data
col1, col2, col3 = st.columns(3)
with col1:
st.markdown("**Major Nutrients (% Dry Matter):**")
dry_matter = selected_sample.get('% Dry Matter', {})
for nutrient in ['N', 'P', 'K', 'Mg', 'Ca']:
if nutrient in dry_matter:
st.write(f"• **{nutrient}:** {dry_matter[nutrient]}%")
with col2:
st.markdown("**Micronutrients (mg/kg):**")
mg_kg = selected_sample.get('mg/kg Dry Matter', {})
for nutrient in ['B', 'Cu', 'Zn', 'Fe', 'Mn']:
if nutrient in mg_kg:
st.write(f"• **{nutrient}:** {mg_kg[nutrient]} mg/kg")
with col3:
st.markdown("**Sample Information:**")
basic_info = {k: v for k, v in selected_sample.items()
if k not in ['% Dry Matter', 'mg/kg Dry Matter']}
for key, value in basic_info.items():
st.write(f"• **{key}:** {value}")
with tab3:
st.markdown("**📋 Complete Leaf Data (JSON)**")
st.code(json.dumps(leaf_data, indent=2, ensure_ascii=False), language='json')
def show_ocr_preview(file, file_type: str, container_type: str) -> None:
"""Enhanced OCR preview with step-by-step processing display"""
# Header section (removed expander to avoid nesting issues)
st.markdown(f"### 🔍 {t('upload_ocr_preview', 'OCR Data Processing & Preview')}")
# Processing status indicators
status_container = st.container()
# Add refresh button and timestamp
col_refresh, col_timestamp = st.columns([1, 3])
with col_refresh:
refresh_ocr = st.button("🔄 Refresh OCR",
key=f"refresh_{container_type}_ocr",
help="Re-process the image with OCR")
with col_timestamp:
st.caption(f"{t('upload_last_processed', 'Last processed')}: {datetime.now().strftime('%H:%M:%S')}")
# Perform OCR processing quietly without step indicators
try:
# Create temporary file
file_ext = os.path.splitext(file.name)[1].lower()
is_image = file_ext in ['.png', '.jpg', '.jpeg']
with tempfile.NamedTemporaryFile(delete=False, suffix=file_ext) as tmp_file:
if is_image:
image = Image.open(file)
image.save(tmp_file.name)
else:
tmp_file.write(file.getvalue())
tmp_file_path = tmp_file.name
# Perform OCR extraction
ocr_result = extract_data_from_image(tmp_file_path)
# Clean up temporary file
try:
time.sleep(0.1)
os.unlink(tmp_file_path)
except (PermissionError, FileNotFoundError):
pass
success = ocr_result.get('success', False)
if success:
tables = ocr_result.get('tables', [])
if tables:
detected_type = tables[0].get('type', 'unknown')
# Check content type match
if detected_type.lower() == container_type.lower():
samples = tables[0].get('samples', [])
# Check if samples are empty objects
samples_are_empty = samples and all(isinstance(s, dict) and not s for s in samples)
# Store raw text in session state for fallback analysis
raw_text = ocr_result.get('raw_data', {}).get('text', '')
if container_type == 'soil':
st.session_state.raw_soil_text = raw_text
else:
st.session_state.raw_leaf_text = raw_text
st.info("🔄 Processing data through OCR text analysis for consistent results...")
# For Excel files, we need to generate raw text from the extracted samples
# to ensure same processing as images/PDFs
file_ext = os.path.splitext(file.name)[1].lower()
if file_ext in ['.xlsx', '.xls']:
if samples and not samples_are_empty:
# Generate raw text from samples for Excel files to match OCR processing
# Format: "S001: pH: 5.26 N (%): 0.07 Org. C (%): 0.58 ..."
raw_text_lines = []
for i, sample in enumerate(samples, 1):
sample_id = sample.get('sample_id', sample.get('Sample No.', sample.get('Lab No.', f'S{i:03d}')))
# Ensure sample_id is a string
sample_id = str(sample_id) if sample_id is not None else f'S{i:03d}'
# For leaf data, use 'L' prefix instead of 'S'
if container_type == 'leaf':
# Remove 'L' or 'P' prefix if it exists and ensure 3-digit format
if sample_id.startswith(('L', 'P')) and sample_id[1:].isdigit():
sample_num = int(sample_id[1:])
sample_id = f'L{sample_num:03d}'
elif sample_id.isdigit():
sample_id = f'L{int(sample_id):03d}'
else:
sample_id = f'L{i:03d}'
else:
# For soil data, use 'S' prefix
if sample_id.startswith('S') and sample_id[1:].isdigit():
sample_num = int(sample_id[1:])
sample_id = f'S{sample_num:03d}'
elif sample_id.isdigit():
sample_id = f'S{int(sample_id):03d}'
else:
sample_id = f'S{i:03d}'
param_values = []
for key, value in sample.items():
if key not in ['sample_id', 'Sample No.', 'Lab No.'] and value is not None:
# Format value as string
if isinstance(value, (int, float)):
param_values.append(str(value))
else:
param_values.append(str(value))
if param_values:
# Join all parameter values with spaces to match OCR format
values_text = ' '.join(param_values)
raw_text_lines.append(f"{sample_id}: {values_text}")
if raw_text_lines:
raw_text = "\n".join(raw_text_lines)
# Update session state with generated raw text
if container_type == 'soil':
st.session_state.raw_soil_text = raw_text
else:
st.session_state.raw_leaf_text = raw_text
else:
st.warning(f"No parameter data found in {container_type} Excel file samples")
else:
st.warning(f"No valid samples found in {container_type} Excel file (samples: {len(samples) if samples else 0}, empty: {samples_are_empty if 'samples_are_empty' in locals() else 'unknown'})")
# Fallback to raw text parsing (same path as images/PDFs)
if not raw_text and ocr_result.get('text'):
raw_text = ocr_result['text']
if raw_text:
# Store raw text in session state for analysis
if container_type == 'soil':
st.session_state.raw_soil_text = raw_text
else:
st.session_state.raw_leaf_text = raw_text
# For Excel files with successfully extracted samples, use them directly
# instead of re-parsing to preserve all samples
if file_ext in ['.xlsx', '.xls'] and samples and not samples_are_empty:
# Convert extracted samples to structured format directly
structured_data = {}
samples_data = {}
for sample in samples:
sid, sample_data = _flatten_ocr_sample(sample, container_type)
if not sid or not sample_data:
continue
sample_id = sid
# Preserve SP Lab IDs (S218/25, P220/25); normalize farm IDs only
if container_type == 'leaf':
if re.match(r'^P\d{1,4}/\d{2}$', sample_id, re.I):
sample_id = sample_id[0].upper() + sample_id[1:]
elif sample_id.startswith(('L', 'P')) and sample_id[1:].isdigit():
sample_id = f'L{int(sample_id[1:]):03d}'
elif sample_id.isdigit():
sample_id = f'L{int(sample_id):03d}'
else:
if re.match(r'^S\d{1,4}/\d{2}$', sample_id, re.I):
sample_id = sample_id[0].upper() + sample_id[1:]
elif sample_id.startswith('S') and sample_id[1:].isdigit():
sample_id = f'S{int(sample_id[1:]):03d}'
elif sample_id.isdigit():
sample_id = f'S{int(sample_id):03d}'
samples_data[sample_id] = sample_data
# Determine container key (SP Lab if any ID has / or SP/LAB markers)
if container_type == 'soil':
if any('/' in str(sid) or 'SP' in str(sid).upper() or 'LAB' in str(sid).upper() for sid in samples_data.keys()):
container_key = "SP_Lab_Test_Report"
else:
container_key = "Farm_Soil_Test_Data"
else:
if any(('/' in str(sid) and str(sid).upper().startswith('P')) for sid in samples_data.keys()):
container_key = "SP_Lab_Test_Report"
else:
container_key = "Farm_Leaf_Test_Data"
structured_data[container_key] = samples_data
# Display structured data directly
st.markdown(f"### 📝 {t('upload_raw_text', 'Raw Extracted Text Data')}")
st.markdown(f"#### 📊 {t('upload_structured_data', 'Structured OCR Data (JSON Format)')}")
st.markdown(f"**✅ {t('upload_extracted_samples', 'Extracted {} samples directly from {} file').format(len(samples_data), container_type)}**")
try:
import json
formatted_json = json.dumps(structured_data, indent=2, ensure_ascii=False)
st.code(formatted_json, language="json")
except Exception as e:
st.json(structured_data)
# Store structured data in session state
if container_type == 'soil':
st.session_state.structured_soil_data = structured_data
else:
st.session_state.structured_leaf_data = structured_data
st.info("💡 **AI Analysis Ready**: The structured data above will be used for comprehensive step-by-step analysis.")
# Show raw text for reference
st.markdown("#### 🔍 Raw OCR Text (Reference Only)")
st.code(raw_text, language="text")
st.caption(f"Raw text length: {len(raw_text)} characters | Container: {container_type}")
else:
# For images/PDFs or when Excel extraction failed, use raw text parsing
_show_raw_text_as_json(raw_text, container_type, ocr_result)
else:
st.error("No data could be extracted from the uploaded file")
else:
st.error(f"**Content Type Mismatch Detected!**")
st.warning(f"🔍 **Detected:** {detected_type.title()} analysis report")
st.warning(f"📁 **Expected:** {container_type.title()} analysis report")
# Try to fall back to raw text parsing even for type mismatch
raw_text = ocr_result.get('raw_data', {}).get('text', '')
if not raw_text and ocr_result.get('text'):
raw_text = ocr_result['text']
# For Excel files with type mismatch, generate raw text from samples
file_ext = os.path.splitext(file.name)[1].lower()
if (file_ext in ['.xlsx', '.xls'] and 'samples' in locals() and samples and
'samples_are_empty' in locals() and not samples_are_empty and not raw_text):
raw_text_lines = []
for i, sample in enumerate(samples, 1):
sample_id = sample.get('sample_id', sample.get('Sample No.', sample.get('Lab No.', f'S{i:03d}')))
# Ensure sample_id is a string
sample_id = str(sample_id) if sample_id is not None else f'S{i:03d}'
# For leaf data, use 'L' prefix instead of 'S'
if container_type == 'leaf':
# Remove 'L' or 'P' prefix if it exists and ensure 3-digit format
if sample_id.startswith(('L', 'P')) and sample_id[1:].isdigit():
sample_num = int(sample_id[1:])
sample_id = f'L{sample_num:03d}'
elif sample_id.isdigit():
sample_id = f'L{int(sample_id):03d}'
else:
sample_id = f'L{i:03d}'
else:
# For soil data, use 'S' prefix
if sample_id.startswith('S') and sample_id[1:].isdigit():
sample_num = int(sample_id[1:])
sample_id = f'S{sample_num:03d}'
elif sample_id.isdigit():
sample_id = f'S{int(sample_id):03d}'
else:
sample_id = f'S{i:03d}'
param_values = []
for key, value in sample.items():
if key not in ['sample_id', 'Sample No.', 'Lab No.'] and value is not None:
# Format value as string
if isinstance(value, (int, float)):
param_values.append(str(value))
else:
param_values.append(str(value))
if param_values:
# Join all parameter values with spaces to match OCR format
values_text = ' '.join(param_values)
raw_text_lines.append(f"{sample_id}: {values_text}")
if raw_text_lines:
raw_text = "\n".join(raw_text_lines)
if raw_text:
# Store raw text in session state for fallback analysis
if container_type == 'soil':
st.session_state.raw_soil_text = raw_text
else:
st.session_state.raw_leaf_text = raw_text
st.info("🔄 **Attempting raw text parsing despite type mismatch...**")
_show_raw_text_as_json(raw_text, container_type, ocr_result)
else:
if container_type.lower() == 'soil':
st.info("💡 **Please upload a soil analysis report in this container.**")
st.info("🍃 **For leaf analysis, use the Leaf Analysis container on the right.**")
else:
st.info("💡 **Please upload a leaf analysis report in this container.**")
st.info("🌱 **For soil analysis, use the Soil Analysis container on the left.**")
else:
# Try to get raw text for fallback processing
raw_text = ocr_result.get('raw_data', {}).get('text', '')
if not raw_text and ocr_result.get('text'):
raw_text = ocr_result['text']
if raw_text:
# Store raw text in session state for fallback analysis
if container_type == 'soil':
st.session_state.raw_soil_text = raw_text
else:
st.session_state.raw_leaf_text = raw_text
_show_raw_text_as_json(raw_text, container_type, ocr_result)
else:
st.error("No data could be extracted from the uploaded file")
except Exception as e:
# Provide better error messages
if 'ocr_result' in locals() and ocr_result:
error_msg = ocr_result.get('error', str(e)) if ocr_result.get('error') else str(e)
else:
error_msg = str(e)
# Add context about what went wrong
if not error_msg or error_msg == 'None':
error_msg = f"Processing failed for {file_ext.upper()} file. Check file format and try again."
st.error(f"**OCR Error:** {error_msg}")
# Show error details (using markdown instead of expander to avoid nesting)
st.markdown("**❌ Error Details:**")
st.code(f"Exception: {str(e)}\nError Message: {error_msg}\nFile: {file.name}\nContainer: {container_type}", language="text")
def upload_section():
"""Handle file upload and preview with enhanced OCR processing"""
# Get UI configuration
ui_config = get_ui_config()
# Import translations
try:
from utils.translations import t
except ImportError:
from translations import t
st.markdown(f"### 📁 {t('upload_section_title')}")
st.info(f"💡 **{t('upload_tip')}**")
# Create separate containers for soil and leaf analysis
col1, col2 = st.columns(2)
# Initialize session state for uploaded files
if 'soil_file' not in st.session_state:
st.session_state.soil_file = None
if 'leaf_file' not in st.session_state:
st.session_state.leaf_file = None
if 'land_yield_data' not in st.session_state:
st.session_state.land_yield_data = {
'land_size': 0,
'land_unit': 'hectares',
'current_yield': 0,
'yield_unit': 'tonnes/hectare',
'palm_density': 148,
'palm_age': 0,
'soil_type': ''
}
# Load saved land & yield data from Firestore using machine ID
try:
from utils.firebase_config import get_firestore_client
db = get_firestore_client()
if db:
machine_id = get_machine_id()
# Use a separate collection for land/yield data
land_yield_ref = db.collection('land_yield_data').document(machine_id)
land_yield_doc = land_yield_ref.get()
if land_yield_doc.exists:
saved_data = land_yield_doc.to_dict()
if (st.session_state.land_yield_data['land_size'] == 0 and
st.session_state.land_yield_data['current_yield'] == 0):
st.session_state.land_yield_data.update({
'land_size': saved_data.get('land_size', 0),
'land_unit': saved_data.get('land_unit', 'hectares'),
'current_yield': saved_data.get('current_yield', 0),
'yield_unit': saved_data.get('yield_unit', 'tonnes/hectare'),
'palm_density': saved_data.get('palm_density', 148),
'palm_age': saved_data.get('palm_age', 0),
'soil_type': saved_data.get('soil_type', '')
})
except Exception:
pass
with col1:
with st.container():
st.markdown(f"#### 🌱 {t('upload_soil_title')}")
st.markdown(t('upload_soil_desc'))
st.info(f"📋 **{t('upload_soil_expected')}**")
soil_file = st.file_uploader(
t('upload_soil_file'),
type=['png', 'jpg', 'jpeg', 'pdf', 'csv', 'xlsx', 'xls'],
help=t('upload_soil_help'),
key="soil_uploader"
)
# Handle file upload - persist in session state
if soil_file is not None:
# Store file in session state and also store file data for persistence
st.session_state.soil_file = soil_file
# Store file bytes for persistence across reruns (Streamlit Cloud compatibility)
try:
st.session_state.soil_file_data = soil_file.getvalue()
st.session_state.soil_file_name = soil_file.name
st.session_state.soil_file_type = soil_file.type
except Exception as e:
st.warning(f"Could not cache soil file data: {e}")
st.markdown("##### 📄 Uploaded Soil Report")
file_ext = os.path.splitext(soil_file.name)[1].lower()
is_image = file_ext in ['.png', '.jpg', '.jpeg']
if is_image:
try:
soil_image = Image.open(soil_file)
st.image(soil_image, caption="Soil Analysis Report", width="stretch")
st.info(f"**File:** {soil_file.name} | **Size:** {soil_file.size:,} bytes | **Format:** {soil_image.format}")
except Exception as e:
st.error(f"Error displaying soil image: {e}")
st.info(f"**File:** {soil_file.name} | **Size:** {soil_file.size:,} bytes")
else:
st.info(f"**File:** {soil_file.name} | **Size:** {soil_file.size:,} bytes | **Type:** {file_ext}")
# Enhanced OCR preview
show_ocr_preview(soil_file, file_ext, 'soil')
elif st.session_state.get('soil_file') is not None:
# File exists in session state but uploader shows None (after rerun)
# Reconstruct file object from cached data if available
if 'soil_file_data' in st.session_state and 'soil_file_name' in st.session_state:
try:
from io import BytesIO
# Recreate file-like object from cached data
cached_file = BytesIO(st.session_state.soil_file_data)
cached_file.name = st.session_state.soil_file_name
cached_file.type = st.session_state.get('soil_file_type', 'application/octet-stream')
st.session_state.soil_file = cached_file
st.markdown("##### 📄 Uploaded Soil Report (Restored)")
file_ext = os.path.splitext(st.session_state.soil_file_name)[1].lower()
is_image = file_ext in ['.png', '.jpg', '.jpeg']
if is_image:
try:
soil_image = Image.open(cached_file)
st.image(soil_image, caption="Soil Analysis Report", width="stretch")
st.info(f"**File:** {st.session_state.soil_file_name} | **Size:** {len(st.session_state.soil_file_data):,} bytes")
except Exception as e:
st.info(f"**File:** {st.session_state.soil_file_name} | **Size:** {len(st.session_state.soil_file_data):,} bytes")
else:
st.info(f"**File:** {st.session_state.soil_file_name} | **Size:** {len(st.session_state.soil_file_data):,} bytes | **Type:** {file_ext}")
except Exception as e:
st.warning(f"Could not restore soil file: {e}")
# Clear invalid cached data
st.session_state.soil_file = None
else:
# Show that file was previously uploaded
st.info("✅ Soil file previously uploaded")
with col2:
with st.container():
st.markdown(f"#### 🍃 {t('upload_leaf_title')}")
st.markdown(t('upload_leaf_desc'))
st.info(f"📋 **{t('upload_expected', 'Expected')}:** {t('upload_leaf_expected', 'Leaf analysis with N%, P%, K%, Mg%, Ca%, B, Cu, Zn content, etc.')}")
leaf_file = st.file_uploader(
t('upload_leaf_file'),
type=['png', 'jpg', 'jpeg', 'pdf', 'csv', 'xlsx', 'xls'],
help=t('upload_leaf_help'),
key="leaf_uploader"
)
# Handle file upload - persist in session state
if leaf_file is not None:
# Store file in session state and also store file data for persistence
st.session_state.leaf_file = leaf_file
# Store file bytes for persistence across reruns (Streamlit Cloud compatibility)
try:
st.session_state.leaf_file_data = leaf_file.getvalue()
st.session_state.leaf_file_name = leaf_file.name
st.session_state.leaf_file_type = leaf_file.type
except Exception as e:
st.warning(f"Could not cache leaf file data: {e}")
st.markdown("##### 📄 Uploaded Leaf Report")
leaf_ext = os.path.splitext(leaf_file.name)[1].lower()
leaf_is_image = leaf_ext in ['.png', '.jpg', '.jpeg']
if leaf_is_image:
try:
leaf_image = Image.open(leaf_file)
st.image(leaf_image, caption="Leaf Analysis Report", width="stretch")
st.info(f"**File:** {leaf_file.name} | **Size:** {leaf_file.size:,} bytes | **Format:** {leaf_image.format}")
except Exception as e:
st.error(f"Error displaying leaf image: {e}")
st.info(f"**File:** {leaf_file.name} | **Size:** {leaf_file.size:,} bytes")
else:
st.info(f"**File:** {leaf_file.name} | **Size:** {leaf_file.size:,} bytes | **Type:** {leaf_ext}")
# Enhanced OCR preview
show_ocr_preview(leaf_file, leaf_ext, 'leaf')
elif st.session_state.get('leaf_file') is not None:
# File exists in session state but uploader shows None (after rerun)
# Reconstruct file object from cached data if available
if 'leaf_file_data' in st.session_state and 'leaf_file_name' in st.session_state:
try:
from io import BytesIO
# Recreate file-like object from cached data
cached_file = BytesIO(st.session_state.leaf_file_data)
cached_file.name = st.session_state.leaf_file_name
cached_file.type = st.session_state.get('leaf_file_type', 'application/octet-stream')
st.session_state.leaf_file = cached_file
st.markdown("##### 📄 Uploaded Leaf Report (Restored)")
leaf_ext = os.path.splitext(st.session_state.leaf_file_name)[1].lower()
leaf_is_image = leaf_ext in ['.png', '.jpg', '.jpeg']
if leaf_is_image:
try:
leaf_image = Image.open(cached_file)
st.image(leaf_image, caption="Leaf Analysis Report", width="stretch")
st.info(f"**File:** {st.session_state.leaf_file_name} | **Size:** {len(st.session_state.leaf_file_data):,} bytes")
except Exception as e:
st.info(f"**File:** {st.session_state.leaf_file_name} | **Size:** {len(st.session_state.leaf_file_data):,} bytes")
else:
st.info(f"**File:** {st.session_state.leaf_file_name} | **Size:** {len(st.session_state.leaf_file_data):,} bytes | **Type:** {leaf_ext}")
except Exception as e:
st.warning(f"Could not restore leaf file: {e}")
# Clear invalid cached data
st.session_state.leaf_file = None
else:
# Show that file was previously uploaded
st.info("✅ Leaf file previously uploaded")
# Land/Yield Size Data Section
st.markdown("---")
st.markdown(f"### 🌾 {t('upload_land_yield_info', 'Land & Yield Information (Required)')}")
st.markdown(f"*{t('upload_land_yield_desc', 'Essential for generating accurate economic forecasts and 5-year yield projections')}*")
col1, col2, col3, col4 = st.columns(4)
with col1:
st.markdown(f"#### 📏 {t('upload_land_size', 'Land Size')}")
land_size = st.number_input(
t('upload_land_size', 'Land Size'),
min_value=0,
max_value=10000,
value=st.session_state.land_yield_data['land_size'],
step=1,
help=t('upload_land_size_help', 'Enter the total land area for analysis'),
key="land_size_input"
)
land_unit = st.selectbox(
t('upload_land_unit', 'Unit'),
options=['hectares', 'acres', 'square_meters'],
index=['hectares', 'acres', 'square_meters'].index(st.session_state.land_yield_data['land_unit']),
key="land_unit_input"
)
st.session_state.land_yield_data['land_size'] = land_size
st.session_state.land_yield_data['land_unit'] = land_unit
with col2:
st.markdown(f"#### 🌾 {t('upload_current_yield', 'Current Yield')}")
current_yield = st.number_input(
t('upload_current_yield', 'Current Yield'),
min_value=0,
max_value=1000,
value=st.session_state.land_yield_data['current_yield'],
step=1,
help=t('upload_current_yield_help', 'Enter the current yield per unit area'),
key="current_yield_input"
)
yield_unit = st.selectbox(
t('upload_yield_unit', 'Yield Unit'),
options=['tonnes/hectare', 'kg/hectare', 'tonnes/acre', 'kg/acre'],
index=['tonnes/hectare', 'kg/hectare', 'tonnes/acre', 'kg/acre'].index(st.session_state.land_yield_data['yield_unit']),
key="yield_unit_input"
)
st.session_state.land_yield_data['current_yield'] = current_yield
st.session_state.land_yield_data['yield_unit'] = yield_unit
with col3:
st.markdown(f"#### 🌴 {t('upload_palm_density', 'Palm Density')}")
palm_density = st.number_input(
t('upload_palms_per_hectare', 'Palms per Hectare'),
min_value=100,
max_value=200,
value=st.session_state.land_yield_data['palm_density'],
step=1,
help=t('upload_palm_density_help', 'Number of oil palm trees per hectare (typical: 136-148)'),
key="palm_density_input"
)
st.session_state.land_yield_data['palm_density'] = palm_density
with col4:
st.markdown(f"#### 🌱 {t('upload_palm_age', 'Palm Tree Age')}")
# Get current palm age value, ensure it's within valid range for display
current_palm_age = st.session_state.land_yield_data.get('palm_age', 0)
if current_palm_age < 0 or current_palm_age > 35:
current_palm_age = 0
palm_age = st.number_input(
t('upload_palm_age_years', 'Age (Years)'),
min_value=0,
max_value=35,
value=current_palm_age,
step=1,
help=t('upload_palm_age_help', 'Enter the age of your palm trees (1-35 years)'),
key="palm_age_input"
)
# Validation message for palm age
if palm_age == 0:
st.caption("⚠️ " + t('upload_palm_age_error', 'Please enter a valid age between 1 and 35 years'))
else:
st.caption(f"✅ {palm_age} " + ("years old" if palm_age > 1 else "year old"))
st.session_state.land_yield_data['palm_age'] = palm_age
# Soil Type Section
st.markdown("---")
st.markdown(f"### 🏔️ {t('upload_soil_type', 'What soil type do you have?')}")
# Define soil type options with translation keys
soil_type_options = [
('mineral', t('soil_type_mineral', 'Mineral soil')),
('peat', t('soil_type_peat', 'Peat soil')),
('alluvial', t('soil_type_alluvial', 'River alluvial soil')),
('bris', t('soil_type_bris', 'Coastal sandy soil (BRIS)')),
('acid_sulfate', t('soil_type_acid_sulfate', 'Acid sulfate soil')),
('saline', t('soil_type_saline', 'Coastal saline soil')),
('laterite', t('soil_type_laterite', 'Shallow laterite soil, stony soil')),
('limestone', t('soil_type_limestone', 'Limestone soil, coral soil')),
('not_sure', t('soil_type_not_sure', 'Not sure'))
]
# Get current soil type selection
current_soil_type = st.session_state.land_yield_data.get('soil_type', '')
# Find the index of current selection
soil_type_keys = [opt[0] for opt in soil_type_options]
soil_type_labels = [opt[1] for opt in soil_type_options]
# Create radio buttons for soil type selection
selected_soil_label = st.radio(
t('upload_soil_type_help', 'Select the soil type of your plantation'),
options=soil_type_labels,
index=soil_type_keys.index(current_soil_type) if current_soil_type in soil_type_keys else None,
key="soil_type_input",
horizontal=True
)
# Map the selected label back to the key
if selected_soil_label:
selected_soil_key = soil_type_keys[soil_type_labels.index(selected_soil_label)]
st.session_state.land_yield_data['soil_type'] = selected_soil_key
else:
st.session_state.land_yield_data['soil_type'] = ''
st.markdown("---")
# Display summary
soil_type_display = selected_soil_label if selected_soil_label else "Not selected"
if land_size > 0 or current_yield > 0:
st.info(f"📊 **Summary:** {land_size} {land_unit} | {current_yield} {yield_unit} | {palm_density} palms/ha | {palm_age} years | {soil_type_display}")
# Save button for land & yield data
col_save1, col_save2, col_save3 = st.columns([1, 2, 1])
with col_save2:
if st.button("💾 Save Land & Yield Data", type="secondary", width="stretch", key="save_land_yield"):
if land_size > 0 and current_yield > 0:
try:
from utils.firebase_config import get_firestore_client, initialize_firebase
db = get_firestore_client()
if not db:
initialize_firebase()
db = get_firestore_client()
if db:
machine_id = get_machine_id()
# Store in separate collection 'land_yield_data' using machine_id as document ID
land_yield_ref = db.collection('land_yield_data').document(machine_id)
# Check if document exists to preserve created_at
existing_doc = land_yield_ref.get()
created_at = existing_doc.to_dict().get('created_at', datetime.now()) if existing_doc.exists else datetime.now()
land_yield_ref.set({
'land_size': land_size,
'land_unit': land_unit,
'current_yield': current_yield,
'yield_unit': yield_unit,
'palm_density': palm_density,
'palm_age': palm_age,
'soil_type': st.session_state.land_yield_data.get('soil_type', ''),
'machine_id': machine_id,
'last_updated': datetime.now(),
'created_at': created_at
}, merge=True)
st.success("✅ Land & Yield data saved successfully!")
else:
st.error("❌ Database connection not available.")
except Exception as e:
st.error(f"❌ Failed to save data: {str(e)}")
else:
st.warning("⚠️ Please enter both land size and current yield before saving.")
# Analysis button section
st.markdown("---")
# Check requirements - validate files are actually accessible
soil_uploaded = False
leaf_uploaded = False
# Check if soil file exists and is valid
if st.session_state.get('soil_file') is not None:
try:
# Try to access file attributes to verify it's valid
soil_file_obj = st.session_state.soil_file
if hasattr(soil_file_obj, 'name') or hasattr(soil_file_obj, 'getvalue'):
soil_uploaded = True
except Exception:
# If file is invalid, try to restore from cache
if 'soil_file_data' in st.session_state:
try:
from io import BytesIO
cached_file = BytesIO(st.session_state.soil_file_data)
cached_file.name = st.session_state.get('soil_file_name', 'soil_file')
cached_file.type = st.session_state.get('soil_file_type', 'application/octet-stream')
st.session_state.soil_file = cached_file
soil_uploaded = True
except Exception:
soil_uploaded = False
# Check if leaf file exists and is valid
if st.session_state.get('leaf_file') is not None:
try:
# Try to access file attributes to verify it's valid
leaf_file_obj = st.session_state.leaf_file
if hasattr(leaf_file_obj, 'name') or hasattr(leaf_file_obj, 'getvalue'):
leaf_uploaded = True
except Exception:
# If file is invalid, try to restore from cache
if 'leaf_file_data' in st.session_state:
try:
from io import BytesIO
cached_file = BytesIO(st.session_state.leaf_file_data)
cached_file.name = st.session_state.get('leaf_file_name', 'leaf_file')
cached_file.type = st.session_state.get('leaf_file_type', 'application/octet-stream')
st.session_state.leaf_file = cached_file
leaf_uploaded = True
except Exception:
leaf_uploaded = False
palm_age_valid = palm_age >= 1 and palm_age <= 35
soil_type_selected = st.session_state.land_yield_data.get('soil_type', '') != ''
land_yield_provided = land_size > 0 and current_yield > 0 and palm_age_valid and soil_type_selected
# CRITICAL: First update session state from URL params to get latest values
# This must happen BEFORE checking can_start_analysis() to ensure accurate limit checking
limit_info = None
uploads_used = 0
uploads_limit = 0
uploads_remaining = 0
try:
from utils.cropdrive_integration import get_upload_limit_info
limit_info = get_upload_limit_info()
except ImportError:
# CropDrive integration not available
pass
except Exception:
# Error getting limit info
pass
if soil_uploaded and leaf_uploaded and land_yield_provided:
# Show upload limit info if available
# CRITICAL: Always refresh from URL params to get latest count (parent may have updated it)
try:
query_params = st.query_params
url_uploads_used = query_params.get('uploadsUsed', '')
url_uploads_limit = query_params.get('uploadsLimit', '')
url_uploads_remaining = query_params.get('uploadsRemaining', '')
# Use URL params if available (most up-to-date), otherwise use limit_info
if url_uploads_used and url_uploads_limit:
try:
uploads_used = int(url_uploads_used)
uploads_limit = int(url_uploads_limit)
if url_uploads_remaining:
try:
uploads_remaining = int(url_uploads_remaining) if url_uploads_remaining != 'infinity' and url_uploads_remaining != '-1' else float('inf')
except (ValueError, TypeError):
uploads_remaining = max(0, uploads_limit - uploads_used)
else:
uploads_remaining = max(0, uploads_limit - uploads_used)
# Update session state with latest values
st.session_state.uploads_used = uploads_used
st.session_state.uploads_limit = uploads_limit
st.session_state.uploads_remaining = uploads_remaining
except (ValueError, TypeError):
# Fallback to limit_info if URL params invalid
uploads_used = limit_info.get('uploads_used', 0) if limit_info else 0
uploads_limit = limit_info.get('uploads_limit', 0) if limit_info else 0
uploads_remaining = limit_info.get('uploads_remaining', 0) if limit_info else 0
else:
# Use limit_info if URL params not available
uploads_used = limit_info.get('uploads_used', 0) if limit_info else 0
uploads_limit = limit_info.get('uploads_limit', 0) if limit_info else 0
uploads_remaining = limit_info.get('uploads_remaining', 0) if limit_info else 0
except Exception as e:
# Fallback to limit_info on error
uploads_used = limit_info.get('uploads_used', 0) if limit_info else 0
uploads_limit = limit_info.get('uploads_limit', 0) if limit_info else 0
uploads_remaining = limit_info.get('uploads_remaining', 0) if limit_info else 0
# CRITICAL: Now check if analysis can start AFTER updating session state
can_start = True
try:
from utils.cropdrive_integration import can_start_analysis
can_start = can_start_analysis()
except ImportError:
# CropDrive integration not available, allow analysis
can_start = True
except Exception:
# Error checking limits, allow analysis to proceed
can_start = True
if limit_info or (uploads_limit > 0):
# Only show limit info if limit is set (greater than 0)
if uploads_limit > 0:
if uploads_remaining == float('inf') or uploads_remaining == -1:
st.info(f"📊 **Analyses:** Unlimited")
else:
# Display: X/Y used (Z remaining)
# Example: 0/2 used (2 remaining), 1/2 used (1 remaining), 2/2 used (0 remaining)
st.info(f"📊 **Analyses:** {uploads_used}/{uploads_limit} used ({uploads_remaining} remaining)")
# Check if limit exceeded
if not can_start:
st.error("❌ **Analysis Limit Reached**")
st.info("💡 You have reached your analysis limit. Please upgrade your plan to continue analyzing lab reports.")
st.button(f"🚀 {t('upload_start_analysis')}", disabled=True, width="stretch", key="start_analysis_disabled_limit")
elif st.button(f"🚀 {t('upload_start_analysis')}", type="primary", width="stretch", key="start_analysis"):
# Ensure files are valid before passing to analysis
try:
# Double-check limit before starting (in case it changed)
try:
from utils.cropdrive_integration import can_start_analysis
if not can_start_analysis():
st.error("❌ **Analysis Limit Reached**")
st.info("💡 You have reached your analysis limit. Please upgrade your plan to continue analyzing lab reports.")
st.stop()
except Exception:
# If check fails, allow to proceed
pass
soil_file = st.session_state.soil_file
leaf_file = st.session_state.leaf_file
# Verify files are accessible
if not hasattr(soil_file, 'getvalue') and 'soil_file_data' in st.session_state:
from io import BytesIO
soil_file = BytesIO(st.session_state.soil_file_data)
soil_file.name = st.session_state.get('soil_file_name', 'soil_file')
soil_file.type = st.session_state.get('soil_file_type', 'application/octet-stream')
if not hasattr(leaf_file, 'getvalue') and 'leaf_file_data' in st.session_state:
from io import BytesIO
leaf_file = BytesIO(st.session_state.leaf_file_data)
leaf_file.name = st.session_state.get('leaf_file_name', 'leaf_file')
leaf_file.type = st.session_state.get('leaf_file_type', 'application/octet-stream')
st.session_state.analysis_data = {
'soil_file': soil_file,
'leaf_file': leaf_file,
'land_yield_data': st.session_state.land_yield_data
}
st.session_state.current_page = 'results'
st.rerun()
except Exception as e:
st.error(f"Error preparing files for analysis: {str(e)}")
st.info("Please try uploading the files again.")
else:
st.warning(f"⚠️ **{t('upload_requirements')}**")
if not soil_uploaded:
st.info(f"• {t('upload_need_soil')}")
if not leaf_uploaded:
st.info(f"• {t('upload_need_leaf')}")
if land_size == 0 or current_yield == 0:
st.info(f"• {t('upload_need_yield')}")
if not palm_age_valid:
st.info(f"• {t('upload_need_palm_age', 'Enter a valid palm tree age (1-35 years)')}")
if not soil_type_selected:
st.info(f"• {t('upload_need_soil_type', 'Select your soil type')}")
st.button(f"🚀 {t('upload_start_analysis')}", disabled=True, width="stretch", key="start_analysis_disabled")
def _parse_numeric_token(token: str):
"""Parse a single OCR token into float; N.D. / ND / <x become 0.0 or the number."""
if token is None:
return None
text = str(token).strip()
if not text:
return None
upper = text.upper().replace(' ', '')
if upper in ('N.D.', 'N.D', 'ND', 'N/A', 'NA', '-', '--'):
return 0.0
# Handle <1, <0.01 etc. as 0 for structured numeric use
if text.startswith('<') or text.startswith('>'):
text = text[1:].strip()
if not text:
return 0.0
cleaned = re.sub(r'[^\d.\-]', '', text)
if not cleaned or cleaned in ('.', '-', '-.'):
return None
try:
return float(cleaned)
except (ValueError, TypeError):
return None
def _score_soil_window(window: list) -> float:
"""Higher = more realistic oil-palm soil lab row (pH, N, OrgC, P, cations, CEC)."""
if len(window) < 5:
return -100.0
score = 0.0
ph = window[0]
# pH must look like pH
if 3.0 <= ph <= 7.5:
score += 5.0
elif 2.5 <= ph <= 9.0:
score += 2.0
else:
score -= 8.0
# N (%) typically 0.03–0.40
if len(window) > 1:
n = window[1]
if 0.02 <= n <= 0.5:
score += 3.0
elif n > 2:
score -= 4.0
# Org C (%) typically 0.2–5
if len(window) > 2:
oc = window[2]
if 0.1 <= oc <= 6.0:
score += 2.0
elif oc > 20:
score -= 3.0
# Total P mg/kg often 5–400
if len(window) > 3:
tp = window[3]
if 1 <= tp <= 800:
score += 2.0
elif 0 < tp < 1:
score -= 2.0
# Avail P
if len(window) > 4:
ap = window[4]
if 0 <= ap <= 80:
score += 1.5
# Exch cations small
for idx in (5, 6, 7):
if len(window) > idx and 0 <= window[idx] <= 15:
score += 0.5
# CEC usually > exch cations and 1–40
if len(window) > 8:
cec = window[8]
if 1 <= cec <= 50:
score += 2.0
if len(window) > 5 and cec >= window[5]:
score += 0.5
return score
def _score_leaf_window(window: list) -> float:
"""Higher = more realistic oil-palm leaf lab row (N,P,K,Mg,Ca,B,Cu,Zn)."""
if len(window) < 5:
return -100.0
score = 0.0
n = window[0]
# Leaf N % typically ~1.8–3.2
if 1.6 <= n <= 3.5:
score += 6.0
elif 1.2 <= n <= 4.0:
score += 3.0
else:
score -= 8.0
# P % typically 0.08–0.25
if len(window) > 1:
p = window[1]
if 0.08 <= p <= 0.30:
score += 4.0
elif p > 1.0:
score -= 5.0 # shifted N sitting in P slot
# K % typically 0.4–1.5
if len(window) > 2:
k = window[2]
if 0.35 <= k <= 2.0:
score += 2.0
elif k > 5:
score -= 2.0
# Mg, Ca %
if len(window) > 3 and 0.05 <= window[3] <= 0.8:
score += 1.0
if len(window) > 4 and 0.2 <= window[4] <= 1.5:
score += 1.0
# B mg/kg typically 8–30; Cu 1–10; Zn 5–30
if len(window) > 5:
b = window[5]
if 5 <= b <= 50:
score += 3.0
elif b < 1:
score -= 3.0 # Ca sitting in B slot
if len(window) > 6 and 0 <= window[6] <= 15:
score += 1.0
if len(window) > 7 and 3 <= window[7] <= 40:
score += 1.0
return score
def _align_numbers_to_params(numbers: list, params: list, sample_id: str = '', kind: str = 'soil') -> dict:
"""Map OCR number sequence onto params by sliding-window fitness (no hardcoded samples).
Handles leading junk (lab serial, row index, year) that otherwise shifts all columns —
e.g. leaf N=2.13 becoming P because a leading 1.0 was treated as N.
"""
if not numbers or not params:
return {}
nums = list(numbers)
# Drop large digits that repeat the sample lab number (e.g. 218 from S218/25)
if sample_id:
id_digits = re.findall(r'\d+', sample_id)
while nums and any(abs(nums[0] - float(d)) < 1e-9 for d in id_digits if float(d) >= 10):
if nums[0] >= 10:
nums = nums[1:]
else:
break
n_params = len(params)
scorer = _score_soil_window if kind == 'soil' else _score_leaf_window
if len(nums) < 3:
return {params[i]: nums[i] for i in range(len(nums))}
best_score = -1e9
best_window = nums[:n_params]
# Try every alignment window (extra leading/trailing OCR numbers)
max_start = max(0, len(nums) - n_params)
for start in range(0, max_start + 1):
window = nums[start:start + n_params]
if len(window) < min(5, n_params):
continue
# pad short windows
while len(window) < n_params:
window = window # no pad for scoring; scorer handles short
break
s = scorer(window)
# prefer earlier windows slightly when scores tie
s -= start * 0.01
if s > best_score:
best_score = s
best_window = window
# If best score is still terrible, fall back to first plausible anchor
if kind == 'soil' and best_score < 0:
for start, val in enumerate(nums):
if 2.5 <= val <= 9.5:
best_window = nums[start:start + n_params]
break
elif kind == 'leaf' and best_score < 0:
for start, val in enumerate(nums):
if 1.5 <= val <= 3.5:
# next should look like P if available
if start + 1 < len(nums) and nums[start + 1] > 1.0:
continue
best_window = nums[start:start + n_params]
break
sample_data = {}
for i, param in enumerate(params):
if i < len(best_window):
sample_data[param] = best_window[i]
return sample_data
def _align_soil_numbers_to_params(numbers: list, params: list, sample_id: str = '') -> dict:
"""Backward-compatible soil aligner."""
return _align_numbers_to_params(numbers, params, sample_id, kind='soil')
def _parse_sp_lab_vertical_document(text: str, sample_prefix: str, params: list) -> dict:
"""Parse SP Lab reports where Document AI dumps one token per line.
Real layout (user OCR)::
S218/25
S219/25
...
1
2
3
PH
5.0
4.3
...
Nitrogen
0.10
...
Sample IDs form columns; each parameter label is followed by one value
per sample (same order). Multiple table sections are supported.
"""
if not text:
return {}
lines = [ln.strip() for ln in text.split('\n') if ln.strip()]
prefix = sample_prefix.upper()
id_re = re.compile(rf'^{prefix}\d{{1,4}}\s*/\s*\d{{2}}$', re.I)
id_find = re.compile(rf'\b({prefix}\d{{1,4}}\s*/\s*\d{{2}})\b', re.I)
def canon(sid: str) -> str:
sid = re.sub(r'\s+', '', sid.strip())
return sid[0].upper() + sid[1:] if sid else sid
# Soil / leaf parameter label matchers (order matters for specificity)
if params and params[0] == 'pH':
label_specs = [
(re.compile(r'^pH$', re.I), 'pH'),
(re.compile(r'^PH$', re.I), 'pH'),
(re.compile(r'^Nitrogen$', re.I), 'N (%)'),
(re.compile(r'^N\s*\(%\)$', re.I), 'N (%)'),
(re.compile(r'^Organic\s*Carbon$', re.I), 'Org. C (%)'),
(re.compile(r'^Org\.?\s*C', re.I), 'Org. C (%)'),
(re.compile(r'^Total\s*P', re.I), 'Total P (mg/kg)'),
(re.compile(r'^Available\s*P', re.I), 'Avail P (mg/kg)'),
(re.compile(r'^Avail\.?\s*P', re.I), 'Avail P (mg/kg)'),
(re.compile(r'^Exch\.?\s*K', re.I), 'Exch. K (meq%)'),
(re.compile(r'^Exch\.?\s*Ca', re.I), 'Exch. Ca (meq%)'),
(re.compile(r'^Exch\.?\s*Mg', re.I), 'Exch. Mg (meq%)'),
(re.compile(r'^C\.?E\.?C\.?$', re.I), 'CEC (meq%)'),
]
else:
label_specs = [
(re.compile(r'^N(?:itrogen)?$', re.I), 'N (%)'),
(re.compile(r'^P(?:hos(?:phorus)?)?$', re.I), 'P (%)'),
(re.compile(r'^K(?:alium|potassium)?$', re.I), 'K (%)'),
(re.compile(r'^Mg(?:agnesium)?$', re.I), 'Mg (%)'),
(re.compile(r'^Ca(?:lcium)?$', re.I), 'Ca (%)'),
(re.compile(r'^B(?:oron)?$', re.I), 'B (mg/kg)'),
(re.compile(r'^Cu(?:pper)?$', re.I), 'Cu (mg/kg)'),
(re.compile(r'^Zn(?:inc)?$', re.I), 'Zn (mg/kg)'),
]
skip_labels = re.compile(
r'^(Exchangeable|Parameter|Lab\s*No.*|Analysis|Date|Page|Serial|'
r'Your\s*Ref|TEST\s*REPORT|SP\s*LAB|Sarawak|Non-?Detected|'
r'%|mg/kg|meq%|meq/?%?)$',
re.I,
)
def match_param_label(line: str):
for preg, key in label_specs:
if preg.match(line.strip()):
return key
return None
def is_pure_index(line: str) -> bool:
"""Column index rows like 1,2,3...10 that appear under sample IDs."""
return bool(re.fullmatch(r'\d{1,2}', line.strip()))
def is_value_token(line: str) -> bool:
s = line.strip()
if not s:
return False
if id_re.match(s) or id_find.search(s):
return False
if match_param_label(s):
return False
if skip_labels.match(s):
return False
# N.D. / ND
if re.fullmatch(r'N\.?\s*D\.?', s, re.I):
return True
if re.fullmatch(r'<?\d+\.?\d*', s):
return True
return False
def parse_value(line: str):
s = line.strip()
if re.fullmatch(r'N\.?\s*D\.?', s, re.I):
return 0.0
return _parse_numeric_token(s)
# Split document into table sections (each starts near sample IDs after "Lab No" / "Parameter")
sections = []
current = []
for line in lines:
if re.search(r'Lab\s*No', line, re.I) and current:
sections.append(current)
current = [line]
else:
current.append(line)
if current:
sections.append(current)
# If no clear section markers, treat whole doc as one section
if len(sections) <= 1:
# Also try splitting when a new block of sample IDs appears after parameters
sections = [lines]
all_samples = {}
def parse_section(sec_lines: list) -> dict:
# Collect sample IDs in order of appearance (within this section)
sample_ids = []
seen = set()
for line in sec_lines:
for m in id_find.finditer(line):
sid = canon(m.group(1))
if sid not in seen:
seen.add(sid)
sample_ids.append(sid)
if len(sample_ids) < 1:
return {}
n = len(sample_ids)
# param -> list of n values
param_cols = {p: None for p in params}
i = 0
while i < len(sec_lines):
line = sec_lines[i]
# Skip sample id lines, indices, units, headers
if id_find.search(line) and not match_param_label(line):
i += 1
continue
if is_pure_index(line):
i += 1
continue
if skip_labels.match(line):
i += 1
continue
pkey = match_param_label(line)
if pkey and pkey in param_cols:
# Collect the next `n` value tokens after this parameter label.
# IMPORTANT: after a label, digits like 59 / 2 / 1 are REAL lab
# values (Total P, Available P), NOT column indices 1..10.
values = []
j = i + 1
while j < len(sec_lines) and len(values) < n:
nxt = sec_lines[j]
# Stop if we hit next param label
if match_param_label(nxt):
break
# Units / section words only — do not treat digits as indices here
if skip_labels.match(nxt):
j += 1
# If we already have values and hit a unit line, we're done
if values:
break
continue
if id_find.search(nxt):
break
if re.fullmatch(r'N\.?\s*D\.?', nxt, re.I) or re.fullmatch(r'<?\d+\.?\d*', nxt):
val = parse_value(nxt)
if val is not None:
values.append(val)
j += 1
else:
j += 1
if values:
break
if len(values) >= n:
param_cols[pkey] = values[:n]
elif len(values) > 0:
# Accept partial only if nearly complete (e.g. trailing missing)
if len(values) >= n - 1:
param_cols[pkey] = values
i = j
continue
i += 1
# Require at least pH/N + a few more params filled
filled = [p for p in params if param_cols.get(p)]
if len(filled) < 3:
return {}
# Normalize lengths: use min available
out = {sid: {} for sid in sample_ids}
for p in params:
col = param_cols.get(p)
if not col:
continue
for idx, sid in enumerate(sample_ids):
if idx < len(col):
out[sid][p] = col[idx]
# Drop empty
return {sid: data for sid, data in out.items() if data}
# Parse each Lab-No section separately (critical: each table has its own
# sample count — do NOT parse the whole doc as one 10-column table).
for sec in sections:
part = parse_section(sec)
for sid, data in part.items():
if sid not in all_samples:
all_samples[sid] = data
else:
# Merge params; prefer non-empty / longer
for k, v in data.items():
if k not in all_samples[sid]:
all_samples[sid][k] = v
# If Lab-No split failed, try sample-id cluster split
if len(all_samples) < 2:
clusters = []
cluster = []
saw_id = False
for line in lines:
has_id = bool(id_find.search(line))
if has_id:
if saw_id is False and cluster and any(id_find.search(x) for x in cluster):
# new cluster after params
clusters.append(cluster)
cluster = []
saw_id = True
cluster.append(line)
else:
if cluster:
cluster.append(line)
# reset saw_id after we leave a run of IDs and hit a param
if match_param_label(line):
saw_id = False
if cluster:
clusters.append(cluster)
for cl in clusters:
part = parse_section(cl)
all_samples.update(part)
# Last resort: whole document only if still empty
if not all_samples:
all_samples = parse_section(lines)
return all_samples
def _extract_sp_lab_samples_from_text(text: str, sample_prefix: str, params: list) -> dict:
"""
Parse SP Lab samples (e.g. S218/25 or P220/25) from real OCR text.
Never uses hardcoded values — structured data must match Raw OCR Text.
Handles:
- Vertical Document AI dumps (one token per line) — SP Lab official layout
- Row-per-sample: S218/25 5.26 0.12 0.91 ...
- Column-major / transposed (samples on one line, params as rows)
- Leading junk numbers before pH (lab serial bleeding into values)
"""
samples = {}
if not text:
return samples
# ------------------------------------------------------------------
# Strategy V (highest priority): vertical one-token-per-line SP Lab OCR
# ------------------------------------------------------------------
vertical = _parse_sp_lab_vertical_document(text, sample_prefix, params)
if vertical:
# Validate soil pH columns look sane before accepting
if params and params[0] == 'pH':
ph_ok = sum(
1 for d in vertical.values()
if isinstance(d.get('pH'), (int, float)) and 2.5 <= d['pH'] <= 9.5
)
if ph_ok >= max(1, len(vertical) // 2):
return vertical
else:
# leaf: N in range
n_ok = sum(
1 for d in vertical.values()
if isinstance(d.get('N (%)'), (int, float)) and 1.2 <= d['N (%)'] <= 4.0
)
if n_ok >= max(1, len(vertical) // 2) or len(vertical) >= 1:
return vertical
# even if validation weak, prefer vertical if it found many samples with many params
avg_params = sum(len(d) for d in vertical.values()) / max(len(vertical), 1)
if len(vertical) >= 3 and avg_params >= 5:
return vertical
lines = [ln.strip() for ln in text.split('\n') if ln.strip()]
joined = ' '.join(lines)
joined_norm = re.sub(r'[^\w\s\.\-\(\)\[\]\/<>%]', ' ', joined)
joined_norm = re.sub(r'\s+', ' ', joined_norm)
prefix = sample_prefix.upper()
id_re = re.compile(rf'\b({prefix}\d{{1,4}}\s*/\s*\d{{2}})\b', re.IGNORECASE)
def _canonical_sample_id(raw_id: str) -> str:
sid = re.sub(r'\s+', '', raw_id.strip())
if sid and sid[0].isalpha():
return sid[0].upper() + sid[1:]
return sid
def _tokens_to_numbers(tokens):
numbers = []
for tok in tokens:
tok_s = str(tok).strip()
if id_re.fullmatch(tok_s) or id_re.fullmatch(re.sub(r'\s+', '', tok_s)):
continue
val = _parse_numeric_token(tok_s)
if val is not None:
numbers.append(val)
return numbers
min_needed = max(3, min(5, len(params) - 2))
is_soil = bool(params and params[0] == 'pH')
# ------------------------------------------------------------------
# Strategy T: transposed / column-major SP Lab layout (values on same line)
# ------------------------------------------------------------------
def _parse_transposed() -> dict:
# Collect ordered unique sample IDs as they appear
id_matches = list(id_re.finditer(joined_norm))
if len(id_matches) < 2:
return {}
sample_ids = []
seen = set()
for m in id_matches:
sid = _canonical_sample_id(m.group(1))
if sid not in seen:
seen.add(sid)
sample_ids.append(sid)
n_samples = len(sample_ids)
if n_samples < 2:
return {}
# Parameter row labels → param key
param_row_patterns = []
if is_soil:
param_row_patterns = [
(re.compile(r'^\s*pH\b', re.I), 'pH'),
(re.compile(r'^\s*N(?:itrogen)?\b(?!\s*/)', re.I), 'N (%)'),
(re.compile(r'^\s*(?:Org\.?\s*C|Organic\s*Carbon)\b', re.I), 'Org. C (%)'),
(re.compile(r'^\s*Total\s*P\b', re.I), 'Total P (mg/kg)'),
(re.compile(r'^\s*Avail(?:able)?\s*P\b', re.I), 'Avail P (mg/kg)'),
(re.compile(r'^\s*Exch\.?\s*K\b', re.I), 'Exch. K (meq%)'),
(re.compile(r'^\s*Exch\.?\s*Ca\b', re.I), 'Exch. Ca (meq%)'),
(re.compile(r'^\s*Exch\.?\s*Mg\b', re.I), 'Exch. Mg (meq%)'),
(re.compile(r'^\s*C\.?E\.?C\b', re.I), 'CEC (meq%)'),
]
else:
param_row_patterns = [
(re.compile(r'^\s*N(?:itrogen)?\b', re.I), 'N (%)'),
(re.compile(r'^\s*P(?:hos)?\b', re.I), 'P (%)'),
(re.compile(r'^\s*K(?:alium|potassium)?\b', re.I), 'K (%)'),
(re.compile(r'^\s*Mg\b', re.I), 'Mg (%)'),
(re.compile(r'^\s*Ca\b', re.I), 'Ca (%)'),
(re.compile(r'^\s*B(?:oron)?\b', re.I), 'B (mg/kg)'),
(re.compile(r'^\s*Cu\b', re.I), 'Cu (mg/kg)'),
(re.compile(r'^\s*Zn\b', re.I), 'Zn (mg/kg)'),
]
# Build param → list of values (one per sample column)
param_columns = {p: [] for p in params}
rows_matched = 0
for line in lines:
# Skip pure sample-id header lines
if id_re.search(line) and len(_tokens_to_numbers(re.split(r'\s+', line))) < n_samples:
continue
matched_param = None
matched_preg = None
for preg, pkey in param_row_patterns:
if preg.search(line) and pkey in param_columns:
matched_param = pkey
matched_preg = preg
break
if not matched_param or not matched_preg:
continue
# Numbers on this row after the label
vals_part = matched_preg.sub('', line, count=1)
vals_part = re.sub(r'\bN\.?\s*D\.?\b', '0', vals_part, flags=re.I)
vals_part = re.sub(r'<(\d+\.?\d*)', r'\1', vals_part)
numbers = _tokens_to_numbers(re.split(r'\s+', vals_part.strip()))
if len(numbers) >= n_samples:
param_columns[matched_param] = numbers[:n_samples]
rows_matched += 1
elif len(numbers) >= 2:
param_columns[matched_param] = numbers
rows_matched += 1
# Need at least pH (or N for leaf) + a few more rows
filled = sum(1 for p in params if param_columns.get(p))
if filled < min_needed:
return {}
found = {}
for col_idx, sid in enumerate(sample_ids):
sample_data = {}
for p in params:
col_vals = param_columns.get(p) or []
if col_idx < len(col_vals):
sample_data[p] = col_vals[col_idx]
if sample_data:
found[sid] = sample_data
return found
transposed = _parse_transposed()
if transposed and len(transposed) >= 2:
# Validate soil pH columns look sane
if is_soil:
ph_ok = sum(
1 for s in transposed.values()
if isinstance(s.get('pH'), (int, float)) and 2.5 <= s['pH'] <= 9.5
)
if ph_ok >= max(1, len(transposed) // 2):
return transposed
else:
return transposed
kind = 'soil' if is_soil else 'leaf'
def _parse_between_ids(source_text: str) -> dict:
"""Split source on sample IDs and map numbers after each ID to params."""
found = {}
matches = list(id_re.finditer(source_text))
for i, m in enumerate(matches):
sample_id = _canonical_sample_id(m.group(1))
start = m.end()
end = matches[i + 1].start() if i + 1 < len(matches) else len(source_text)
chunk = source_text[start:end]
chunk = re.sub(r'\bN\.?\s*D\.?\b', '0', chunk, flags=re.IGNORECASE)
chunk = re.sub(r'<(\d+\.?\d*)', r'\1', chunk)
tokens = re.split(r'\s+', chunk.strip()) if chunk.strip() else []
numbers = _tokens_to_numbers(tokens)
if len(numbers) >= min_needed:
sample_data = _align_numbers_to_params(numbers, params, sample_id, kind=kind)
if sample_data:
found[sample_id] = sample_data
return found
# --- Strategy A: sequential IDs on each line ---
for line in lines:
samples.update(_parse_between_ids(line))
# --- Strategy B: full flattened text ---
full_samples = _parse_between_ids(joined_norm)
if not samples:
samples = full_samples
elif len(full_samples) > len(samples):
samples = full_samples
else:
# Merge: prefer samples with valid pH / more params
for sid, data in full_samples.items():
if sid not in samples:
samples[sid] = data
elif is_soil:
old_ph = samples[sid].get('pH')
new_ph = data.get('pH')
if (not isinstance(old_ph, (int, float)) or not (2.5 <= old_ph <= 9.5)) and (
isinstance(new_ph, (int, float)) and 2.5 <= new_ph <= 9.5
):
samples[sid] = data
elif len(data) > len(samples[sid]):
samples[sid] = data
# If we only got one sample but many IDs, transposed may have partially failed —
# try building from pH-labeled line alone as recovery
id_count = len(list(id_re.finditer(joined_norm)))
if id_count >= 2 and len(samples) < id_count and is_soil:
# Find "pH ..." line with enough numbers
for line in lines:
if re.search(r'\bpH\b', line, re.I):
ph_nums = _tokens_to_numbers(re.split(r'\s+', re.sub(r'(?i)\bpH\b', ' ', line)))
sids = [_canonical_sample_id(m.group(1)) for m in id_re.finditer(joined_norm)]
# unique preserve order
uniq = []
seen = set()
for s in sids:
if s not in seen:
seen.add(s)
uniq.append(s)
if len(ph_nums) >= len(uniq) >= 2:
for i, sid in enumerate(uniq):
if sid not in samples:
samples[sid] = {}
samples[sid]['pH'] = ph_nums[i]
break
# --- Strategy C: labeled pairs per sample block ---
if not samples:
matches = list(id_re.finditer(joined_norm))
label_aliases = {
'ph': 'pH',
'n': 'N (%)',
'nitrogen': 'N (%)',
'org': 'Org. C (%)',
'organic': 'Org. C (%)',
'orgc': 'Org. C (%)',
'total': 'Total P (mg/kg)',
'totalp': 'Total P (mg/kg)',
'avail': 'Avail P (mg/kg)',
'available': 'Avail P (mg/kg)',
'availp': 'Avail P (mg/kg)',
'exchk': 'Exch. K (meq%)',
'exchca': 'Exch. Ca (meq%)',
'exchmg': 'Exch. Mg (meq%)',
'k': 'Exch. K (meq%)' if prefix == 'S' else 'K (%)',
'ca': 'Exch. Ca (meq%)' if prefix == 'S' else 'Ca (%)',
'mg': 'Exch. Mg (meq%)' if prefix == 'S' else 'Mg (%)',
'cec': 'CEC (meq%)',
'ce': 'CEC (meq%)',
'p': 'P (%)',
'b': 'B (mg/kg)',
'cu': 'Cu (mg/kg)',
'zn': 'Zn (mg/kg)',
}
for i, m in enumerate(matches):
sample_id = _canonical_sample_id(m.group(1))
start = m.end()
end = matches[i + 1].start() if i + 1 < len(matches) else len(joined_norm)
chunk = joined_norm[start:end]
sample_data = {}
pair_re = re.compile(
r'(pH|N(?:itrogen)?|Org\.?\s*C|Organic\s*Carbon|Total\s*P|Avail(?:able)?\s*P|'
r'Exch\.?\s*K|Exch\.?\s*Ca|Exch\.?\s*Mg|C\.?E\.?C|P|K|Mg|Ca|B|Cu|Zn)'
r'(?:\s*\([^)]*\))?\s*[:=\s]\s*(N\.?D\.?|<?\d+\.?\d*)',
re.IGNORECASE,
)
for pm in pair_re.finditer(chunk):
label = re.sub(r'[^a-zA-Z]', '', pm.group(1)).lower()
if label.startswith('org'):
key = 'Org. C (%)'
elif label.startswith('total'):
key = 'Total P (mg/kg)'
elif label.startswith('avail'):
key = 'Avail P (mg/kg)'
elif label.startswith('exch') and 'k' in pm.group(1).lower():
key = 'Exch. K (meq%)'
elif label.startswith('exch') and 'ca' in pm.group(1).lower():
key = 'Exch. Ca (meq%)'
elif label.startswith('exch') and 'mg' in pm.group(1).lower():
key = 'Exch. Mg (meq%)'
elif label in ('cec', 'ce'):
key = 'CEC (meq%)'
else:
key = label_aliases.get(label)
if not key or key not in params:
continue
val = _parse_numeric_token(pm.group(2))
if val is not None:
sample_data[key] = val
if sample_data:
samples[sample_id] = sample_data
return samples
def format_raw_text_as_structured_json(raw_text: str, container_type: str) -> dict:
"""Format raw OCR text into structured JSON format by analyzing the content.
Structured OCR Data must be derived from the same Raw OCR Text — never hardcoded samples.
"""
try:
# Initialize structured data
structured_data = {}
samples_data = {}
# Enhanced text preprocessing
raw_text = (raw_text or '').strip()
# Split text into lines and clean them
lines = [line.strip() for line in raw_text.split('\n') if line.strip()]
# Remove common OCR artifacts and normalize (used by farm-format strategies)
all_text = ' '.join(lines)
all_text = re.sub(r'[^\w\s\.\-\(\)\[\]\/]', ' ', all_text)
all_text = re.sub(r'\s+', ' ', all_text)
is_sp_soil = bool(
'SP LAB' in all_text.upper() or re.search(r'\bS\d{1,4}/\d{2}\b', all_text, re.IGNORECASE)
)
is_sp_leaf = bool(
'SP LAB' in all_text.upper() or re.search(r'\bP\d{1,4}/\d{2}\b', all_text, re.IGNORECASE)
)
if container_type.lower() == 'soil':
soil_params = [
'pH', 'N (%)', 'Org. C (%)', 'Total P (mg/kg)', 'Avail P (mg/kg)',
'Exch. K (meq%)', 'Exch. Ca (meq%)', 'Exch. Mg (meq%)', 'CEC (meq%)',
]
if is_sp_soil:
structured_data["SP_Lab_Test_Report"] = {}
else:
structured_data["Farm_Soil_Test_Data"] = {}
# Strategy 1: Parse REAL SP Lab samples from OCR text (S###/##)
if is_sp_soil:
samples_data = _extract_sp_lab_samples_from_text(raw_text, 'S', soil_params)
if samples_data:
structured_data["SP_Lab_Test_Report"] = samples_data
# Strategy 2: Farm sample patterns (S001, S002, etc.) — not SP Lab IDs
if not samples_data and not is_sp_soil:
sample_pattern = r'S(\d{1,3})\s*[:\-]?\s*([^S\n]*(?=S\d|\n|$))'
matches = re.findall(sample_pattern, all_text, re.IGNORECASE | re.DOTALL)
for match in matches[:12]:
sample_num, sample_values = match
sample_id = f"S{sample_num.zfill(3)}"
numbers = [float(x) for x in re.findall(r'(\d+\.?\d*)', sample_values)]
if len(numbers) >= 3:
sample_data = _align_numbers_to_params(numbers, soil_params, sample_id, kind='soil')
if sample_data and any(v != 0.0 for v in sample_data.values()):
samples_data[sample_id] = sample_data
# Strategy 3: Numbered samples (Sample 1, Sample 2, ...)
if not samples_data and not is_sp_soil:
sample_pattern = r'Sample\s*(\d{1,3})\s*[:\-]?\s*([^S\n]*(?=Sample\s*\d|\n|$))'
matches = re.findall(sample_pattern, all_text, re.IGNORECASE | re.DOTALL)
for match in matches[:12]:
sample_num, sample_values = match
sample_id = f"S{sample_num.zfill(3)}"
numbers = [float(x) for x in re.findall(r'(\d+\.?\d*)', sample_values)]
if len(numbers) >= 3:
sample_data = _align_numbers_to_params(numbers, soil_params, sample_id, kind='soil')
if sample_data and any(v != 0.0 for v in sample_data.values()):
samples_data[sample_id] = sample_data
# Strategy 4: Chunk all numbers into groups of 9 (farm tabular dump only)
if not samples_data and not is_sp_soil:
all_numbers = [float(x) for x in re.findall(r'(\d+\.?\d*)', all_text)]
if len(all_numbers) >= 27:
num_samples = min(12, len(all_numbers) // 9)
for i in range(num_samples):
sample_id = f"S{i+1:03d}"
start_idx = i * 9
chunk = all_numbers[start_idx:start_idx + 9]
sample_data = _align_numbers_to_params(chunk, soil_params, sample_id, kind='soil')
if sample_data and any(v != 0.0 for v in sample_data.values()):
samples_data[sample_id] = sample_data
# Strategy 5: Parameter-value pairs → single sample (last resort)
if not samples_data:
for param in soil_params:
param_pattern = rf'{re.escape(param.split()[0])}\s*[:=]\s*([^,\s\n]+)'
matches = re.findall(param_pattern, all_text, re.IGNORECASE)
for match in matches[:1]:
if not samples_data:
samples_data["S001"] = {}
value = match.strip()
if value.upper().replace('.', '') in ('ND', 'N/A', 'NA'):
value = '0'
try:
samples_data["S001"][param] = float(re.sub(r'[^\d.\-]', '', value) or '0')
except (ValueError, TypeError):
pass
# Attach samples to the correct container when not already set by SP Lab strategy
if samples_data and not structured_data.get("SP_Lab_Test_Report"):
container_key = "SP_Lab_Test_Report" if is_sp_soil else "Farm_Soil_Test_Data"
if container_key not in structured_data:
structured_data[container_key] = {}
for sample_id, sample_data in samples_data.items():
if sample_data and any(
(isinstance(v, (int, float)) and v != 0.0) or (not isinstance(v, (int, float)) and v)
for v in sample_data.values()
):
structured_data[container_key][sample_id] = sample_data
elif samples_data and is_sp_soil and not structured_data.get("SP_Lab_Test_Report"):
structured_data["SP_Lab_Test_Report"] = samples_data
elif container_type.lower() == 'leaf':
leaf_params = [
'N (%)', 'P (%)', 'K (%)', 'Mg (%)', 'Ca (%)',
'B (mg/kg)', 'Cu (mg/kg)', 'Zn (mg/kg)',
]
if is_sp_leaf:
structured_data["SP_Lab_Test_Report"] = {}
else:
structured_data["Farm_Leaf_Test_Data"] = {}
# Strategy 1: Parse REAL SP Lab leaf samples (P###/##)
if is_sp_leaf:
samples_data = _extract_sp_lab_samples_from_text(raw_text, 'P', leaf_params)
if samples_data:
structured_data["SP_Lab_Test_Report"] = samples_data
# Strategies for farm leaf (L001 style) — skip when SP Lab already filled
matches = []
if not samples_data and not is_sp_leaf:
sample_pattern = r'L(\d{1,3})\s*[:\-]?\s*([^L\n]*(?=L\d|\n|$))'
matches = re.findall(sample_pattern, all_text, re.IGNORECASE | re.DOTALL)
if not matches:
sample_pattern = r'Sample\s*(\d{1,3})\s*[:\-]?\s*([^S\n]*(?=Sample\s*\d|\n|$))'
matches = re.findall(sample_pattern, all_text, re.IGNORECASE | re.DOTALL)
processed_count = 0
for match in matches:
if processed_count >= 12:
break
sample_num, sample_values = match
sample_id = f"L{str(sample_num).zfill(3)}"
numbers = [float(x) for x in re.findall(r'(\d+\.?\d*)', sample_values)]
if len(numbers) >= 5:
sample_data = _align_numbers_to_params(numbers, leaf_params, sample_id, kind='leaf')
if sample_data and any(v != 0.0 for v in sample_data.values()):
samples_data[sample_id] = sample_data
processed_count += 1
# Tabular number chunking for farm leaf only
if not samples_data and not is_sp_leaf:
all_numbers = [float(x) for x in re.findall(r'(\d+\.?\d*)', all_text)]
if len(all_numbers) >= 40:
for i in range(min(12, len(all_numbers) // 8)):
sample_id = f"L{i+1:03d}"
start_idx = i * 8
chunk = all_numbers[start_idx:start_idx + 8]
sample_data = _align_numbers_to_params(chunk, leaf_params, sample_id, kind='leaf')
if sample_data and any(v != 0.0 for v in sample_data.values()):
samples_data[sample_id] = sample_data
# Parameter-value pairs last resort
if not samples_data:
for param in leaf_params:
param_name = param.split()[0]
param_pattern = rf'{re.escape(param_name)}\s*[:=]\s*(\d+\.?\d*)'
matches = re.findall(param_pattern, all_text, re.IGNORECASE)
if matches:
if not samples_data:
samples_data["L001"] = {}
try:
samples_data["L001"][param] = float(matches[0])
except (ValueError, TypeError):
samples_data["L001"][param] = 0.0
if samples_data and not structured_data.get("SP_Lab_Test_Report"):
container_key = "SP_Lab_Test_Report" if is_sp_leaf else "Farm_Leaf_Test_Data"
if container_key not in structured_data:
structured_data[container_key] = {}
for sample_id, sample_data in samples_data.items():
if sample_data and any(
(isinstance(v, (int, float)) and v != 0.0) or (not isinstance(v, (int, float)) and v)
for v in sample_data.values()
):
structured_data[container_key][sample_id] = sample_data
# Final normalization: prefer populated SP Lab report when present
if structured_data.get("SP_Lab_Test_Report"):
return {"SP_Lab_Test_Report": structured_data["SP_Lab_Test_Report"]}
if structured_data.get("Farm_Soil_Test_Data"):
return {"Farm_Soil_Test_Data": structured_data["Farm_Soil_Test_Data"]}
if structured_data.get("Farm_Leaf_Test_Data"):
return {"Farm_Leaf_Test_Data": structured_data["Farm_Leaf_Test_Data"]}
# Empty but format-detected
if container_type.lower() == 'soil':
if is_sp_soil:
return {"SP_Lab_Test_Report": {}}
return {"Farm_Soil_Test_Data": {}}
if is_sp_leaf:
return {"SP_Lab_Test_Report": {}}
return {"Farm_Leaf_Test_Data": {}}
except Exception as e:
# Return empty structure on error - no hardcoded fallback data
if container_type.lower() == 'soil':
if 'SP LAB' in (raw_text or '').upper() or re.search(r'S\d{1,4}/\d{2}', raw_text or ''):
return {"SP_Lab_Test_Report": {}}
return {"Farm_Soil_Test_Data": {}}
if 'SP LAB' in (raw_text or '').upper() or re.search(r'P\d{1,4}/\d{2}', raw_text or ''):
return {"SP_Lab_Test_Report": {}}
return {"Farm_Leaf_Test_Data": {}}
def _flatten_ocr_sample(sample: dict, container_type: str) -> tuple:
"""Return (sample_id, flat_param_dict) from a Document AI / Excel sample object."""
if not isinstance(sample, dict):
return None, {}
sample_id = (
sample.get('sample_id')
or sample.get('Sample ID')
or sample.get('Sample No.')
or sample.get('Lab No.')
or sample.get('lab_no')
)
# Nested soil: {sample_id, data: {...}}
data = sample.get('data') if isinstance(sample.get('data'), dict) else None
# Nested leaf Document AI shape
dry = sample.get('% Dry Matter') if isinstance(sample.get('% Dry Matter'), dict) else None
mgkg = sample.get('mg/kg Dry Matter') if isinstance(sample.get('mg/kg Dry Matter'), dict) else None
flat = {}
if data:
for k, v in data.items():
if v is None or isinstance(v, dict):
continue
num = _parse_numeric_token(v) if not isinstance(v, (int, float)) else float(v)
flat[k] = num if num is not None else v
elif dry or mgkg:
# Flatten leaf nested structure into standard keys
key_map_dry = {
'N': 'N (%)', 'P': 'P (%)', 'K': 'K (%)', 'MG': 'Mg (%)', 'Mg': 'Mg (%)',
'CA': 'Ca (%)', 'Ca': 'Ca (%)',
}
key_map_mg = {
'B': 'B (mg/kg)', 'CU': 'Cu (mg/kg)', 'Cu': 'Cu (mg/kg)',
'ZN': 'Zn (mg/kg)', 'Zn': 'Zn (mg/kg)',
}
if dry:
for k, v in dry.items():
out_k = key_map_dry.get(str(k).upper() if len(str(k)) <= 2 else str(k), key_map_dry.get(str(k), None))
if not out_k:
# try direct
ku = str(k).upper()
out_k = {'N': 'N (%)', 'P': 'P (%)', 'K': 'K (%)', 'MG': 'Mg (%)', 'CA': 'Ca (%)'}.get(ku)
if out_k:
num = _parse_numeric_token(v) if not isinstance(v, (int, float)) else float(v)
if num is not None:
flat[out_k] = num
if mgkg:
for k, v in mgkg.items():
ku = str(k).upper()
out_k = {'B': 'B (mg/kg)', 'CU': 'Cu (mg/kg)', 'ZN': 'Zn (mg/kg)'}.get(ku)
if out_k:
num = _parse_numeric_token(v) if not isinstance(v, (int, float)) else float(v)
if num is not None:
flat[out_k] = num
else:
skip = {
'sample_id', 'Sample ID', 'Sample No.', 'Lab No.', 'sample_no', 'lab_no',
'data', '% Dry Matter', 'mg/kg Dry Matter',
}
for k, v in sample.items():
if k in skip or v is None or isinstance(v, dict):
continue
num = _parse_numeric_token(v) if not isinstance(v, (int, float)) else float(v)
flat[k] = num if num is not None else v
# Normalize common soil param aliases to compact keys
if container_type.lower() == 'soil' and flat:
alias = {
'Nitrogen (%)': 'N (%)',
'Organic Carbon (%)': 'Org. C (%)',
'Available P (mg/kg)': 'Avail P (mg/kg)',
'C.E.C (meq%)': 'CEC (meq%)',
'CEC (meq/100 g)': 'CEC (meq%)',
}
flat = {alias.get(k, k): v for k, v in flat.items()}
return (str(sample_id) if sample_id is not None else None), flat
def _score_structured_samples(samples: dict, container_type: str) -> float:
"""Score how realistic a structured sample dict is (detect shifted columns)."""
if not samples:
return -1e9
total = 0.0
count = 0
for data in samples.values():
if not isinstance(data, dict) or not data:
continue
if container_type.lower() == 'soil':
order = [
'pH', 'N (%)', 'Org. C (%)', 'Total P (mg/kg)', 'Avail P (mg/kg)',
'Exch. K (meq%)', 'Exch. Ca (meq%)', 'Exch. Mg (meq%)', 'CEC (meq%)',
]
# also accept expanded aliases already normalized
window = []
for p in order:
v = data.get(p)
if isinstance(v, (int, float)):
window.append(float(v))
else:
break
if len(window) >= 5:
total += _score_soil_window(window)
count += 1
else:
order = ['N (%)', 'P (%)', 'K (%)', 'Mg (%)', 'Ca (%)', 'B (mg/kg)', 'Cu (mg/kg)', 'Zn (mg/kg)']
window = []
for p in order:
v = data.get(p)
if isinstance(v, (int, float)):
window.append(float(v))
else:
break
if len(window) >= 5:
total += _score_leaf_window(window)
count += 1
if count == 0:
return -1e9
return total / count
def _structured_from_ocr_tables(ocr_result: dict, container_type: str) -> dict:
"""Build structured samples from Document AI / OCR table objects when present."""
if not ocr_result:
return {}
tables = ocr_result.get('tables') or []
samples_data = {}
for table in tables:
ttype = str(table.get('type', '')).lower()
if ttype not in (container_type.lower(), 'unknown', ''):
# still try unknown tables if empty so far
if ttype not in ('soil', 'leaf') and samples_data:
continue
if ttype in ('soil', 'leaf') and ttype != container_type.lower():
continue
for sample in table.get('samples') or []:
sid, flat = _flatten_ocr_sample(sample, container_type)
if sid and flat:
samples_data[sid] = flat
if not samples_data:
details = (ocr_result.get('raw_data') or {}).get('extraction_details') or {}
for key in ('SP_Lab_Test_Report', 'Farm_Soil_Test_Data', 'Farm_3_Soil_Test_Data', 'Farm_Leaf_Test_Data', 'leaf_data'):
block = details.get(key)
if isinstance(block, dict) and block.get('samples'):
for sample in block['samples']:
sid, flat = _flatten_ocr_sample(sample, container_type)
if sid and flat:
samples_data[sid] = flat
elif isinstance(block, dict) and block:
# map of sample_id -> data
for sid, data in block.items():
if sid in ('total_samples', 'sample_ids', 'parameters', 'report_type', 'samples'):
continue
if isinstance(data, dict):
_, flat = _flatten_ocr_sample({'sample_id': sid, **data} if 'data' not in data else {'sample_id': sid, 'data': data.get('data', data)}, container_type)
if not flat and isinstance(data, dict):
flat = {k: v for k, v in data.items() if not isinstance(v, dict)}
if flat:
samples_data[str(sid)] = flat
if not samples_data:
return {}
ids = list(samples_data.keys())
is_sp = any('/' in str(i) for i in ids) or 'SP LAB' in str(
(ocr_result.get('raw_data') or {}).get('text', '')
).upper()
if is_sp:
return {"SP_Lab_Test_Report": samples_data}
if container_type.lower() == 'soil':
return {"Farm_Soil_Test_Data": samples_data}
return {"Farm_Leaf_Test_Data": samples_data}
def _merge_structured_sources(text_struct: dict, table_struct: dict, raw_text: str, container_type: str) -> dict:
"""Pick the best structured source so output matches real OCR (tables or text).
Document AI header-mapped tables usually beat positional text parsing.
Bad text parses (shifted columns) must not overwrite good tables.
"""
text_key = next(iter(text_struct), None) if text_struct else None
table_key = next(iter(table_struct), None) if table_struct else None
text_samples = (text_struct or {}).get(text_key or '', {}) or {}
table_samples = (table_struct or {}).get(table_key or '', {}) or {}
text_score = _score_structured_samples(text_samples, container_type)
table_score = _score_structured_samples(table_samples, container_type)
# Prefer higher score; if close, prefer more samples
prefer_table = False
if table_samples and not text_samples:
prefer_table = True
elif text_samples and not table_samples:
prefer_table = False
elif table_samples and text_samples:
if table_score > text_score + 0.5:
prefer_table = True
elif text_score > table_score + 0.5:
prefer_table = False
else:
# tie-breaker: more complete params / more samples
def completeness(samples):
if not samples:
return 0
return sum(len(v) for v in samples.values() if isinstance(v, dict)) / max(len(samples), 1)
prefer_table = completeness(table_samples) >= completeness(text_samples)
else:
return text_struct or table_struct or {}
if prefer_table:
chosen_key = table_key or text_key or 'SP_Lab_Test_Report'
merged = dict(table_samples)
# Fill missing samples from text if text score is ok
if text_score > 0:
for sid, data in text_samples.items():
if sid not in merged:
merged[sid] = data
else:
chosen_key = text_key or table_key or 'SP_Lab_Test_Report'
merged = dict(text_samples)
if table_score > 0:
for sid, data in table_samples.items():
if sid not in merged:
merged[sid] = data
else:
# fill missing keys from table
for k, v in data.items():
if k not in merged[sid] or merged[sid][k] in (None, '', 0):
merged[sid][k] = v
return {chosen_key: merged}
def _show_raw_text_as_json(raw_text: str, container_type: str, ocr_result: dict = None) -> None:
"""Display structured OCR data that faithfully follows the uploaded document.
Priority: Document AI / Excel tables (header-mapped) when realistic, else raw-text
parse with sliding-window alignment. Never uses hardcoded sample templates.
"""
structured_data = format_raw_text_as_structured_json(raw_text, container_type)
try:
table_struct = _structured_from_ocr_tables(ocr_result or {}, container_type)
if table_struct:
structured_data = _merge_structured_sources(
structured_data, table_struct, raw_text, container_type
)
except Exception:
pass
# Final pass: for SP Lab IDs, re-parse raw text and keep the higher-scoring version
try:
text_only = format_raw_text_as_structured_json(raw_text, container_type)
structured_data = _merge_structured_sources(
text_only, structured_data, raw_text, container_type
)
except Exception:
pass
st.markdown(f"### 📝 {t('upload_raw_text', 'Raw Extracted Text Data')}")
st.markdown(f"#### 📊 {t('upload_structured_data', 'Structured OCR Data (JSON Format)')}")
st.markdown(f"**{t('upload_ai_analysis_note', 'This data will be used by the AI for analysis. Each sample ID contains its parameter values:')}**")
try:
import json
formatted_json = json.dumps(structured_data, indent=2, ensure_ascii=False)
st.code(formatted_json, language="json")
except Exception:
st.json(structured_data)
if container_type == 'soil':
st.session_state.structured_soil_data = structured_data
else:
st.session_state.structured_leaf_data = structured_data
st.info("💡 **AI Analysis Ready**: The structured data above will be used for comprehensive step-by-step analysis.")
st.markdown("#### 🔍 Raw OCR Text (Reference Only)")
st.code(raw_text, language="text")
st.caption(f"Raw text length: {len(raw_text)} characters | Container: {container_type}")
if __name__ == "__main__":
# Initialize session state defaults
if 'current_page' not in st.session_state:
st.session_state.current_page = 'upload'
if 'user_id' not in st.session_state:
st.session_state.user_id = None
if 'analysis_data' not in st.session_state:
st.session_state.analysis_data = {}
# Display the upload page
show_upload_page()
import streamlit as st
import sys
import os
from datetime import datetime
from PIL import Image
import json
import tempfile
import time
import re
import hashlib
import platform
# Add utils to path
utils_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'utils')
if utils_path not in sys.path:
sys.path.append(utils_path)
# Import utilities with error handling and robust fallbacks
UPLOAD_IMPORT_ERROR = None
try:
from utils.ocr_utils import extract_data_from_image
from utils.parsing_utils import _parse_raw_text_to_structured_json
from utils.analysis_engine import validate_soil_data, validate_leaf_data
from utils.parameter_standardizer import parameter_standardizer
from utils.config_manager import get_ui_config
except Exception:
try:
from ocr_utils import extract_data_from_image, validate_soil_data, validate_leaf_data
from parsing_utils import _parse_raw_text_to_structured_json
from parameter_standardizer import parameter_standardizer
from config_manager import get_ui_config
except Exception as e:
UPLOAD_IMPORT_ERROR = str(e)
extract_data_from_image = None
_parse_raw_text_to_structured_json = None
validate_soil_data = None
validate_leaf_data = None
parameter_standardizer = None
def get_ui_config():
return None
# Import translations
try:
from utils.translations import t
except ImportError:
try:
from translations import t
except ImportError:
# Fallback translation function
def t(key, default=None):
return default or key
def get_machine_id():
"""Generate a persistent machine identifier that persists across sessions"""
if 'machine_id' not in st.session_state:
machine_id_file = os.path.join(os.path.expanduser('~'), '.cropdrive_ai_machine_id')
machine_id = None
# Try to read existing machine ID from file
try:
if os.path.exists(machine_id_file):
with open(machine_id_file, 'r') as f:
machine_id = f.read().strip()
if machine_id and len(machine_id) == 16:
# Valid existing ID
st.session_state.machine_id = machine_id
return machine_id
except Exception:
pass
# Generate new machine ID based on system information
machine_info = {
'platform': platform.platform(),
'system': platform.system(),
'processor': platform.processor(),
'node': platform.node(), # Network name
}
# Create a hash from machine info
machine_string = json.dumps(machine_info, sort_keys=True)
machine_id = hashlib.md5(machine_string.encode()).hexdigest()[:16] # 16 char ID
# Save to file for persistence
try:
with open(machine_id_file, 'w') as f:
f.write(machine_id)
except Exception:
# If file write fails, still use the generated ID
pass
# Store in session state
st.session_state.machine_id = machine_id
return st.session_state.machine_id
def show_upload_page():
"""Main upload page - focused only on file upload and preview"""
# Import translations
try: