| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546 |
- import random
- import logging
- import asyncio
- from .shared import aiohttp_client, point_has_streetview, reverse_geocode
- 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")
- logger = logging.getLogger(__name__)
- async 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, in the country indicated by
- country_lock. No size guarantee is given on the returned array - it could be empty.
- This function calls the streetview metadata endpoint - there is no quota consumed.
- """
-
- try:
- async with aiohttp_client.post(RSV_URL, data={"country": country_lock.lower()}) as response:
- rsv_js = await response.json(content_type=None)
- except:
- logger.exception("Failed RSV call")
- return []
- if not rsv_js["success"]:
- return []
- points = []
- async def add_point_if_valid(point):
- if await point_has_streetview(point["lat"], point["lng"]):
- country_code = await reverse_geocode(point["lat"], point["lng"])
- points.append((country_code or country_lock, point["lat"], point["lng"]))
-
- await asyncio.gather(*[add_point_if_valid(p) for p in rsv_js["locations"]])
- return points
|
