terrassumptions/server/app/point_gen/random_street_view.py

import random

import requests

from .shared import point_has_streetview, GeoPointSource, CachedGeoPointSource, ExhaustedSourceError, GeoPointSourceGroup

RSV_URL = "https://randomstreetview.com/data"
VALID_COUNTRIES = ("ad", "au", "ar", "bd", "be", "bt", "bw", 
                   "br", "bg", "kh", "ca", "cl", "hr", "co", 
                   "cz", "dk", "ae", "ee", "fi", "fr", "de", 
                   "gr", "hu", "hk", "is", "id", "ie", "it", 
                   "il", "jp", "lv", "lt", "my", "mx", "nl", 
                   "nz", "no", "pe", "pl", "pt", "ro", "ru", 
                   "sg", "sk", "si", "za", "kr", "es", "sz", 
                   "se", "ch", "tw", "th", "ua", "gb", "us")


def call_random_street_view(country_lock=None):
    """
    Returns an array of (some number of) tuples, each being (latitude, longitude).
    All points will be valid streetview coordinates. There is no guarantee as to the
    length of this array (it may be empty), but it will never be None.

    This function calls the streetview metadata endpoint - there is no quota consumed.
    """
    if country_lock is None:
        country_lock = random.choice(VALID_COUNTRIES)
    try:
        rsv_js = requests.post(RSV_URL, data={"country": country_lock.lower()}).json()
    except:
        return []

    if not rsv_js["success"]:
        return []
    
    return [
        (country_lock, point["lat"], point["lng"])
        for point in rsv_js["locations"]
        if point_has_streetview(point["lat"], point["lng"])
    ]


class RSVPointSource(GeoPointSource):
    def __init__(self, country_lock=None, max_attempts=100):
        self.country_lock = country_lock
        self.max_attempts = max_attempts

    def get_name(self):
        return f"RSV-{self.country_lock or 'all'}"

    def get_points(self, n):
        attempts = 0
        points = []
        while len(points) < n:
            if attempts > self.max_attempts:
                raise ExhaustedSourceError(points)
            points.extend(call_random_street_view(country_lock=self.country_lock))
            attempts += 1
        return points


WORLD_SOURCE = CachedGeoPointSource(RSVPointSource(), 10)
COUNTRY_SOURCES = {
    "us": CachedGeoPointSource(RSVPointSource("us"), 10),   # cache US specifically since it is commonly used
    **{ k: RSVPointSource(k) for k in VALID_COUNTRIES if k not in ("us",) }
}
SOURCE_GROUP = GeoPointSourceGroup(COUNTRY_SOURCES, WORLD_SOURCE)