Upload 7 files

app.py

@@ -0,0 +1,269 @@
+import os
+import numpy as np
+from PIL import Image
+import requests
+import time
+import multiprocessing
+import json
+import sys
+from typing import Tuple, List, Dict, Any
+
+# Add Florence model path to Python path
+florence_path = os.path.join(os.path.dirname(__file__), 'florence-2-large')
+sys.path.append(florence_path)
+
+try:
+    from florence_2_large.processing_florence2 import Florence2Processor
+    from florence_2_large.configuration_florence2 import Florence2Config
+    import torch
+    import torch.nn.functional as F
+    
+    # Initialize processor with local files
+    config = Florence2Config.from_json_file(os.path.join(florence_path, 'config.json'))
+    processor = Florence2Processor(config)
+    HAVE_PROCESSOR = True
+    print("Successfully loaded Florence processor")
+except Exception as e:
+    print(f"Warning: Could not load Florence processor: {e}")
+    print("Using basic output interpretation")
+    HAVE_PROCESSOR = False
+
+# Task-specific configuration
+TASK = "<MORE_DETAILED_CAPTION>"  # For detailed image captioning
+
+# Model configuration
+MODEL_ID = "microsoft/florence-2-base"
+
+
+
+def load_and_preprocess_image(image_path):
+    # Load image and resize to 32x32
+    img = Image.open(image_path)
+    img = img.resize((32, 32))
+    
+    # Convert to numpy array and normalize to [0,1]
+    img_array = np.array(img).astype(np.float32) / 255.0
+    
+    # Ensure array has shape (32, 32, 3)
+    if len(img_array.shape) == 2:
+        img_array = np.stack([img_array] * 3, axis=-1)
+    
+    # Add batch dimension
+    img_array = img_array[np.newaxis, ...]
+    
+    # Convert tensor to list of single-element lists for API
+    tensor_data = [[float(x)] for x in img_array.flatten()]
+    
+    return tensor_data
+
+def run_inference(args: Tuple[str, str, int]) -> dict:
+    """Run inference on a specific server with given chunk ID."""
+    server_url, image_path, chunk_id = args
+    
+    try:
+        print(f"\nProcessing server {server_url} with chunk {chunk_id}...")
+        
+        # Load and preprocess image
+        input_tensor = load_and_preprocess_image(image_path)
+        
+        # Prepare request data
+        data = {
+            "inputs": input_tensor
+        }
+        
+        # Send request with timeout
+        print(f"Sending request to {server_url}/compute/{chunk_id}")
+        start_time = time.time()
+        response = requests.post(
+            f"{server_url}/compute/{chunk_id}",
+            json=data,
+            headers={"Content-Type": "application/json"},
+            timeout=10
+        )
+        
+        inference_time = time.time() - start_time
+        
+        if response.status_code == 200:
+            result = response.json()
+            return {
+                "server": server_url,
+                "chunk_id": chunk_id,
+                "success": True,
+                "time": inference_time,
+                "result": result
+            }
+        else:
+            error_msg = f"HTTP {response.status_code}"
+            if hasattr(response, 'text'):
+                error_msg += f": {response.text}"
+            return {
+                "server": server_url,
+                "chunk_id": chunk_id,
+                "success": False,
+                "error": error_msg,
+                "time": inference_time
+            }
+            
+    except Exception as e:
+        return {
+            "server": server_url,
+            "chunk_id": chunk_id,
+            "success": False,
+            "error": str(e),
+            "time": time.time() - start_time if 'start_time' in locals() else None
+        }
+
+def process_model_outputs(outputs, original_shape=(1, -1, 51289)):
+    """Process model outputs using Florence processor for sequence generation."""
+    # Convert outputs to numpy array
+    outputs_array = np.array([x[0] for x in outputs])
+    
+    if HAVE_PROCESSOR:
+        try:
+            # Reshape logits to [batch, seq_len, vocab_size]
+            logits = outputs_array.reshape(original_shape)
+            
+            if torch.is_tensor(logits):
+                # Use torch operations if available
+                token_ids = torch.argmax(logits, dim=-1)
+            else:
+                # Fallback to numpy
+                token_ids = np.argmax(logits, axis=-1)
+            
+            # Decode tokens to text
+            text = processor.batch_decode(token_ids, skip_special_tokens=True)
+            
+            # Post-process for the specific task
+            processed_text = processor.post_process_generation(
+                text[0] if isinstance(text, list) else text,
+                task=TASK
+            )
+            
+            return {
+                'text': processed_text,
+                'tokens': token_ids.tolist() if torch.is_tensor(token_ids) else token_ids.tolist(),
+                'logits_shape': logits.shape,
+                'distribution': {
+                    'min': float(outputs_array.min()),
+                    'max': float(outputs_array.max()),
+                    'mean': float(outputs_array.mean()),
+                    'std': float(outputs_array.std())
+                }
+            }
+        except Exception as e:
+            print(f"Warning: Error in sequence processing: {e}")
+    
+    # Fallback to basic statistics if processor not available
+    return {
+        'overall_mean': float(outputs_array.mean()),
+        'overall_std': float(outputs_array.std()),
+        'shape': outputs_array.shape,
+        'distribution': {
+            'min': float(outputs_array.min()),
+            'max': float(outputs_array.max()),
+            'median': float(np.median(outputs_array))
+        }
+    }
+
+def process_results(results):
+    """Process and combine results from all servers."""
+    # Filter successful results
+    successful_results = [r for r in results if r['success']]
+    if not successful_results:
+        print("\nError: No servers returned successful results")
+        return
+    
+    # Sort successful results by chunk ID
+    successful_results.sort(key=lambda x: x['chunk_id'])
+    
+    print(f"\nModel Output Analysis ({len(successful_results)}/{len(results)} servers succeeded):")
+    print("-" * 80)
+    
+    # Get total sequence length from all chunks
+    total_outputs = []
+    for result in successful_results:
+        total_outputs.extend(result['result']['outputs'])
+    
+    # Process the combined sequence
+    print("\nProcessing complete sequence...")
+    analysis = process_model_outputs(total_outputs, original_shape=(1, -1, 51289))
+    
+    if 'text' in analysis:
+        print("\nGenerated Description:")
+        print("-" * 80)
+        print(analysis['text'])
+        
+        print("\nSequence Statistics:")
+        print(f"- Logits shape: {analysis['logits_shape']}")
+        print(f"- Distribution:")
+        for key, value in analysis['distribution'].items():
+            print(f"  {key}: {value:.4f}")
+    else:
+        print("\nBasic Analysis (Florence processor not available):")
+        print(f"- Sequence length: {len(total_outputs)}")
+        print(f"- Overall activation: {analysis['overall_mean']:.4f} ± {analysis['overall_std']:.4f}")
+        print("\nValue Distribution:")
+        for key, value in analysis['distribution'].items():
+            print(f"- {key}: {value:.4f}")
+    
+    # Check server consistency
+    if len(successful_results) > 1:
+        all_outputs = [np.array([x[0] for x in r['result']['outputs']]) 
+                      for r in successful_results]
+        differences = [np.max(np.abs(all_outputs[0] - tensor)) 
+                      for tensor in all_outputs[1:]]
+        
+        print("\nServer Consistency:")
+        if np.max(differences) < 1e-6:
+            print("Successful servers provided identical results")
+        else:
+            print(f"Variations detected between servers (max diff: {np.max(differences):.6f})")
+    
+    # Print timing summary
+    successful_times = [r['time'] for r in successful_results]
+    print(f"\nProcessing Time Summary:")
+    print(f"- Average: {np.mean(successful_times):.2f}s")
+    print(f"- Range: {min(successful_times):.2f}s - {max(successful_times):.2f}s")
+
+def main():
+    # Server configurations with their respective chunk IDs
+    servers = [
+        ("https://fred808-ilob.hf.space", 0),
+        ("https://fred808-tserv.hf.space", 1),
+        ("https://fred808-tserve2.hf.space", 2)
+    ]
+    
+    # Image path - using the same image for all servers
+    image_path = "sample_task/test1.png"
+    print(f"\nTesting with image: {image_path}")
+    
+    # Create process pool
+    with multiprocessing.Pool() as pool:
+        # Prepare arguments for each server
+        args = [(server_url, image_path, chunk_id) for server_url, chunk_id in servers]
+        
+        # Run inference in parallel
+        print("\nStarting parallel inference across all servers...")
+        results = pool.map(run_inference, args)
+        
+        # Display individual server results
+        print("\nServer Results:")
+        print("-" * 80)
+        for result in results:
+            print(f"\nServer: {result['server']}")
+            print(f"Chunk ID: {result['chunk_id']}")
+            print(f"Success: {result['success']}")
+            print(f"Time: {result['time']:.4f}s" if result['time'] else "Time: N/A")
+            
+            if result['success']:
+                print(f"Output shape: {len(result['result']['outputs'])} elements")
+                print("First few outputs:", result['result']['outputs'][:5])
+            else:
+                print(f"Error: {result['error']}")
+            print("-" * 80)
+        
+        # Process and display combined results
+        process_results(results)
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,5 @@
+numpy
+torch
+aiohttp
+asyncio
+transformers