boc/iom/sdk/python/landvex.py

"""
Landvex Python SDK
Official Python client for Landvex API
"""

import requests
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
from datetime import datetime


@dataclass
class Observation:
    """Observation data model"""
    id: str
    lat: float
    lng: float
    condition: int
    obj_type: str
    zoomer: str
    status: str
    date: str
    rgi: Optional[float] = None


@dataclass
class Mission:
    """Mission data model"""
    id: str
    title: str
    description: str
    category: str
    difficulty: str
    reward_usd: float
    total_observations: int
    completed_observations: int
    deadline: str


class LandvexClient:
    """Landvex API Client"""
    
    def __init__(self, api_key: str, base_url: str = "https://api.landvex.com/v1"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """Make API request"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    
    # === Observations ===
    
    def list_observations(
        self,
        lat: Optional[float] = None,
        lng: Optional[float] = None,
        radius: Optional[float] = None,
        limit: int = 100,
        status: Optional[str] = None,
    ) -> List[Observation]:
        """List observations with optional filtering"""
        params = {"limit": limit}
        if lat is not None:
            params["lat"] = lat
        if lng is not None:
            params["lng"] = lng
        if radius is not None:
            params["radius"] = radius
        if status is not None:
            params["status"] = status
        
        data = self._request("GET", "/observations", params=params)
        
        return [
            Observation(
                id=obs["id"],
                lat=obs["lat"],
                lng=obs["lng"],
                condition=obs["condition"],
                obj_type=obs["type"],
                zoomer=obs["zoomer"],
                status=obs["status"],
                date=obs["date"],
                rgi=obs.get("rgi"),
            )
            for obs in data.get("observations", [])
        ]
    
    def get_observation(self, observation_id: str) -> Observation:
        """Get single observation by ID"""
        data = self._request("GET", f"/observations/{observation_id}")
        obs = data["observation"]
        return Observation(
            id=obs["id"],
            lat=obs["lat"],
            lng=obs["lng"],
            condition=obs["condition"],
            obj_type=obs["type"],
            zoomer=obs["zoomer"],
            status=obs["status"],
            date=obs["date"],
            rgi=obs.get("rgi"),
        )
    
    def create_observation(
        self,
        lat: float,
        lng: float,
        image_url: Optional[str] = None,
        evidence: Optional[Dict] = None,
    ) -> Observation:
        """Create new observation"""
        payload = {
            "lat": lat,
            "lng": lng,
        }
        if image_url:
            payload["image_url"] = image_url
        if evidence:
            payload["evidence"] = evidence
        
        data = self._request("POST", "/observations", json=payload)
        return self.get_observation(data["id"])
    
    # === Missions ===
    
    def list_missions(self, status: Optional[str] = None) -> List[Mission]:
        """List available missions"""
        params = {}
        if status:
            params["status"] = status
        
        data = self._request("GET", "/missions", params=params)
        
        return [
            Mission(
                id=m["id"],
                title=m["title"],
                description=m["description"],
                category=m["category"],
                difficulty=m["difficulty"],
                reward_usd=m["reward_usd"],
                total_observations=m["total_observations"],
                completed_observations=m["completed_observations"],
                deadline=m["deadline"],
            )
            for m in data.get("missions", [])
        ]
    
    def accept_mission(self, mission_id: str) -> Dict:
        """Accept a mission"""
        return self._request("POST", f"/missions/{mission_id}/accept")
    
    # === Indexes ===
    
    def get_rgi(self, lat: float, lng: float) -> Dict:
        """Get Reality Gap Index for location"""
        return self._request("GET", "/indexes/rgi", params={"lat": lat, "lng": lng})
    
    def get_umi(self, lat: float, lng: float) -> Dict:
        """Get Urban Morphology Index for location"""
        return self._request("GET", "/indexes/umi", params={"lat": lat, "lng": lng})
    
    # === Visual Geolocation ===
    
    def analyze_image(
        self,
        image_path: str,
        image_id: Optional[str] = None,
    ) -> Dict:
        """Analyze image with visual geolocation"""
        payload = {"image_path": image_path}
        if image_id:
            payload["image_id"] = image_id
        
        return self._request("POST", "/visual-geolocation/analyze", json=payload)
    
    def batch_analyze(self, images: List[Dict]) -> Dict:
        """Batch analyze multiple images"""
        return self._request("POST", "/visual-geolocation/batch", json={"images": images})
    
    # === Decisions ===
    
    def get_recommendations(
        self,
        stakeholder: str,
        lat: float,
        lng: float,
    ) -> Dict:
        """Get decision recommendations"""
        return self._request("POST", "/decisions/recommend", json={
            "stakeholder": stakeholder,
            "lat": lat,
            "lng": lng,
        })
    
    # === Stats ===
    
    def get_stats(self) -> Dict:
        """Get dashboard statistics"""
        return self._request("GET", "/stats")
    
    def get_earnings(self) -> Dict:
        """Get earnings data"""
        return self._request("GET", "/stats/earnings")
    
    # === Health ===
    
    def health_check(self) -> Dict:
        """Check API health"""
        return self._request("GET", "/health")


# Convenience functions

def create_client(api_key: str, base_url: Optional[str] = None) -> LandvexClient:
    """Create Landvex client"""
    if base_url:
        return LandvexClient(api_key, base_url)
    return LandvexClient(api_key)


# Example usage
if __name__ == "__main__":
    # Initialize client
    client = create_client("your-api-key")
    
    # Check health
    health = client.health_check()
    print(f"API Status: {health['status']}")
    
    # List observations
    observations = client.list_observations(lat=59.3293, lng=18.0686, radius=1000)
    print(f"Found {len(observations)} observations")
    
    # Get RGI
    rgi = client.get_rgi(59.3293, 18.0686)
    print(f"RGI Score: {rgi['rgi']}")
    
    # Get recommendations
    recommendations = client.get_recommendations("municipality", 59.3293, 18.0686)
    print(f"Recommendations: {len(recommendations['recommendations'])}")