Add complete scanner-workflow project with auto-scan functionality
This commit is contained in:
quiet
261
scanner/scanner-workflow/scanner-auto.py
Normal file
261
scanner/scanner-workflow/scanner-auto.py
Normal file
@@ -0,0 +1,261 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Auto-scanner script for Brother DS mobile scanners.
|
||||
Auto-detects when a document is placed and starts scanning automatically.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
import time
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict
|
||||
|
||||
# Add parent directory to path for config import
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
try:
|
||||
import requests
|
||||
from dotenv import load_dotenv
|
||||
import yaml
|
||||
except ImportError as e:
|
||||
print(f"Missing dependency: {e}")
|
||||
print("Run: pip install -r requirements.txt")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
class ScannerAuto:
|
||||
"""Automated Brother scanner with LLM-powered naming."""
|
||||
|
||||
def __init__(self, config_path: str = "config.yml"):
|
||||
self.load_config(config_path)
|
||||
self.scan_count = 0
|
||||
|
||||
def load_config(self, config_path: str):
|
||||
"""Load configuration from YAML file."""
|
||||
with open(config_path, 'r') as f:
|
||||
config = yaml.safe_load(f)
|
||||
|
||||
self.scan_dir = Path(config.get('scan_dir', 'scans'))
|
||||
self.scan_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.brother_cmd = config.get('brother_cmd', 'brscan-skey -s')
|
||||
self.api_url = config.get('api_url', 'http://localhost:11434/v1/chat/completions')
|
||||
self.api_key = config.getenv('API_KEY', '')
|
||||
self.model = config.get('model', 'llama3')
|
||||
|
||||
self.log_file = self.scan_dir / 'scan_log.json'
|
||||
|
||||
def log_scan(self, original_name: str, final_name: str):
|
||||
"""Log scan metadata."""
|
||||
log_entry = {
|
||||
'timestamp': datetime.now().isoformat(),
|
||||
'original_name': original_name,
|
||||
'final_name': final_name
|
||||
}
|
||||
|
||||
if self.log_file.exists():
|
||||
with open(self.log_file, 'r') as f:
|
||||
logs = json.load(f)
|
||||
else:
|
||||
logs = []
|
||||
|
||||
logs.append(log_entry)
|
||||
|
||||
with open(self.log_file, 'w') as f:
|
||||
json.dump(logs, f, indent=2)
|
||||
|
||||
def detect_document(self) -> bool:
|
||||
"""
|
||||
Detect if a document is placed in the scanner.
|
||||
Returns True if document detected, False otherwise.
|
||||
"""
|
||||
try:
|
||||
# Brother scanner detection
|
||||
result = subprocess.run(
|
||||
self.brother_cmd,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
|
||||
# Check if scanner reports a document is ready
|
||||
# This may vary based on Brother scanner model and tools
|
||||
if result.returncode == 0 and ('ready' in result.stdout.lower() or 'scan' in result.stdout.lower()):
|
||||
print(f"✓ Document detected by scanner")
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
print("✗ Scanner timeout")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ Scanner detection error: {e}")
|
||||
return False
|
||||
|
||||
def start_scan(self) -> Optional[str]:
|
||||
"""Start scanning and return the saved filename."""
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
|
||||
# Create output filename (temporary until LLM generates title)
|
||||
temp_name = f"scan_{timestamp}.pdf"
|
||||
output_path = self.scan_dir / temp_name
|
||||
|
||||
try:
|
||||
# Start scanning using Brother CLI
|
||||
scan_cmd = f"{self.brother_cmd} -f {output_path}"
|
||||
print(f"→ Starting scan: {scan_cmd}")
|
||||
|
||||
result = subprocess.run(
|
||||
scan_cmd,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60 # 1 minute timeout for scan
|
||||
)
|
||||
|
||||
if result.returncode == 0 and output_path.exists():
|
||||
file_size = output_path.stat().st_size
|
||||
print(f"✓ Scan completed: {temp_name} ({file_size} bytes)")
|
||||
|
||||
# Use LLM to generate meaningful title
|
||||
title = self.generate_title(output_path)
|
||||
|
||||
# Rename with final title
|
||||
final_name = f"{title} - {timestamp}.pdf"
|
||||
final_path = self.scan_dir / final_name
|
||||
|
||||
output_path.rename(final_path)
|
||||
self.log_scan(temp_name, final_name)
|
||||
|
||||
print(f"✓ Saved as: {final_name}")
|
||||
return final_name
|
||||
|
||||
else:
|
||||
print(f"✗ Scan failed: {result.stderr}")
|
||||
return None
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
print("✗ Scan timeout")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"✗ Scan error: {e}")
|
||||
return None
|
||||
|
||||
def generate_title(self, pdf_path: Path) -> str:
|
||||
"""Generate a meaningful title using LLM."""
|
||||
if not self.api_key:
|
||||
# Fallback to basic naming if no API key
|
||||
return "document"
|
||||
|
||||
try:
|
||||
# Read PDF content (first 10 pages, or just metadata)
|
||||
# This is a simplified version - in production you might want to use pdfminer or similar
|
||||
print("→ Generating title with LLM...")
|
||||
|
||||
# Simple prompt for LLM
|
||||
prompt = f"""
|
||||
Analyze this document and suggest a concise, descriptive title (no more than 5 words).
|
||||
Focus on the document type (invoice, receipt, contract, letter, etc.).
|
||||
Return only the title, no other text.
|
||||
"""
|
||||
|
||||
response = requests.post(
|
||||
self.api_url,
|
||||
headers={
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json"
|
||||
},
|
||||
json={
|
||||
"model": self.model,
|
||||
"messages": [
|
||||
{"role": "system", "content": "You are a helpful assistant that generates document titles."},
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
"max_tokens": 50,
|
||||
"temperature": 0.3
|
||||
},
|
||||
timeout=30
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
title = response.json()['choices'][0]['message']['content'].strip()
|
||||
# Clean up title (remove quotes, extra whitespace)
|
||||
title = title.strip('"\'').strip()
|
||||
print(f"✓ LLM title: {title}")
|
||||
return title
|
||||
else:
|
||||
print(f"✗ LLM API error: {response.status_code}")
|
||||
return "document"
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ LLM error: {e}")
|
||||
return "document"
|
||||
|
||||
def run(self, max_scans: int = None):
|
||||
"""Main loop - auto-detects and scans documents."""
|
||||
print("=" * 60)
|
||||
print("🤖 Brother Scanner - Auto-Detect Mode")
|
||||
print("=" * 60)
|
||||
print(f"📁 Scan directory: {self.scan_dir}")
|
||||
print(f"🔄 Brother command: {self.brother_cmd}")
|
||||
print(f"🧠 LLM API: {self.api_url}")
|
||||
print("=" * 60)
|
||||
|
||||
if max_scans:
|
||||
print(f"⏱️ Max scans: {max_scans}")
|
||||
print("=" * 60)
|
||||
|
||||
self.scan_count = 0
|
||||
|
||||
try:
|
||||
while True:
|
||||
if max_scans and self.scan_count >= max_scans:
|
||||
print(f"\n✓ Completed {max_scans} scans")
|
||||
break
|
||||
|
||||
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Waiting for document...")
|
||||
|
||||
# Wait for document detection
|
||||
while not self.detect_document():
|
||||
time.sleep(2)
|
||||
|
||||
# Document detected - start scanning
|
||||
self.scan_count += 1
|
||||
self.start_scan()
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print(f"\n\n✓ Stopped after {self.scan_count} scans")
|
||||
except Exception as e:
|
||||
print(f"\n✗ Error: {e}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description='Auto-scan Brother scanner with LLM naming')
|
||||
parser.add_argument('--config', default='config.yml', help='Config file path')
|
||||
parser.add_argument('--max-scans', type=int, help='Maximum number of scans')
|
||||
parser.add_argument('--test', action='store_true', help='Test detection without scanning')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
scanner = ScannerAuto(args.config)
|
||||
|
||||
if args.test:
|
||||
print("🔍 Testing scanner detection...")
|
||||
if scanner.detect_document():
|
||||
print("✓ Scanner detected")
|
||||
else:
|
||||
print("✗ No document detected")
|
||||
else:
|
||||
scanner.run(args.max_scans)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user