Chapter 8: Geospatial Intelligence¶

from astral import LocationInfo
from astral.sun import sun
from datetime import datetime, timezone
import math

def solar_position_at(latitude, longitude, datetime_utc):
    """Calculate solar azimuth and elevation at a given location/time"""
    from astral import LocationInfo
    from astral.sun import sun, azimuth as solar_azimuth, elevation as solar_elevation

    loc = LocationInfo(
        name="Investigation Location",
        region="",
        timezone="UTC",
        latitude=latitude,
        longitude=longitude
    )

    s = sun(loc.observer, date=datetime_utc.date())
    az = solar_azimuth(loc.observer, datetime_utc)
    el = solar_elevation(loc.observer, datetime_utc)

    return {
        'azimuth': az,
        'elevation': el,
        'sunrise': str(s['sunrise']),
        'sunset': str(s['sunset']),
        'noon': str(s['noon']),
    }

def estimate_photo_time(latitude, longitude, shadow_direction_degrees, target_date):
    """
    Estimate when a photo was taken based on shadow direction
    shadow_direction_degrees: direction the shadow points (0=North, 90=East, etc.)
    """
    from astral.sun import azimuth as solar_azimuth

    loc_info = LocationInfo(
        latitude=latitude,
        longitude=longitude,
        timezone="UTC"
    )

    # Shadow direction is opposite solar azimuth
    solar_azimuth_from_shadow = (shadow_direction_degrees + 180) % 360

    # Check each hour of the day
    possible_times = []
    for hour in range(6, 20):  # Daylight hours
        for minute in [0, 15, 30, 45]:
            dt = datetime(target_date.year, target_date.month, target_date.day,
                         hour, minute, tzinfo=timezone.utc)
            az = solar_azimuth(loc_info.observer, dt)

            if abs(az - solar_azimuth_from_shadow) < 5:  # Within 5 degrees
                possible_times.append({
                    'time_utc': str(dt),
                    'solar_azimuth': az,
                    'match_quality': abs(az - solar_azimuth_from_shadow)
                })

    return sorted(possible_times, key=lambda x: x['match_quality'])

# Working with Sentinel-2 imagery via SentinelHub
from sentinelhub import (
    SHConfig, BBox, CRS, DataCollection,
    SentinelHubRequest, MimeType, MosaickingOrder
)
from datetime import date
import numpy as np
import matplotlib.pyplot as plt

def get_sentinel_imagery(bbox_coords, date_from, date_to, resolution=10):
    """
    Retrieve Sentinel-2 imagery for an area of interest
    bbox_coords: (min_lon, min_lat, max_lon, max_lat)
    """
    config = SHConfig()
    config.sh_client_id = 'YOUR_CLIENT_ID'
    config.sh_client_secret = 'YOUR_CLIENT_SECRET'

    bbox = BBox(bbox=bbox_coords, crs=CRS.WGS84)

    request = SentinelHubRequest(
        evalscript="""
        //VERSION=3
        function setup() {
          return { input: ["B04", "B03", "B02"], output: { bands: 3 } };
        }
        function evaluatePixel(sample) {
          return [2.5*sample.B04, 2.5*sample.B03, 2.5*sample.B02];
        }
        """,
        input_data=[
            SentinelHubRequest.input_data(
                data_collection=DataCollection.SENTINEL2_L2A,
                time_interval=(date_from, date_to),
                mosaicking_order=MosaickingOrder.LEAST_CC,
            )
        ],
        responses=[SentinelHubRequest.output_response('default', MimeType.PNG)],
        bbox=bbox,
        size=(1024, 1024),
        config=config,
    )

    images = request.get_data()
    return images[0] if images else None

def detect_changes(image_before, image_after, threshold=30):
    """Simple change detection between two imagery dates"""
    diff = np.abs(image_before.astype(int) - image_after.astype(int))
    change_mask = diff.mean(axis=2) > threshold

    change_percentage = change_mask.mean() * 100
    return change_mask, change_percentage

import requests
from datetime import datetime, timedelta

class MarineTrafficAPI:
    """MarineTraffic API client for vessel intelligence"""

    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://services.marinetraffic.com/api"

    def get_vessel_track(self, mmsi, from_date, to_date):
        """Get historical track for a vessel"""
        endpoint = f"{self.base_url}/exportvesseltrack/v:5/{self.api_key}"
        params = {
            'v': 5,
            'mmsi': mmsi,
            'fromdate': from_date,
            'todate': to_date,
            'protocol': 'json'
        }

        response = requests.get(endpoint, params=params)
        return response.json() if response.status_code == 200 else {}

    def get_vessel_photos(self, mmsi):
        """Get photos associated with a vessel"""
        endpoint = f"{self.base_url}/getphotos/v:1/{self.api_key}"
        response = requests.get(endpoint, params={'mmsi': mmsi})
        return response.json() if response.status_code == 200 else {}

    def search_vessels_in_area(self, lat, lon, radius_km, vessel_type=None):
        """Find vessels in a geographic area"""
        endpoint = f"{self.base_url}/getvessel/v:8/{self.api_key}"
        params = {
            'lat': lat,
            'lon': lon,
            'radius': radius_km * 1000,  # API expects meters
        }
        if vessel_type:
            params['type'] = vessel_type

        response = requests.get(endpoint, params=params)
        return response.json() if response.status_code == 200 else {}

