Spaces:
Sleeping
Sleeping
| from flask import Flask, render_template, request, jsonify | |
| import os | |
| import base64 | |
| import requests | |
| from PIL import Image | |
| from io import BytesIO | |
| import logging | |
| from requests.adapters import HTTPAdapter | |
| from urllib3.util.retry import Retry | |
| # Configure logging | |
| logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') | |
| app = Flask(__name__) | |
| # Configuration | |
| UPLOAD_FOLDER = 'static/captures' | |
| os.makedirs(UPLOAD_FOLDER, exist_ok=True) | |
| app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER | |
| # Hugging Face API settings | |
| HF_API_TOKEN = os.getenv('HF_API_TOKEN') # Load token from environment variable | |
| HF_API_URL = "https://api-inference.huggingface.co/models/facebook/detr-resnet-50" # Example model for object detection | |
| def query_hugging_face(image_data): | |
| if not HF_API_TOKEN: | |
| logging.error("Hugging Face API token not set.") | |
| return {"error": "Hugging Face API token not set. Please set the HF_API_TOKEN environment variable."} | |
| try: | |
| session = requests.Session() | |
| # Configure retries: 3 attempts, backoff factor of 1s, retry on 502/503/504 | |
| retries = Retry(total=3, backoff_factor=1, status_forcelist=[502, 503, 504]) | |
| session.mount('https://', HTTPAdapter(max_retries=retries)) | |
| headers = {"Authorization": f"Bearer {HF_API_TOKEN}"} | |
| logging.debug(f"Sending request to {HF_API_URL}") | |
| response = session.post(HF_API_URL, headers=headers, data=image_data, timeout=10) | |
| response.raise_for_status() | |
| logging.debug("Hugging Face API request successful.") | |
| return response.json() | |
| except requests.exceptions.RequestException as e: | |
| logging.error(f"Failed to connect to Hugging Face API: {str(e)}") | |
| return {"error": f"Failed to connect to Hugging Face API: {str(e)}"} | |
| def index(): | |
| return render_template('index.html') | |
| def capture(): | |
| try: | |
| # Get the base64 image data from the request | |
| data = request.form['image'] | |
| header, encoded = data.split(",", 1) | |
| binary_data = base64.b64decode(encoded) | |
| # Save the image | |
| filename = f"capture_{len(os.listdir(app.config['UPLOAD_FOLDER'])) + 1}.jpg" | |
| filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) | |
| with open(filepath, "wb") as f: | |
| f.write(binary_data) | |
| # Process with Hugging Face API | |
| with open(filepath, "rb") as f: | |
| hf_result = query_hugging_face(f.read()) | |
| # Return the image URL and Hugging Face result | |
| image_url = f"/{filepath}" | |
| return jsonify({ | |
| 'status': 'success', | |
| 'image_url': image_url, | |
| 'hf_result': hf_result | |
| }) | |
| except Exception as e: | |
| logging.error(f"Error in capture: {str(e)}") | |
| return jsonify({'status': 'error', 'message': str(e)}) | |
| def test_connectivity(): | |
| try: | |
| response = requests.get('https://www.google.com', timeout=5) | |
| logging.debug(f"Connectivity test to Google: {response.status_code}") | |
| return jsonify({'status': 'success', 'code': response.status_code}) | |
| except requests.exceptions.RequestException as e: | |
| logging.error(f"Connectivity test failed: {str(e)}") | |
| return jsonify({'status': 'error', 'message': str(e)}) | |
| if __name__ == '__main__': | |
| app.run(debug=True, host='0.0.0.0', port=7860) |