viktor/wrongmove
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

allow caching routing by destination and travel mode; also export all travel methods to csv column

Browse source
This commit is contained in:
Viktor Barzin 2025-05-20 21:58:08 +00:00
parent f2118c9bc4
commit 10ae25e0d3
No known key found for this signature in database
GPG key ID: 4056458DBDBF8863
5 changed files with 190 additions and 84 deletions
Download patch file Download diff file Expand all files Collapse all files

150
crawler/data_access.py
View file

@ -1,4 +1,5 @@
import asyncio
from collections import defaultdict
from dataclasses import dataclass
import json
import pathlib
@ -106,8 +107,8 @@ class Listing:
def calculate_sqm_model(self):
objs = []
for floorplan_path in self.list_floorplans():
estimated_sqm, model_output, predictions = floorplan.calculate_model(
floorplan_path)
estimated_sqm, model_output, predictions = (
floorplan.calculate_model(floorplan_path))
objs.append({
"floorplan_path": str(floorplan_path),
"estimated_sqm": estimated_sqm,
@ -149,10 +150,9 @@ class Listing:
with open(self.path_floorplan_ocr_json(), "w") as f:
json.dump(objs, f)
@property
def sqm_ocr(self, recalculate=False):
async def sqm_ocr(self, recalculate=False):
if not self.path_floorplan_ocr_json().exists() or recalculate:
self.calculate_sqm_ocr()
await self.calculate_sqm_ocr()
with open(self.path_floorplan_ocr_json()) as f:
objs = json.load(f)
@ -168,9 +168,12 @@ class Listing:
def calculate_route(self,
dest_address: str,
travel_mode: routing.TravelMode,
recalculate=False):
if self.path_routing_json().exists() and not recalculate:
return
recalculate=False) -> dict[str, Any]:
routing_cache = self.__get_routing_cache()
cache_key = self.__routing_cache_key(dest_address, travel_mode)
if (route_cache :=
routing_cache.get(cache_key)) is not None and not recalculate:
return {cache_key: route_cache}
result = routing.transit_route(
self.latitude,
@ -178,17 +181,22 @@ class Listing:
dest_address,
travel_mode,
)
if not result:
raise Exception((f"Error calculating route from {self.identifier} "
f"to '{dest_address}' by {travel_mode}"))
result = {**{cache_key: result}, **routing_cache}
with open(self.path_routing_json(), "w") as f:
json.dump(result, f)
return result
@property
def travel_time(self) -> List:
if not self.path_routing_json().exists():
return []
with open(self.path_routing_json()) as f:
d = json.load(f)
return routing.extract_time(d)
def travel_time(
self,
destination_address: str,
travel_mode: routing.TravelMode,
) -> list[dict[str, Any]]:
data = self.calculate_route(destination_address, travel_mode)
return self.__extract_travel_times(data, destination_address,
travel_mode)
@property
def url(self):
@ -215,11 +223,11 @@ class Listing:
def tenure_type(self) -> str:
return self.detailobject["property"]["tenureType"]
@property
def price_per_sqm(self) -> float:
if self.sqm_ocr is None or self.sqm_ocr == 0:
async def price_per_sqm(self) -> float:
sqm_ocr = await self.sqm_ocr()
if sqm_ocr is None or sqm_ocr == 0:
return -1
return self.price / self.sqm_ocr
return self.price / sqm_ocr
@property
def bedrooms(self) -> int:
@ -302,24 +310,44 @@ class Listing:
return self.detailobject['property']["councilTaxInfo"]["content"][0][
"value"]
def dict_nicely(self):
async def dict_nicely(self):
travel_time_fastest = {}
travel_time_second = {}
if self.path_routing_json().exists():
with open(self.path_routing_json(), "r") as f:
travel_times = json.load(f)
for destination_mode in travel_times.keys():
destination_mode_clean = destination_mode.replace(" ",
"_").replace(
",", "_")
destination, travel_mode = self.__from_routing_cache_key(
destination_mode)
travel_time_fastest[destination_mode_clean] = self.travel_time(
destination, travel_mode)[0]['duration']
travel_time_second[destination_mode_clean] = self.travel_time(
destination, travel_mode)[1]['duration']
return {
"identifier":
self.identifier,
"sqm_ocr":
self.sqm_ocr,
await self.sqm_ocr(),
"price":
self.price,
"price_per_sqm":
self.price_per_sqm,
await self.price_per_sqm(),
"url":
self.url,
"bedrooms":
self.bedrooms,
"travel_time_fastest":
None if len(self.travel_time) == 0 else self.travel_time[0],
":".join(
sorted(f'{dest} in {travel_mode//60}min'
for dest, travel_mode in travel_time_fastest.items())),
"travel_time_second":
None if len(self.travel_time) < 2 else self.travel_time[1],
":".join(
sorted(f'{dest} in {travel_mode//60}min'
for dest, travel_mode in travel_time_second.items())),
"lease_left":
self.leaseLeft,
"service_charge":
@ -339,3 +367,75 @@ class Listing:
"council_tax_band":
self.councilTaxBand,
}
def __routing_cache_key(
self,
dest_address: str,
travel_mode: routing.TravelMode,
) -> str:
return f"{dest_address} by {travel_mode}"
def __from_routing_cache_key(
self,
cache_key: str,
) -> tuple[str, routing.TravelMode]:
match = re.match(r"(.+) by (.+)", cache_key)
if not match:
raise ValueError(f"Invalid cache key: {cache_key}")
return match.group(1), routing.TravelMode[match.group(2)]
def __extract_travel_times(
self,
routing_data: dict[str, Any],
destination_address: str,
travel_mode: routing.TravelMode,
limit: int = 2,
) -> list[dict[str, Any]]:
res = []
cache_key = self.__routing_cache_key(destination_address, travel_mode)
for route in routing_data[cache_key]["routes"]:
distance = route["distanceMeters"]
duration = int(route["duration"].strip("s"))
duration_static = int(route["staticDuration"].strip("s"))
steps = route["legs"][0]["steps"]
initial_walk_duration = 0
used_transit = False
duration_per_transit = defaultdict(lambda: 0)
distance_per_transit = defaultdict(lambda: 0)
number_of_transit_stops = 0
for step in steps:
if not used_transit and step["travelMode"] == "WALK":
initial_walk_duration += int(
step["staticDuration"].strip("s"))
else:
used_transit = True
duration_per_transit[step["travelMode"]] += int(
step["staticDuration"].strip("s"))
distance_per_transit[step["travelMode"]] += step.get(
"distanceMeters", 0)
if step["travelMode"] == "TRANSIT":
number_of_transit_stops += 1
res.append({
"duration": duration,
"distance": distance,
"duration_static": duration_static,
"initial_walk_duration": initial_walk_duration,
"duration_per_transit": dict(duration_per_transit),
"distance_per_transit": dict(distance_per_transit),
"number_of_transit_stops": number_of_transit_stops,
})
return res[:limit]
def __get_routing_cache(self) -> dict[str, Any]:
try:
with open(self.path_routing_json(), 'x') as f:
json.dump({}, f)
return {}
except FileExistsError:
pass
with open(self.path_routing_json(), 'r') as f:
return json.load(f)