SenseNova-Vision Unified Vision Model Deep Dive: One Model for Detection, Segmentation, Depth, 3D Reconstruction — Zero Architecture Changes, Surpassing All Specialized Models

"""
SenseNova-Vision Corpus: Unified task format conversion.
Transforms heterogeneous CV annotations into instruction-response pairs.
"""

import json
import numpy as np
from typing import Dict, List, Tuple, Union
from PIL import Image
import base64
from io import BytesIO

class UnifiedTaskFormatter:
    """
    Converts task-specific CV annotations into unified instruction-response format.
    Supports: detection, segmentation, depth, normal, OCR, 3D reconstruction, pose.
    """
    
    def __init__(self, image_size: Tuple[int, int] = (1024, 1024)):
        self.image_size = image_size
    
    def detection_to_instruction(
        self, image_path: str, bboxes: List[Dict],
        task_prompt: str = "Detect all objects in the image and output coordinates"
    ) -> Dict:
        """Convert COCO-style detection annotations to instruction-response pair."""
        response_parts = []
        for bbox in bboxes:
            x1, y1, x2, y2 = bbox['bbox']
            label = bbox['category_name']
            response_parts.append(f"{label} at [{x1:.3f},{y1:.3f},{x2:.3f},{y2:.3f}]")
        
        return {
            "image": image_path,
            "instruction": task_prompt,
            "response": ", ".join(response_parts),
            "task_type": "detection",
            "output_mode": "text"
        }
    
    def segmentation_to_instruction(
        self, image_path: str, mask: np.ndarray, label: str,
        task_prompt: str = "Segment the {label} in the image"
    ) -> Dict:
        """Convert segmentation mask to instruction-response pair."""
        mask_img = Image.fromarray((mask * 255).astype(np.uint8))
        buffer = BytesIO()
        mask_img.save(buffer, format='PNG')
        mask_b64 = base64.b64encode(buffer.getvalue()).decode()
        
        return {
            "image": image_path,
            "instruction": task_prompt.format(label=label),
            "response": f"<image>{mask_b64}</image>",
            "response_image": mask_b64,
            "task_type": "segmentation",
            "output_mode": "image"
        }
    
    def depth_to_instruction(
        self, image_path: str, depth_map: np.ndarray,
        task_prompt: str = "Estimate the depth map of this image"
    ) -> Dict:
        """Convert depth map to instruction-response pair."""
        depth_norm = ((depth_map - depth_map.min()) / 
                     (depth_map.max() - depth_map.min() + 1e-8) * 255)
        depth_img = Image.fromarray(depth_norm.astype(np.uint8))
        buffer = BytesIO()
        depth_img.save(buffer, format='PNG')
        depth_b64 = base64.b64encode(buffer.getvalue()).decode()
        
        return {
            "image": image_path,
            "instruction": task_prompt,
            "response": f"<image>{depth_b64}</image>",
            "response_image": depth_b64,
            "task_type": "depth_estimation",
            "output_mode": "image"
        }

import torch
from torch.utils.data import DataLoader, Dataset
from transformers import AutoModelForVision2Seq, AutoProcessor
from typing import Dict, List
from PIL import Image

class UnifiedVisionDataset(Dataset):
    """Unified dataset for all vision tasks."""
    
    def __init__(self, unified_data: List[Dict], processor):
        self.data = unified_data
        self.processor = processor
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        item = self.data[idx]
        image = Image.open(item['image']).convert('RGB')
        text = f"User: {item['instruction']}\nAssistant: {item['response']}"
        
        inputs = self.processor(
            images=image, text=text,
            return_tensors="pt", padding="max_length", max_length=2048,
        )
        
        return {
            'pixel_values': inputs['pixel_values'].squeeze(0),
            'input_ids': inputs['input_ids'].squeeze(0),
            'attention_mask': inputs['attention_mask'].squeeze(0),
            'labels': inputs['input_ids'].squeeze(0).clone(),
        }


class SenseNovaVisionTrainer:
    """Trainer — no architectural changes, pure data-driven unification."""
    
    def __init__(self, model_name: str = "OpenSenseNova/Bagel-7B-MoT",
                 learning_rate: float = 2e-5, batch_size: int = 32):
        self.model = AutoModelForVision2Seq.from_pretrained(
            model_name, torch_dtype=torch.bfloat16, device_map="auto"
        )
        self.processor = AutoProcessor.from_pretrained(model_name)
        self.optimizer = torch.optim.AdamW(self.model.parameters(), lr=learning_rate)
        self.batch_size = batch_size
    
    def train_epoch(self, dataloader: DataLoader) -> float:
        self.model.train()
        total_loss = 0.0
        
        for batch in dataloader:
            pixel_values = batch['pixel_values'].to(self.model.device)
            input_ids = batch['input_ids'].to(self.model.device)
            attention_mask = batch['attention_mask'].to(self.model.device)
            labels = batch['labels'].to(self.model.device)
            
            outputs = self.model(
                pixel_values=pixel_values, input_ids=input_ids,
                attention_mask=attention_mask, labels=labels,
            )
            
            loss = outputs.loss
            loss.backward()
            self.optimizer.step()
            self.optimizer.zero_grad()
            total_loss += loss.item()
        
        return total_loss / len(dataloader)

package sensenova

import (
	"image"
	"strings"
)

// DetectionResult represents parsed detection output
type DetectionResult struct {
	Label string
	BBox  [4]float64
}

// ParseDetectionOutput parses text output for detection
func ParseDetectionOutput(text string) []DetectionResult {
	var results []DetectionResult
	parts := strings.Split(text, ", ")
	
	for _, part := range parts {
		openBracket := strings.Index(part, "[")
		closeBracket := strings.Index(part, "]")
		if openBracket < 0 || closeBracket <= openBracket {
			continue
		}
		
		label := strings.TrimSpace(part[:openBracket])
		label = strings.TrimRight(label, " at")
		
		coordStr := part[openBracket+1 : closeBracket]
		coords := strings.Split(coordStr, ",")
		if len(coords) != 4 {
			continue
		}
		
		var bbox [4]float64
		for i, c := range coords {
			val, _ := parseFloat(strings.TrimSpace(c))
			bbox[i] = val
		}
		results = append(results, DetectionResult{Label: label, BBox: bbox})
	}
	return results
}

// DepthMapResult represents parsed depth estimation output
type DepthMapResult struct {
	DepthMap [][]float32
}

// DecodeDepthMap decodes image output to depth map
func DecodeDepthMap(img image.Image) *DepthMapResult {
	bounds := img.Bounds()
	height := bounds.Dy()
	width := bounds.Dx()
	
	depthMap := make([][]float32, height)
	for y := 0; y < height; y++ {
		depthMap[y] = make([]float32, width)
		for x := 0; x < width; x++ {
			r, g, b, _ := img.At(x, y).RGBA()
			gray := float32(r+g+b) / (3.0 * 65535.0)
			depthMap[y][x] = gray
		}
	}
	return &DepthMapResult{DepthMap: depthMap}
}