def investigate_vessel(mmsi, investigation_window_days=30):
    """Build vessel intelligence profile"""
    # Get vessel information
    response = requests.get(
        f"https://www.marinetraffic.com/getData/get_data_json_4/z:1/X:0/Y:0/station:0",
        params={'mmsi': mmsi}
    )

    profile = {
        'mmsi': mmsi,
        'investigation_note': 'Use MarineTraffic API with valid key for production use'
    }

    return profile

import requests
import json
from datetime import datetime

class OpenSkyAPI:
    """OpenSky Network API client"""

    BASE_URL = "https://opensky-network.org/api"

    def __init__(self, username=None, password=None):
        self.auth = (username, password) if username else None

    def get_states(self, time=None, icao24=None, bbox=None):
        """Get current or historical aircraft states"""
        params = {}
        if time:
            params['time'] = time
        if icao24:
            params['icao24'] = icao24 if isinstance(icao24, str) else ','.join(icao24)
        if bbox:  # (min_lat, max_lat, min_lon, max_lon)
            params.update({
                'lamin': bbox[0], 'lamax': bbox[1],
                'lomin': bbox[2], 'lomax': bbox[3]
            })

        response = requests.get(
            f"{self.BASE_URL}/states/all",
            params=params,
            auth=self.auth
        )

        if response.status_code == 200:
            data = response.json()
            states = []
            for s in (data.get('states') or []):
                states.append({
                    'icao24': s[0],
                    'callsign': s[1].strip() if s[1] else None,
                    'origin_country': s[2],
                    'last_contact': datetime.fromtimestamp(s[3]).isoformat() if s[3] else None,
                    'longitude': s[5],
                    'latitude': s[6],
                    'altitude_m': s[7],
                    'on_ground': s[8],
                    'velocity_ms': s[9],
                    'heading': s[10],
                    'squawk': s[14],
                })
            return states
        return []

    def get_flights_by_aircraft(self, icao24, begin_timestamp, end_timestamp):
        """Get flight history for a specific aircraft"""
        response = requests.get(
            f"{self.BASE_URL}/flights/aircraft",
            params={
                'icao24': icao24,
                'begin': begin_timestamp,
                'end': end_timestamp
            },
            auth=self.auth
        )
        return response.json() if response.status_code == 200 else []

def lookup_faa_registration(n_number):
    """Look up FAA aircraft registration by N-number"""
    # FAA Aircraft Registry (ReleasedAircraftInformation on FAA.gov)
    # API endpoint for individual lookups
    clean_n = n_number.upper().replace('N', '', 1) if n_number.startswith('N') else n_number

    response = requests.get(
        f"https://api.faa.gov/aircraft/{n_number}",
        headers={'apikey': 'your_faa_api_key'}  # FAA API registration required
    )

    # Alternative: scrape the publicly accessible search
    alt_response = requests.get(
        "https://registry.faa.gov/aircraftinquiry/Search/NNumberInquiry",
        params={'nNumberTxt': n_number}
    )

    return alt_response.text  # Parse HTML for registration data

import tweepy
import json
from geopy.geocoders import Nominatim
from geopy.distance import distance

def collect_geotagged_tweets(query, bbox, max_results=100):
    """
    Collect geotagged tweets within a bounding box
    bbox: [southwest_lon, southwest_lat, northeast_lon, northeast_lat]
    """
    # Note: Twitter API v2 requires appropriate access tier for geo search
    client = tweepy.Client(bearer_token='YOUR_BEARER_TOKEN')

    # Build query with geo filter
    geo_query = f"{query} bounding_box:[{' '.join(map(str, bbox))}]"

    tweets = client.search_recent_tweets(
        query=geo_query,
        tweet_fields=['created_at', 'geo', 'author_id', 'entities'],
        expansions=['geo.place_id', 'author_id'],
        place_fields=['contained_within', 'country', 'full_name', 'geo', 'name', 'place_type'],
        max_results=min(max_results, 100)
    )

    geolocated = []
    if tweets.data:
        places = {p.id: p for p in (tweets.includes.get('places') or [])}
        for tweet in tweets.data:
            if tweet.geo:
                place = places.get(tweet.geo.get('place_id'))
                geolocated.append({
                    'tweet_id': tweet.id,
                    'text': tweet.text,
                    'created_at': str(tweet.created_at),
                    'coordinates': tweet.geo.get('coordinates'),
                    'place_name': place.full_name if place else None,
                    'place_type': place.place_type if place else None,
                })

    return geolocated

