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'

๐Ÿ“ค {t("upload_title")}

', 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 / '): 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' 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'= 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\.?| 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()