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):
    """
    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.
    """
    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 RSVCountryPointSource(GeoPointSource):
    def __init__(self, country_lock, max_attempts=100):
        self.country_lock = country_lock
        self.max_attempts = max_attempts

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

    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


COUNTRY_SOURCES = { k: CachedGeoPointSource(RSVCountryPointSource(k), 10) for k in VALID_COUNTRIES }


class RSVWorldPointSource(GeoPointSource):
    def __init__(self, max_attempts=10):
        self.max_attempts = max_attempts

    def get_name(self):
        return "RSV-global"

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


WORLD_SOURCE = RSVWorldPointSource()
SOURCE_GROUP = GeoPointSourceGroup(COUNTRY_SOURCES, WORLD_SOURCE)