def analyze_location_patterns(geolocated_posts):
    """Analyze spatial patterns in geolocated content"""
    from collections import Counter
    import statistics

    coordinates = [
        (p['coordinates']['coordinates'][1], p['coordinates']['coordinates'][0])
        for p in geolocated_posts
        if p.get('coordinates') and p['coordinates'].get('coordinates')
    ]

    if not coordinates:
        return {}

    lats = [c[0] for c in coordinates]
    lons = [c[1] for c in coordinates]

    return {
        'total_posts': len(coordinates),
        'centroid': (statistics.mean(lats), statistics.mean(lons)),
        'lat_range': (min(lats), max(lats)),
        'lon_range': (min(lons), max(lons)),
        'geographic_spread_km': distance(
            (min(lats), min(lons)),
            (max(lats), max(lons))
        ).km
    }

from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
import json

@dataclass
class GeospatialEvent:
    """A confirmed or probable event at a geographic location"""
    location: Tuple[float, float]  # (lat, lon)
    location_name: str
    timestamp: str
    event_type: str  # 'sighting', 'activity', 'structure', 'vessel', 'aircraft'
    source: str
    confidence: str  # 'confirmed', 'probable', 'possible'
    evidence: List[str] = field(default_factory=list)
    notes: str = ""

class GeospatialInvestigation:
    """Manages geospatial data in an OSINT investigation"""

    def __init__(self, subject_name: str):
        self.subject = subject_name
        self.events: List[GeospatialEvent] = []
        self.locations_of_interest: List[Dict] = []

    def add_event(self, event: GeospatialEvent):
        self.events.append(event)

    def get_location_timeline(self):
        """Return events sorted by timestamp"""
        return sorted(self.events, key=lambda e: e.timestamp)

    def find_colocation_patterns(self, time_window_hours=24):
        """Identify events that occurred in the same area within a time window"""
        from geopy.distance import distance
        from datetime import datetime, timedelta

        patterns = []
        for i, event1 in enumerate(self.events):
            for event2 in self.events[i+1:]:
                try:
                    dist = distance(event1.location, event2.location).km
                    dt1 = datetime.fromisoformat(event1.timestamp)
                    dt2 = datetime.fromisoformat(event2.timestamp)
                    time_diff = abs((dt2 - dt1).total_seconds() / 3600)

                    if dist < 1.0 and time_diff < time_window_hours:
                        patterns.append({
                            'event1': event1,
                            'event2': event2,
                            'distance_km': round(dist, 3),
                            'time_diff_hours': round(time_diff, 2)
                        })
                except Exception:
                    continue

        return patterns

    def export_geojson(self):
        """Export events as GeoJSON for visualization"""
        features = []
        for event in self.events:
            features.append({
                "type": "Feature",
                "geometry": {
                    "type": "Point",
                    "coordinates": [event.location[1], event.location[0]]
                },
                "properties": {
                    "name": event.location_name,
                    "timestamp": event.timestamp,
                    "type": event.event_type,
                    "source": event.source,
                    "confidence": event.confidence,
                    "notes": event.notes
                }
            })

        return json.dumps({
            "type": "FeatureCollection",
            "name": f"OSINT Investigation: {self.subject}",
            "features": features
        }, indent=2)

from transformers import pipeline
import anthropic

def ai_assisted_geolocation(image_path, description=None):
    """
    Use multimodal AI to assist with image geolocation
    """
    import base64

    with open(image_path, 'rb') as f:
        image_data = base64.b64encode(f.read()).decode('utf-8')

    client = anthropic.Anthropic()

    # Use Claude's vision capability for geolocation analysis
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/jpeg",
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": """Analyze this image for geographic indicators to help with location identification.

Please identify:
1. Any visible text (signs, labels, logos)
2. Language of any text visible
3. Vehicle types and any visible license plate patterns
4. Architectural style and building materials
5. Vegetation types
6. Infrastructure features (road markings, electrical infrastructure, etc.)
7. Any distinctive landmarks
8. Possible geographic region based on all visible indicators

Do not speculate beyond what is visible. Note confidence level for each indicator."""
                    }
                ],
            }
        ],
    )

    return {
        'ai_analysis': message.content[0].text,
        'image_path': image_path,
        'description': description
    }