Bernt
bae705aa97
- Add NFC ePassport roadmap (ICAO 9303, eIDAS) - Add TensorFlow.js edge face detection (BlazeFace) - Add structured audit logger (GDPR-compliant) - Risk scoring support Part of KYC Apple Native UX v1.1.0
550 lines
18 KiB
Python
550 lines
18 KiB
Python
"""
|
|
Evidence Extractor
|
|
Extract multi-layer evidence from images
|
|
"""
|
|
|
|
from typing import Dict, List, Optional, Tuple
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class ImageMetadata:
|
|
"""Image metadata layer"""
|
|
gps_lat: Optional[float] = None
|
|
gps_lng: Optional[float] = None
|
|
altitude: Optional[float] = None
|
|
compass_heading: Optional[float] = None # degrees
|
|
pitch: Optional[float] = None
|
|
roll: Optional[float] = None
|
|
timestamp: Optional[str] = None
|
|
camera_model: Optional[str] = None
|
|
focal_length: Optional[float] = None # mm
|
|
exposure: Optional[float] = None
|
|
aperture: Optional[float] = None
|
|
iso: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class VisualObject:
|
|
"""Detected visual object"""
|
|
label: str
|
|
confidence: float
|
|
bbox: List[float] # [x1, y1, x2, y2]
|
|
attributes: Dict = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class TextDetection:
|
|
"""OCR text detection"""
|
|
text: str
|
|
confidence: float
|
|
bbox: List[float]
|
|
language: Optional[str] = None
|
|
text_type: Optional[str] = None # street_name, business_name, etc.
|
|
|
|
|
|
@dataclass
|
|
class GeometricFeature:
|
|
"""3D geometric feature"""
|
|
feature_type: str # vanishing_point, horizon_line, etc.
|
|
coordinates: List[float]
|
|
confidence: float
|
|
|
|
|
|
@dataclass
|
|
class EnvironmentalSignal:
|
|
"""Environmental signal"""
|
|
signal_type: str # sun_position, shadow_direction, weather, vegetation
|
|
value: Dict
|
|
confidence: float
|
|
|
|
|
|
@dataclass
|
|
class EvidencePackage:
|
|
"""Complete evidence package from image"""
|
|
image_id: str
|
|
timestamp: datetime
|
|
|
|
# Layer 1: Metadata
|
|
metadata: ImageMetadata
|
|
|
|
# Layer 2: Visual objects
|
|
visual_objects: List[VisualObject]
|
|
|
|
# Layer 3: Semantic objects
|
|
semantic_objects: List[VisualObject]
|
|
|
|
# Layer 4: OCR
|
|
text_detections: List[TextDetection]
|
|
|
|
# Layer 5: 3D Geometry
|
|
geometric_features: List[GeometricFeature]
|
|
|
|
# Layer 6: Environment
|
|
environmental_signals: List[EnvironmentalSignal]
|
|
|
|
# Layer 7: Temporal
|
|
temporal_signals: Dict
|
|
|
|
# Raw embeddings
|
|
visual_embedding: Optional[List[float]] = None
|
|
scene_embedding: Optional[List[float]] = None
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""Convert to dictionary"""
|
|
return {
|
|
"image_id": self.image_id,
|
|
"timestamp": self.timestamp.isoformat(),
|
|
"metadata": {
|
|
"gps": {
|
|
"lat": self.metadata.gps_lat,
|
|
"lng": self.metadata.gps_lng,
|
|
"altitude": self.metadata.altitude
|
|
},
|
|
"orientation": {
|
|
"heading": self.metadata.compass_heading,
|
|
"pitch": self.metadata.pitch,
|
|
"roll": self.metadata.roll
|
|
},
|
|
"camera": {
|
|
"model": self.metadata.camera_model,
|
|
"focal_length": self.metadata.focal_length,
|
|
"exposure": self.metadata.exposure,
|
|
"aperture": self.metadata.aperture,
|
|
"iso": self.metadata.iso
|
|
}
|
|
},
|
|
"visual_objects": [
|
|
{
|
|
"label": obj.label,
|
|
"confidence": obj.confidence,
|
|
"bbox": obj.bbox,
|
|
"attributes": obj.attributes
|
|
}
|
|
for obj in self.visual_objects
|
|
],
|
|
"semantic_objects": [
|
|
{
|
|
"label": obj.label,
|
|
"confidence": obj.confidence,
|
|
"bbox": obj.bbox
|
|
}
|
|
for obj in self.semantic_objects
|
|
],
|
|
"text_detections": [
|
|
{
|
|
"text": text.text,
|
|
"confidence": text.confidence,
|
|
"bbox": text.bbox,
|
|
"language": text.language
|
|
}
|
|
for text in self.text_detections
|
|
],
|
|
"geometric_features": [
|
|
{
|
|
"type": feat.feature_type,
|
|
"coordinates": feat.coordinates,
|
|
"confidence": feat.confidence
|
|
}
|
|
for feat in self.geometric_features
|
|
],
|
|
"environmental_signals": [
|
|
{
|
|
"type": sig.signal_type,
|
|
"value": sig.value,
|
|
"confidence": sig.confidence
|
|
}
|
|
for sig in self.environmental_signals
|
|
],
|
|
"temporal_signals": self.temporal_signals
|
|
}
|
|
|
|
|
|
class EvidenceExtractor:
|
|
"""
|
|
Extract multi-layer evidence from images
|
|
|
|
Layers:
|
|
1. Metadata (GPS, IMU, camera)
|
|
2. Visual objects (YOLO)
|
|
3. Semantic objects (CLIP)
|
|
4. OCR (text detection)
|
|
5. 3D Geometry (vanishing points, horizon)
|
|
6. Environment (sun, shadows, weather)
|
|
7. Temporal (historical comparison)
|
|
"""
|
|
|
|
def __init__(self, use_real_ai: bool = True):
|
|
self.use_real_ai = use_real_ai
|
|
|
|
# Initialize AI models
|
|
if use_real_ai:
|
|
from ai_pipeline.real_ai import RealAIClassifier
|
|
self.ai_classifier = RealAIClassifier(use_real_ai=True)
|
|
|
|
# Initialize OCR
|
|
self.ocr_available = self._check_ocr()
|
|
|
|
# Initialize geometric analysis
|
|
self.geometric_available = self._check_geometric()
|
|
|
|
def _check_ocr(self) -> bool:
|
|
"""Check if OCR is available"""
|
|
try:
|
|
import pytesseract
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
def _check_geometric(self) -> bool:
|
|
"""Check if geometric analysis is available"""
|
|
try:
|
|
import cv2
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
def extract_metadata(self, image_path: str) -> ImageMetadata:
|
|
"""Extract metadata from image EXIF"""
|
|
from PIL import Image
|
|
from PIL.ExifTags import TAGS, GPSTAGS
|
|
|
|
metadata = ImageMetadata()
|
|
|
|
try:
|
|
img = Image.open(image_path)
|
|
exif = img._getexif()
|
|
|
|
if exif:
|
|
for tag_id, value in exif.items():
|
|
tag = TAGS.get(tag_id, tag_id)
|
|
|
|
if tag == "GPSInfo":
|
|
gps_data = {}
|
|
for gps_tag_id, gps_value in value.items():
|
|
gps_tag = GPSTAGS.get(gps_tag_id, gps_tag_id)
|
|
gps_data[gps_tag] = gps_value
|
|
|
|
# Extract GPS coordinates
|
|
if "GPSLatitude" in gps_data and "GPSLongitude" in gps_data:
|
|
lat = self._convert_gps_coords(gps_data["GPSLatitude"], gps_data.get("GPSLatitudeRef", "N"))
|
|
lng = self._convert_gps_coords(gps_data["GPSLongitude"], gps_data.get("GPSLongitudeRef", "E"))
|
|
metadata.gps_lat = lat
|
|
metadata.gps_lng = lng
|
|
|
|
if "GPSAltitude" in gps_data:
|
|
metadata.altitude = float(gps_data["GPSAltitude"])
|
|
|
|
elif tag == "DateTimeOriginal":
|
|
metadata.timestamp = value
|
|
elif tag == "Make":
|
|
metadata.camera_model = value
|
|
elif tag == "Model":
|
|
metadata.camera_model = f"{metadata.camera_model} {value}".strip()
|
|
elif tag == "FocalLength":
|
|
metadata.focal_length = float(value)
|
|
elif tag == "ExposureTime":
|
|
metadata.exposure = float(value)
|
|
elif tag == "FNumber":
|
|
metadata.aperture = float(value)
|
|
elif tag == "ISOSpeedRatings":
|
|
metadata.iso = int(value)
|
|
|
|
except Exception as e:
|
|
print(f"Error extracting metadata: {e}")
|
|
|
|
return metadata
|
|
|
|
def _convert_gps_coords(self, coords, ref):
|
|
"""Convert GPS coordinates from EXIF"""
|
|
degrees = float(coords[0])
|
|
minutes = float(coords[1])
|
|
seconds = float(coords[2])
|
|
|
|
decimal = degrees + minutes / 60 + seconds / 3600
|
|
|
|
if ref in ["S", "W"]:
|
|
decimal = -decimal
|
|
|
|
return decimal
|
|
|
|
def extract_visual_objects(self, image_path: str) -> List[VisualObject]:
|
|
"""Extract visual objects using YOLO"""
|
|
if not self.use_real_ai:
|
|
return []
|
|
|
|
result = self.ai_classifier.analyze_image(image_path)
|
|
|
|
objects = []
|
|
for obj in result.detected_objects:
|
|
visual_obj = VisualObject(
|
|
label=obj.get("label", "unknown"),
|
|
confidence=obj.get("confidence", 0.0),
|
|
bbox=obj.get("bbox", [0, 0, 0, 0]),
|
|
attributes=obj.get("attributes", {})
|
|
)
|
|
objects.append(visual_obj)
|
|
|
|
return objects
|
|
|
|
def extract_semantic_objects(self, image_path: str) -> List[VisualObject]:
|
|
"""Extract semantic objects using CLIP"""
|
|
if not self.use_real_ai:
|
|
return []
|
|
|
|
# Use CLIP for scene classification
|
|
result = self.ai_classifier.analyze_image(image_path)
|
|
|
|
# Create semantic objects from scene type
|
|
semantic_objects = []
|
|
if result.scene_type:
|
|
semantic_obj = VisualObject(
|
|
label=result.scene_type,
|
|
confidence=result.confidence,
|
|
bbox=[0, 0, 0, 0], # Full image
|
|
attributes={"type": "scene"}
|
|
)
|
|
semantic_objects.append(semantic_obj)
|
|
|
|
return semantic_objects
|
|
|
|
def extract_text(self, image_path: str) -> List[TextDetection]:
|
|
"""Extract text using OCR"""
|
|
if not self.ocr_available:
|
|
return []
|
|
|
|
try:
|
|
import pytesseract
|
|
from PIL import Image
|
|
|
|
img = Image.open(image_path)
|
|
|
|
# Get OCR data
|
|
data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)
|
|
|
|
text_detections = []
|
|
n_boxes = len(data["text"])
|
|
|
|
for i in range(n_boxes):
|
|
if int(data["conf"][i]) > 30: # Confidence threshold
|
|
text = data["text"][i].strip()
|
|
if text:
|
|
x, y, w, h = data["left"][i], data["top"][i], data["width"][i], data["height"][i]
|
|
|
|
text_det = TextDetection(
|
|
text=text,
|
|
confidence=float(data["conf"][i]) / 100,
|
|
bbox=[x, y, x + w, y + h],
|
|
language=None # Could detect language
|
|
)
|
|
text_detections.append(text_det)
|
|
|
|
return text_detections
|
|
|
|
except Exception as e:
|
|
print(f"OCR error: {e}")
|
|
return []
|
|
|
|
def extract_geometric_features(self, image_path: str) -> List[GeometricFeature]:
|
|
"""Extract 3D geometric features"""
|
|
if not self.geometric_available:
|
|
return []
|
|
|
|
try:
|
|
import cv2
|
|
import numpy as np
|
|
|
|
img = cv2.imread(image_path)
|
|
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
|
|
|
features = []
|
|
|
|
# Detect lines (potential vanishing points)
|
|
edges = cv2.Canny(gray, 50, 150)
|
|
lines = cv2.HoughLinesP(edges, 1, np.pi / 180, threshold=100, minLineLength=100, maxLineGap=10)
|
|
|
|
if lines is not None:
|
|
# Estimate vanishing points
|
|
# Simplified: just detect horizon line
|
|
horizon_y = img.shape[0] // 2
|
|
|
|
horizon_feature = GeometricFeature(
|
|
feature_type="horizon_line",
|
|
coordinates=[0, horizon_y, img.shape[1], horizon_y],
|
|
confidence=0.6
|
|
)
|
|
features.append(horizon_feature)
|
|
|
|
# Detect perspective
|
|
# Simplified: estimate camera height from object sizes
|
|
camera_height = 1.6 # Default human eye level
|
|
|
|
perspective_feature = GeometricFeature(
|
|
feature_type="camera_height_estimate",
|
|
coordinates=[camera_height],
|
|
confidence=0.5
|
|
)
|
|
features.append(perspective_feature)
|
|
|
|
return features
|
|
|
|
except Exception as e:
|
|
print(f"Geometric analysis error: {e}")
|
|
return []
|
|
|
|
def extract_environmental_signals(self, image_path: str) -> List[EnvironmentalSignal]:
|
|
"""Extract environmental signals"""
|
|
signals = []
|
|
|
|
try:
|
|
from PIL import Image
|
|
import numpy as np
|
|
|
|
img = Image.open(image_path)
|
|
img_array = np.array(img)
|
|
|
|
# Estimate sun position from brightness
|
|
brightness = np.mean(img_array)
|
|
|
|
sun_signal = EnvironmentalSignal(
|
|
signal_type="brightness",
|
|
value={"mean_brightness": float(brightness), "is_daytime": brightness > 100},
|
|
confidence=0.7
|
|
)
|
|
signals.append(sun_signal)
|
|
|
|
# Estimate weather from color distribution
|
|
# Simplified: check for blue sky
|
|
sky_color = np.mean(img_array[:img_array.shape[0]//3, :, 2]) # Blue channel in top third
|
|
|
|
weather_signal = EnvironmentalSignal(
|
|
signal_type="weather_estimate",
|
|
value={"sky_blue_intensity": float(sky_color), "likely_clear": sky_color > 100},
|
|
confidence=0.5
|
|
)
|
|
signals.append(weather_signal)
|
|
|
|
return signals
|
|
|
|
except Exception as e:
|
|
print(f"Environmental analysis error: {e}")
|
|
return []
|
|
|
|
def extract_temporal_signals(self, image_path: str, metadata: ImageMetadata) -> Dict:
|
|
"""Extract temporal signals"""
|
|
signals = {
|
|
"timestamp": metadata.timestamp,
|
|
"time_of_day": None,
|
|
"season": None,
|
|
"day_of_week": None
|
|
}
|
|
|
|
if metadata.timestamp:
|
|
try:
|
|
dt = datetime.strptime(metadata.timestamp, "%Y:%m:%d %H:%M:%S")
|
|
|
|
signals["time_of_day"] = "day" if 6 <= dt.hour < 18 else "night"
|
|
signals["day_of_week"] = dt.strftime("%A")
|
|
|
|
# Estimate season (Northern Hemisphere)
|
|
month = dt.month
|
|
if month in [12, 1, 2]:
|
|
signals["season"] = "winter"
|
|
elif month in [3, 4, 5]:
|
|
signals["season"] = "spring"
|
|
elif month in [6, 7, 8]:
|
|
signals["season"] = "summer"
|
|
else:
|
|
signals["season"] = "autumn"
|
|
|
|
except:
|
|
pass
|
|
|
|
return signals
|
|
|
|
def extract_all_evidence(self, image_path: str, image_id: str = None) -> EvidencePackage:
|
|
"""Extract all evidence layers from image"""
|
|
print(f"Extracting evidence from {image_path}...")
|
|
|
|
# Generate image ID if not provided
|
|
if not image_id:
|
|
import hashlib
|
|
with open(image_path, "rb") as f:
|
|
image_id = hashlib.md5(f.read()).hexdigest()[:12]
|
|
|
|
# Extract all layers
|
|
print(" 1. Metadata...")
|
|
metadata = self.extract_metadata(image_path)
|
|
|
|
print(" 2. Visual objects...")
|
|
visual_objects = self.extract_visual_objects(image_path)
|
|
|
|
print(" 3. Semantic objects...")
|
|
semantic_objects = self.extract_semantic_objects(image_path)
|
|
|
|
print(" 4. OCR text...")
|
|
text_detections = self.extract_text(image_path)
|
|
|
|
print(" 5. Geometric features...")
|
|
geometric_features = self.extract_geometric_features(image_path)
|
|
|
|
print(" 6. Environmental signals...")
|
|
environmental_signals = self.extract_environmental_signals(image_path)
|
|
|
|
print(" 7. Temporal signals...")
|
|
temporal_signals = self.extract_temporal_signals(image_path, metadata)
|
|
|
|
# Create evidence package
|
|
package = EvidencePackage(
|
|
image_id=image_id,
|
|
timestamp=datetime.now(),
|
|
metadata=metadata,
|
|
visual_objects=visual_objects,
|
|
semantic_objects=semantic_objects,
|
|
text_detections=text_detections,
|
|
geometric_features=geometric_features,
|
|
environmental_signals=environmental_signals,
|
|
temporal_signals=temporal_signals
|
|
)
|
|
|
|
print(f"\nEvidence extraction complete:")
|
|
print(f" Visual objects: {len(visual_objects)}")
|
|
print(f" Semantic objects: {len(semantic_objects)}")
|
|
print(f" Text detections: {len(text_detections)}")
|
|
print(f" Geometric features: {len(geometric_features)}")
|
|
print(f" Environmental signals: {len(environmental_signals)}")
|
|
|
|
return package
|
|
|
|
|
|
# Example usage
|
|
def example_extraction():
|
|
"""Example: Extract evidence from image"""
|
|
extractor = EvidenceExtractor(use_real_ai=True)
|
|
|
|
# In production, use real image
|
|
# package = extractor.extract_all_evidence("image.jpg")
|
|
|
|
# For demo, create empty package
|
|
package = EvidencePackage(
|
|
image_id="demo_001",
|
|
timestamp=datetime.now(),
|
|
metadata=ImageMetadata(),
|
|
visual_objects=[],
|
|
semantic_objects=[],
|
|
text_detections=[],
|
|
geometric_features=[],
|
|
environmental_signals=[],
|
|
temporal_signals={}
|
|
)
|
|
|
|
print(json.dumps(package.to_dict(), indent=2))
|
|
|
|
return package
|
|
|
|
|
|
if __name__ == '__main__':
|
|
example_extraction()
|