terrassumptions/server/app/point_gen/shared.py

from typing import List, Tuple, Union, Dict
import collections
import logging

import requests

# Google API key, with access to Street View Static API
# this can be safely committed due to permission restriction
google_api_key = "AIzaSyAqjCYR6Szph0X0H_iD6O1HenFhL9jySOo"
metadata_url = "https://maps.googleapis.com/maps/api/streetview/metadata"

logger = logging.getLogger(__name__)


def point_has_streetview(lat, lng):
    """
    Returns True if the streetview metadata endpoint says a given point has
    data available, and False otherwise.

    This function calls the streetview metadata endpoint - there is no quota consumed.
    """
    return requests.get(metadata_url, params={
        "key": google_api_key,
        "location": f"{lat},{lng}",
    }).json()["status"] == "OK"


class ExhaustedSourceError(Exception):
    def __init__(self, partial=[]):
        self.partial = partial


class GeoPointSource:
    """
    Abstract base class for a source of geo points
    """
    def get_name(self) -> str:
        """
        Return a human-readable name for this point source, for debugging purposes.
        """
        raise NotImplemented("Must be implemented by subclasses")

    def get_points(self, n: int) -> List[Tuple[float, float]]:
        """
        Return a list of at least n valid geo points, as (latitude, longitude) pairs.
        In the event that the GeoPointSource cannot reasonably supply enough points,
        most likely due to time constraints, it should raise an ExhaustedSourceError.
        """
        raise NotImplemented("Must be implemented by subclasses")


class CachedGeoPointSource(GeoPointSource):
    """
    Wrapper tool for maintaing a cache of points from a GeoPointSource to
    make get_points faster, at the exchange of needing to restock those
    points after the fact. This can be done in another thread, however, to
    hide this cost from the user.
    """

    def __init__(self, source: GeoPointSource, stock_target: int):
        self.source = source
        self.stock = collections.deque()
        self.stock_target = stock_target

    def get_name(self):
        return f"Cached({self.source.get_name()}, {self.stock_target})"

    def restock(self, n: Union[int, None] = None):
        """
        Restock at least n points into this source.
        If n is not provided, it will default to stock_target, as set during the
        construction of this point source.
        """
        n = n if n is not None else self.stock_target - len(self.stock)
        if n > 0:
            logger.info(f"Restocking {type(self).__name__} with {n} points")
            try:
                pts = self.source.get_points(n)
            except ExhaustedSourceError as e:
                pts = e.partial  # take what we can get
            self.stock.extend(pts)
            diff = n - len(pts)
            if diff > 0:
                # if implementations of source.get_points are well behaved, this will
                # never actually need to recurse to finish the job.
                self.restock(n=diff)
            logger.info(f"Finished restocking {type(self).__name__}")

    def get_points(self, n: int) -> List[Tuple[float, float]]:
        """
        Pull n points from the current stock.
        It is recommended to call CachedGeoPointSource.restock after this, to ensure 
        the stock is not depleted. If possible, calling restock in another thread is
        recommended, as it can be a long operation depending on implementation.
        """
        if len(self.stock) >= n:
            pts = []
            for _ in range(n):
                pts.append(self.stock.popleft())
            return pts
        self.restock(n=n)
        # this is safe as long as restock does actually add enough new points.
        # unless this object is being rapidly drained by another thread,
        # this will recur at most once.
        return self.get_points(n=n)


class GeoPointSourceGroup:
    """
    Container of multiple GeoPointSources, each with some key.
    """
    def __init__(self, sources: Dict[str, GeoPointSource], default: GeoPointSource):
        self.sources = sources
        self.default = default

    def restock(self, key: Union[str, None] = None):
        """
        Restock a CachedGeoPointSources managed by this group.
        If the targeted GeoPointSource is uncached, this method does nothing.
        """
        src = self.sources.get(key, self.default)
        if isinstance(src, CachedGeoPointSource):
            src.restock()

    def get_points_from(self, n: int, key: Union[str, None] = None) -> List[Tuple[float, float]]:
        """
        Return a list of at least n valid geo points, as (latitude, longitude) pairs,
        for a given key. If no key is provided, or no matching GeoPointSource is found,
        the default GeoPointSource will be used.
        """
        return self.sources.get(key, self.default).get_points(n)