Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| import os | |
| import json | |
| import shutil | |
| import tempfile | |
| from app_kit.config import load_app_config | |
| from app_kit.demo_packs import load_demo_pack | |
| from app_kit.logging_utils import setup_logging | |
| from app_kit.model_registry import load_model_registry | |
| from app_kit.project import ProjectSpec | |
| from app_kit.storage import SQLiteStore | |
| from app_kit.tracing import utc_now, write_trace_artifact | |
| THEME_CSS_PATH = Path(__file__).resolve().parents[2] / "assets" / "theme.css" | |
| class AppRuntime: | |
| spec: ProjectSpec | |
| config: object | |
| store: SQLiteStore | |
| registry: dict | |
| def run_pack_with_trace(spec: ProjectSpec, store: SQLiteStore, config: object, path: str): | |
| demo_pack = load_demo_pack(path) | |
| started_at = utc_now() | |
| output = spec.run_pack(demo_pack, store, config) | |
| finished_at = utc_now() | |
| trace_path = write_trace_artifact( | |
| config.artifact_dir, | |
| { | |
| 'kind': 'app-load', | |
| 'project': spec.key, | |
| 'pack_id': demo_pack.pack_id, | |
| 'pack_path': str(path), | |
| 'started_at': started_at, | |
| 'finished_at': finished_at, | |
| 'result': output, | |
| }, | |
| ) | |
| return output, f'✅ Loaded **{demo_pack.pack_id}** successfully! Trace artifact written.', trace_path | |
| def _format_pipeline_result(output: dict | list | None) -> str: | |
| """Format pipeline result as readable Markdown instead of raw JSON.""" | |
| if not output: | |
| return "" | |
| if isinstance(output, list): | |
| output = output[0] if output else {} | |
| lines = [] | |
| triage_icons = {'urgent': '🔴 URGENT', 'important': '🟡 IMPORTANT', 'FYI': '🟢 FYI'} | |
| if any(key in output for key in ('waste_category', 'suggestions', 'generation_stats', 'model_report', 'receipt_table')): | |
| lines.append('### 🧾 Household Food Waste Report') | |
| lines.append('') | |
| model_name = output.get('model_name') or output.get('model_id') or '' | |
| if model_name: | |
| lines.append(f"**Model:** `{model_name}`") | |
| generation_stats = output.get('generation_stats') or {} | |
| if generation_stats: | |
| backend = generation_stats.get('backend', '') | |
| adapter_name = generation_stats.get('adapter_name', '') | |
| elapsed = generation_stats.get('elapsed_seconds', '') | |
| total_tokens = generation_stats.get('total_tokens', '') | |
| lines.append( | |
| f"**Inference:** `{backend}` via `{adapter_name}` — {total_tokens} token(s), {elapsed} sec" | |
| ) | |
| if generation_stats.get('model_path'): | |
| lines.append(f"**Model path:** `{generation_stats['model_path']}`") | |
| if generation_stats.get('receipt_model_id'): | |
| lines.append(f"**Receipt model:** `{generation_stats['receipt_model_id']}`") | |
| receipt_adapter = output.get('receipt_adapter') or {} | |
| if receipt_adapter: | |
| lines.append( | |
| f"**Receipt adapter:** `{receipt_adapter.get('adapter_kind', 'unknown')}` · base `{receipt_adapter.get('base_model', 'unknown')}`" | |
| ) | |
| if receipt_adapter.get('artifact_path'): | |
| lines.append(f"**Adapter artifact:** `{receipt_adapter['artifact_path']}`") | |
| lines.append('') | |
| waste_category = output.get('waste_category', '') | |
| if waste_category: | |
| lines.append(f"**Waste category:** {waste_category.replace('_', ' ')}") | |
| model_report = output.get('model_report') or {} | |
| overbought_category = model_report.get('overbought_category', output.get('reconciliation', {}).get('model_overbought_category')) | |
| if overbought_category: | |
| lines.append(f"**Overbought category:** {str(overbought_category).replace('_', ' ')}") | |
| commitment_sentence = output.get('commitment_sentence', '') | |
| if commitment_sentence: | |
| lines.append(f"**Commitment:** {commitment_sentence}") | |
| spend = output.get('waste_spend_kpis') or {} | |
| if spend: | |
| lines.append('') | |
| lines.append('### 💰 Spend summary') | |
| lines.append(f"- Total spend: ${spend.get('total_spend', 0):.2f}") | |
| lines.append(f"- Estimated wasted spend: ${spend.get('approx_wasted_spend', 0):.2f}") | |
| lines.append(f"- Waste share: {spend.get('waste_share', 0):.2%}") | |
| suggestions = output.get('suggestions') or [] | |
| if suggestions: | |
| lines.append('') | |
| lines.append('### ✅ Suggestions') | |
| for suggestion in suggestions: | |
| lines.append(f'- {suggestion}') | |
| detected_categories = output.get('detected_categories') or [] | |
| if detected_categories: | |
| lines.append('') | |
| lines.append('### 📊 Model-backed categories') | |
| for item in detected_categories[:5]: | |
| category = str(item.get('category', 'unknown')).replace('_', ' ') | |
| count = item.get('count', 0) | |
| evidence = item.get('evidence') or [] | |
| if evidence: | |
| lines.append(f'- {category}: {count} — evidence: {", ".join(map(str, evidence[:2]))}') | |
| else: | |
| lines.append(f'- {category}: {count}') | |
| receipt_table = output.get('receipt_table') or [] | |
| if receipt_table: | |
| lines.append('') | |
| lines.append('### 🧾 Parsed receipt rows') | |
| lines.append('| item | category | qty | total | source |') | |
| lines.append('|---|---:|---:|---:|---|') | |
| for row in receipt_table[:5]: | |
| item = str(row.get('canonical_item') or row.get('item') or '').replace('|', '\\|') | |
| category = str(row.get('canonical_category') or row.get('category') or '').replace('|', '\\|') | |
| qty = row.get('qty', '') | |
| total = row.get('total_price', '') | |
| source = str(row.get('source_label') or '').replace('|', '\\|') | |
| lines.append(f'| {item} | {category} | {qty} | {total} | {source} |') | |
| if len(receipt_table) > 5: | |
| lines.append(f'*{len(receipt_table) - 5} more receipt row(s) hidden.*') | |
| return '\n'.join(lines) | |
| # Triage badge | |
| triage = output.get('triage', '') | |
| triage_display = triage_icons.get(triage, triage.upper()) | |
| lines.append(f"### {triage_display}") | |
| lines.append("") | |
| # Summary | |
| summary = output.get('summary', '') | |
| if summary: | |
| lines.append(f"**Summary:** {summary}") | |
| lines.append("") | |
| # Q&A Section | |
| qa = output.get('qa', []) | |
| if qa: | |
| lines.append("---") | |
| lines.append("### 📋 Document Analysis") | |
| for item in qa: | |
| q = item.get('question', '') | |
| a = item.get('answer', 'not stated') | |
| icon = '✅' if a != 'not stated' else '❔' | |
| lines.append(f"- {icon} **{q}**") | |
| lines.append(f" > {a}") | |
| lines.append("") | |
| # File info | |
| file_type = output.get('file_type', '') | |
| source_file = output.get('source_file', '') | |
| title = output.get('title', '') | |
| if title or file_type: | |
| lines.append("---") | |
| lines.append(f"📄 **Document:** {title} ({file_type})") | |
| # Inbox items | |
| inbox_items = output.get('inbox_items', []) | |
| if inbox_items and len(inbox_items) > 1: | |
| lines.append("") | |
| lines.append("### 📥 Processed Documents") | |
| for item in inbox_items: | |
| t = item.get('triage', '') | |
| badge = triage_icons.get(t, t) | |
| lines.append(f"- {badge} **{item.get('title', 'Untitled')}** — {item.get('summary', '')[:120]}") | |
| return "\n".join(lines) | |
| def _format_search_results(results: list | None) -> str: | |
| """Format search results as readable Markdown.""" | |
| if not results: | |
| return "*No results found. Try a different search query.*" | |
| lines = ["### 🔍 Search Results", ""] | |
| for i, result in enumerate(results, 1): | |
| title = result.get('title', 'Untitled') | |
| text = result.get('primary_text', '')[:200] | |
| status = result.get('status', '') | |
| lines.append(f"**{i}. {title}** `{status}`") | |
| lines.append(f"> {text}") | |
| lines.append("") | |
| return "\n".join(lines) | |
| def _format_history(records: list | None) -> str: | |
| """Format history/inbox as readable Markdown.""" | |
| if not records: | |
| return "*No records yet. Upload a document to get started.*" | |
| lines = ["### 📥 Document History", ""] | |
| triage_icons = {'urgent': '🔴', 'important': '🟡', 'FYI': '🟢'} | |
| for record in records: | |
| title = record.get('title', 'Untitled') | |
| created = record.get('created_at', '')[:19] | |
| try: | |
| blob = json.loads(record.get('json_blob', '{}')) if isinstance(record.get('json_blob'), str) else record.get('json_blob', {}) | |
| except Exception: | |
| blob = {} | |
| triage = blob.get('triage', '') | |
| icon = triage_icons.get(triage, '📄') | |
| summary = blob.get('summary', record.get('primary_text', ''))[:150] | |
| lines.append(f"{icon} **{title}** — `{created}`") | |
| lines.append(f"> {summary}") | |
| lines.append("") | |
| return "\n".join(lines) | |
| def _uploaded_file_kind(path: Path) -> str: | |
| suffix = path.suffix.lower() | |
| if suffix in {'.txt', '.md', '.json', '.yaml', '.yml', '.csv'}: | |
| return 'text' | |
| if suffix in {'.png', '.jpg', '.jpeg', '.webp', '.gif'}: | |
| return 'image' | |
| if suffix == '.pdf': | |
| return 'pdf' | |
| return 'file' | |
| def _uploaded_file_label(path: Path) -> str: | |
| label = path.stem.strip() | |
| return label or path.name | |
| def _write_upload_manifest(temp_dir: Path, file_paths: list[Path], spec: ProjectSpec) -> Path: | |
| manifest = { | |
| 'project': spec.key, | |
| 'pack_id': temp_dir.name, | |
| 'description': f'Uploaded {spec.key} documents', | |
| 'inputs': [ | |
| { | |
| 'path': path.name, | |
| 'kind': _uploaded_file_kind(path), | |
| 'label': _uploaded_file_label(path), | |
| } | |
| for path in file_paths | |
| ], | |
| } | |
| manifest_path = temp_dir / 'manifest.json' | |
| manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding='utf-8') | |
| return manifest_path | |
| def _process_uploaded_files(files, spec, store, config): | |
| """Process uploaded files through the pipeline.""" | |
| from app_kit.demo_packs import DemoPack | |
| if not files: | |
| return "⚠️ No files uploaded.", "Please upload one or more documents." | |
| # Copy uploaded files to a temp directory that looks like a demo pack | |
| temp_dir = Path(tempfile.mkdtemp(prefix="upload_")) | |
| file_paths = [] | |
| for f in files: | |
| src = Path(f) | |
| dst = temp_dir / src.name | |
| shutil.copy2(src, dst) | |
| file_paths.append(dst) | |
| _write_upload_manifest(temp_dir, file_paths, spec) | |
| try: | |
| # Build a minimal DemoPack-like structure | |
| demo_pack = load_demo_pack(str(temp_dir)) | |
| started_at = utc_now() | |
| output = spec.run_pack(demo_pack, store, config) | |
| finished_at = utc_now() | |
| write_trace_artifact( | |
| config.artifact_dir, | |
| { | |
| 'kind': 'app-upload', | |
| 'project': spec.key, | |
| 'pack_id': demo_pack.pack_id, | |
| 'file_count': len(file_paths), | |
| 'started_at': started_at, | |
| 'finished_at': finished_at, | |
| 'result': output, | |
| }, | |
| ) | |
| formatted = _format_pipeline_result(output) | |
| status = f"✅ Processed {len(file_paths)} document(s) successfully." | |
| return formatted, status | |
| except Exception as e: | |
| return f"❌ **Error processing documents:** {e}", f"Error: {e}" | |
| def run_app(spec: ProjectSpec) -> int: | |
| import gradio as gr | |
| config = load_app_config(spec.key) | |
| logger = setup_logging(spec.key) | |
| registry = load_model_registry(config.model_registry_path) | |
| logger.info('%s app listening', spec.key.upper()) | |
| store = SQLiteStore(config.sqlite_path, config.artifact_dir) | |
| # Friendly titles | |
| display_titles = { | |
| 'p1': ('Elder Care Document Assistant', 'Upload documents to get instant triage, summaries, and action items for elderly care paperwork.'), | |
| 'p4': ('Household Food Waste Tracker', 'Upload receipts and fridge notes to generate waste analysis reports.'), | |
| } | |
| display_title = display_titles.get(spec.key, (spec.title, spec.description)) | |
| with gr.Blocks(title=display_title[0], css_paths=THEME_CSS_PATH) as demo: | |
| # Header | |
| gr.Markdown(f"""# {display_title[0]} | |
| {display_title[1]}""") | |
| with gr.Tabs(): | |
| with gr.Tab("📁 App Workspace"): | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| # File upload area | |
| file_upload = gr.File( | |
| label="📂 Upload Documents", | |
| file_count="multiple", | |
| file_types=[".pdf", ".png", ".jpg", ".jpeg", ".txt", ".md", ".json", ".csv"], | |
| type="filepath", | |
| elem_classes=["upload-area"], | |
| ) | |
| status_display = gr.Markdown( | |
| value="*Upload documents above to get started.*", | |
| elem_classes=["status-box"], | |
| ) | |
| upload_btn = gr.Button("📤 Process Documents", variant="primary", size="lg") | |
| with gr.Column(scale=3): | |
| # Pipeline result display | |
| result_display = gr.Markdown( | |
| value="### 👋 Welcome\nUpload a PDF, image, or text document to see the AI-powered triage and analysis.", | |
| elem_classes=["result-card"], | |
| ) | |
| gr.Markdown("---") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| search_query = gr.Textbox( | |
| label="🔍 Search Documents", | |
| placeholder="Type a keyword to search your document history...", | |
| elem_classes=["search-box"], | |
| ) | |
| search_btn = gr.Button("Search", variant="secondary") | |
| with gr.Column(scale=2): | |
| search_result_display = gr.Markdown( | |
| value="*Enter a search query to find documents.*", | |
| elem_classes=["result-card"], | |
| ) | |
| gr.Markdown("---") | |
| # History section | |
| gr.Markdown("### 📋 Document History") | |
| history_display = gr.Markdown( | |
| value="*No documents processed yet.*", | |
| elem_classes=["history-card"], | |
| ) | |
| refresh_btn = gr.Button("🔄 Refresh History", variant="secondary") | |
| with gr.Tab("📖 How It Works"): | |
| if spec.key == 'p1': | |
| gr.Markdown( | |
| """ | |
| ### How to use the Elder Care Document Assistant | |
| 1. **Upload Documents:** Drag and drop or click the **Upload Documents** area to upload paperwork, medical receipts, invoices, or letters related to elder care (supports PDF, images, text). | |
| 2. **Process:** Click the **Process Documents** button. The local AI agent will parse the text, assign a triage level (e.g., `🔴 URGENT`, `🟡 IMPORTANT`, `🟢 FYI`), extract a concise summary, and answer relevant clinical or administrative questions. | |
| 3. **View Results:** The AI output will be displayed immediately as a formatted card. | |
| 4. **Search and Reference:** Use the **Search Documents** feature to search past logs by query keyword. Click **Refresh History** to fetch the full database history of processed files. | |
| *All data is stored and processed locally on your offline device for compliance and privacy.* | |
| """ | |
| ) | |
| else: # p4 | |
| gr.Markdown( | |
| """ | |
| ### How to use the Household Food Waste Tracker | |
| 1. **Upload Grocery Data:** Drag and drop or browse shopping receipts, food inventory CSVs, or daily logs of discarded food. | |
| 2. **Analyze Waste:** Click the **Process Documents** button to analyze purchases, flag high-risk perishables, estimate shelf-lives, and generate a household food conservation summary. | |
| 3. **View Diagnostics:** Review the formatted report detailing waste trends, warnings, and sustainability tips. | |
| 4. **Search & History:** Retrieve previous inventory reviews using the **Search** box, and click **Refresh History** to list your cumulative food waste entries. | |
| *Processes data locally to ensure household privacy and secure offline storage.* | |
| """ | |
| ) | |
| # Event handlers | |
| def handle_upload(files): | |
| if not files: | |
| return "### 👋 Welcome\nUpload a PDF, image, or text document to see the AI-powered triage and analysis.", "*Please upload at least one file.*" | |
| formatted, status = _process_uploaded_files(files, spec, store, config) | |
| return formatted, status | |
| def refresh_history(_=None): | |
| records = store.history(spec.key) | |
| return _format_history(records) | |
| def search_history(query: str): | |
| if not spec.search_enabled: | |
| return "*Search is not enabled for this project.*" | |
| if not query.strip(): | |
| return "*Enter a search query to find documents.*" | |
| results = store.search_records(spec.key, query) | |
| return _format_search_results(results) | |
| upload_btn.click( | |
| handle_upload, | |
| inputs=[file_upload], | |
| outputs=[result_display, status_display], | |
| ) | |
| refresh_btn.click(refresh_history, inputs=[], outputs=[history_display]) | |
| search_btn.click(search_history, inputs=[search_query], outputs=[search_result_display]) | |
| gr.Markdown("---") | |
| gr.Markdown( | |
| "### 🤖 Powered by Model Inference\n" | |
| "This application uses **MiniCPM-5-1B** for coaching, a **Real LoRA** linear adapter for NeMoTRON-PARS receipt parsing, and **all-MiniLM-L6-v2** for semantic search. All fallback and deterministic paths have been strictly removed." | |
| ) | |
| server_name = os.environ.get('GRADIO_SERVER_NAME', '0.0.0.0') | |
| server_port = int(os.environ.get('PORT', '7860')) | |
| demo.launch( | |
| server_name=server_name, | |
| show_error=True, | |
| share=False, | |
| ) | |
| return 0 | |