85 lines
3 KiB
Python
85 lines
3 KiB
Python
from data_access import Listing
|
|
from tqdm import tqdm
|
|
from rec import routing
|
|
|
|
|
|
async def calculate_route(
|
|
listing_paths: list[str],
|
|
destination_address: str,
|
|
travel_mode: routing.TravelMode,
|
|
) -> None:
|
|
|
|
listings = Listing.get_all_listings(listing_paths)
|
|
|
|
# reduce listings to everything within 7 miles
|
|
filtered_listings = []
|
|
for listing in listings:
|
|
print(f"Processing {listing.identifier}")
|
|
if listing.isRemoved:
|
|
print(f"Removed-Skip: Skipping {listing.identifier} " "is already removed.")
|
|
continue
|
|
sqm_ocr = await listing.sqm_ocr()
|
|
if sqm_ocr is None or sqm_ocr < 30 or sqm_ocr > 200:
|
|
print(
|
|
(
|
|
f"Floorplan-Skip: Skipping {listing.identifier} as "
|
|
f"sqm_ocr is {sqm_ocr}"
|
|
)
|
|
)
|
|
continue
|
|
filtered_listings.append(listing)
|
|
|
|
print(f"Filtered listings from {len(listings)} to {len(filtered_listings)}")
|
|
|
|
for listing in tqdm(filtered_listings):
|
|
listing.calculate_route(
|
|
destination_address,
|
|
travel_mode,
|
|
recalculate=False,
|
|
)
|
|
traveltime = listing.travel_time(destination_address, travel_mode)[0]
|
|
duration_minutes = traveltime["duration"] / 60.0
|
|
tqdm.write(f"{listing.identifier} {duration_minutes}")
|
|
|
|
|
|
# async def geocode_address(
|
|
# address: str,
|
|
# geocoding_cache: pathlib.Path,
|
|
# ) -> tuple[int, int]:
|
|
# cache = get_geocoding_cache(geocoding_cache)
|
|
# cached_results = cache.get(address)
|
|
# if cached_results is None:
|
|
# # resolve
|
|
# async with aiohttp.ClientSession() as session:
|
|
# async with session.get(
|
|
# ("https://maps.googleapis.com/maps/api/geocode/json"
|
|
# f"?address={address}"
|
|
# f"&key={API_KEY_ENVIRONMENT_VARIABLE}")) as response:
|
|
# if response.status != 200:
|
|
# raise Exception(
|
|
# f"Error {response.status} from geocoding API")
|
|
# cached_results = await response.json()
|
|
# with open(geocoding_cache, 'w') as f:
|
|
# json.dump({
|
|
# **{
|
|
# address: cached_results,
|
|
# },
|
|
# **cache
|
|
# }, f)
|
|
# # API format
|
|
# lat = cached_results["results"][0]["geometry"]["location"]["lat"]
|
|
# lng = cached_results["results"][0]["geometry"]["location"]["lng"]
|
|
# cache[address] = (lat, lng)
|
|
# with open(geocoding_cache, 'w') as f:
|
|
# json.dump(cache, f)
|
|
# return lat, lng
|
|
|
|
# def get_geocoding_cache(geocoding_cache: pathlib.Path) -> dict[str, Any]:
|
|
# try:
|
|
# with open(geocoding_cache, 'x') as f:
|
|
# json.dump({}, f)
|
|
# return {}
|
|
# except FileExistsError:
|
|
# pass # File already exists
|
|
# with open(geocoding_cache, 'r') as f:
|
|
# return json.load(f)
|