fix types and format
This commit is contained in:
Viktor Barzin
parent
91d3237516
commit
b873eaf203
8 changed files with 117 additions and 172 deletions
|
|
@ -2,7 +2,7 @@ import asyncio
|
|||
from dataclasses import dataclass
|
||||
import json
|
||||
import pathlib
|
||||
from typing import List, Dict
|
||||
from typing import Any, List, Dict
|
||||
from rec import floorplan, routing
|
||||
import re
|
||||
import datetime
|
||||
|
|
@ -11,7 +11,7 @@ import datetime
|
|||
@dataclass()
|
||||
class Listing:
|
||||
identifier: int
|
||||
_cached: Dict = None
|
||||
_cached: Dict | None = None
|
||||
data_dir: pathlib.Path = pathlib.Path("data/rs/")
|
||||
ALL_COLUMNS = [
|
||||
"identifier",
|
||||
|
|
@ -46,10 +46,8 @@ class Listing:
|
|||
while str(d['identifier']) in str(data_dir.resolve().absolute()):
|
||||
data_dir = data_dir.parent
|
||||
listing = Listing(d["identifier"], data_dir=data_dir)
|
||||
if (
|
||||
listing.last_seen is not None
|
||||
and listing.last_seen < seen_in_the_last_n_days
|
||||
):
|
||||
if (listing.last_seen is not None
|
||||
and listing.last_seen < seen_in_the_last_n_days):
|
||||
identifiers.append(listing)
|
||||
|
||||
return identifiers
|
||||
|
|
@ -107,18 +105,15 @@ class Listing:
|
|||
objs = []
|
||||
for floorplan_path in self.list_floorplans():
|
||||
estimated_sqm, model_output, predictions = floorplan.calculate_model(
|
||||
floorplan_path
|
||||
)
|
||||
objs.append(
|
||||
{
|
||||
"floorplan_path": str(floorplan_path),
|
||||
"estimated_sqm": estimated_sqm,
|
||||
"model_output": model_output,
|
||||
"no_predictions": len(
|
||||
predictions
|
||||
), # cant serialize the predictions itself since its a tensor
|
||||
}
|
||||
)
|
||||
floorplan_path)
|
||||
objs.append({
|
||||
"floorplan_path": str(floorplan_path),
|
||||
"estimated_sqm": estimated_sqm,
|
||||
"model_output": model_output,
|
||||
"no_predictions": len(
|
||||
predictions
|
||||
), # cant serialize the predictions itself since its a tensor
|
||||
})
|
||||
|
||||
with open(self.path_floorplan_model_json(), "w") as f:
|
||||
json.dump(objs, f)
|
||||
|
|
@ -131,9 +126,8 @@ class Listing:
|
|||
with open(self.path_floorplan_json()) as f:
|
||||
objs = json.load(f)
|
||||
|
||||
max_sqm = max(
|
||||
[o["estimated_sqm"] for o in objs if o is None]
|
||||
) # filter out Nones
|
||||
max_sqm = max([o["estimated_sqm"] for o in objs
|
||||
if o is None]) # filter out Nones
|
||||
return max_sqm
|
||||
|
||||
async def calculate_sqm_ocr(self, recalculate=True):
|
||||
|
|
@ -143,15 +137,12 @@ class Listing:
|
|||
objs = []
|
||||
for floorplan_path in self.list_floorplans():
|
||||
estimated_sqm, model_output = await asyncio.to_thread(
|
||||
floorplan.calculate_ocr, floorplan_path
|
||||
)
|
||||
objs.append(
|
||||
{
|
||||
"floorplan_path": str(floorplan_path),
|
||||
"estimated_sqm": estimated_sqm,
|
||||
"text": model_output,
|
||||
}
|
||||
)
|
||||
floorplan.calculate_ocr, floorplan_path)
|
||||
objs.append({
|
||||
"floorplan_path": str(floorplan_path),
|
||||
"estimated_sqm": estimated_sqm,
|
||||
"text": model_output,
|
||||
})
|
||||
|
||||
with open(self.path_floorplan_ocr_json(), "w") as f:
|
||||
json.dump(objs, f)
|
||||
|
|
@ -164,19 +155,23 @@ class Listing:
|
|||
with open(self.path_floorplan_ocr_json()) as f:
|
||||
objs = json.load(f)
|
||||
|
||||
sqms = [o["estimated_sqm"] for o in objs if o["estimated_sqm"] is not None]
|
||||
sqms = [
|
||||
o["estimated_sqm"] for o in objs if o["estimated_sqm"] is not None
|
||||
]
|
||||
if len(sqms) == 0:
|
||||
return None
|
||||
max_sqm = max(sqms)
|
||||
return max_sqm
|
||||
|
||||
def calculate_route(self, dest_lat: float, dest_lon: float, recalculate=False):
|
||||
def calculate_route(self,
|
||||
dest_lat: float,
|
||||
dest_lon: float,
|
||||
recalculate=False):
|
||||
if self.path_routing_json().exists() and not recalculate:
|
||||
return
|
||||
|
||||
result = routing.transit_route(
|
||||
self.latitude, self.longitude, dest_lat, dest_lon
|
||||
)
|
||||
result = routing.transit_route(self.latitude, self.longitude, dest_lat,
|
||||
dest_lon)
|
||||
with open(self.path_routing_json(), "w") as f:
|
||||
json.dump(result, f)
|
||||
|
||||
|
|
@ -200,11 +195,11 @@ class Listing:
|
|||
return json.load(f)
|
||||
|
||||
@property
|
||||
def detailobject(self):
|
||||
def detailobject(self) -> dict[str, Any]:
|
||||
if self._cached is None:
|
||||
with open(self.path_detail_json()) as f:
|
||||
self._cached = json.load(f)
|
||||
return self._cached
|
||||
return self._cached # type: ignore
|
||||
|
||||
@property
|
||||
def price(self) -> float:
|
||||
|
|
@ -217,7 +212,7 @@ class Listing:
|
|||
@property
|
||||
def price_per_sqm(self) -> float:
|
||||
if self.sqm_ocr is None or self.sqm_ocr == 0:
|
||||
return None
|
||||
return -1
|
||||
return self.price / self.sqm_ocr
|
||||
|
||||
@property
|
||||
|
|
@ -233,8 +228,9 @@ class Listing:
|
|||
return self.detailobject["property"]["longitude"]
|
||||
|
||||
@property
|
||||
def leaseLeft(self) -> int:
|
||||
ds = self.detailobject["property"].get("tenureInfo", {}).get("content", [])
|
||||
def leaseLeft(self) -> float | None:
|
||||
ds = self.detailobject["property"].get("tenureInfo",
|
||||
{}).get("content", [])
|
||||
for d in ds:
|
||||
if d["type"] == "lengthOfLease":
|
||||
matches = re.findall(r"(\d+\.?\d*)", d["value"])
|
||||
|
|
@ -250,7 +246,7 @@ class Listing:
|
|||
return (now - ds).days
|
||||
|
||||
@property
|
||||
def last_seen(self) -> int:
|
||||
def last_seen(self) -> int | None:
|
||||
if not self.path_last_seen_listing().exists():
|
||||
return None
|
||||
|
||||
|
|
@ -260,8 +256,9 @@ class Listing:
|
|||
return (datetime.datetime.now() - dt).days
|
||||
|
||||
@property
|
||||
def serviceCharge(self) -> float:
|
||||
ds = self.detailobject["property"].get("tenureInfo", {}).get("content", [])
|
||||
def serviceCharge(self) -> float | None:
|
||||
ds = self.detailobject["property"].get("tenureInfo",
|
||||
{}).get("content", [])
|
||||
for d in ds:
|
||||
if d["type"] == "annualServiceCharge":
|
||||
matches = re.findall(r"([\d,.]+)", d["value"])
|
||||
|
|
@ -276,8 +273,7 @@ class Listing:
|
|||
# aka new home
|
||||
try:
|
||||
return self.detailobject["property"]["development"]
|
||||
except:
|
||||
print(self.identifier)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
@property
|
||||
|
|
@ -294,39 +290,33 @@ class Listing:
|
|||
def dict_nicely(self):
|
||||
return {
|
||||
"identifier":
|
||||
self.identifier,
|
||||
self.identifier,
|
||||
"sqm_ocr":
|
||||
self.sqm_ocr,
|
||||
self.sqm_ocr,
|
||||
"price":
|
||||
self.price,
|
||||
self.price,
|
||||
"price_per_sqm":
|
||||
self.price_per_sqm,
|
||||
self.price_per_sqm,
|
||||
"url":
|
||||
self.url,
|
||||
self.url,
|
||||
"bedrooms":
|
||||
self.bedrooms,
|
||||
self.bedrooms,
|
||||
"travel_time_fastest":
|
||||
None if len(self.travel_time) == 0 else self.travel_time[0],
|
||||
None if len(self.travel_time) == 0 else self.travel_time[0],
|
||||
"travel_time_second":
|
||||
None if len(self.travel_time) < 2 else self.travel_time[1],
|
||||
None if len(self.travel_time) < 2 else self.travel_time[1],
|
||||
"lease_left":
|
||||
self.leaseLeft,
|
||||
self.leaseLeft,
|
||||
"service_charge":
|
||||
self.serviceCharge,
|
||||
self.serviceCharge,
|
||||
"development":
|
||||
self.development,
|
||||
self.development,
|
||||
"tenure_type":
|
||||
self.tenure_type,
|
||||
self.tenure_type,
|
||||
"updated_days":
|
||||
self.updateDaysAgo,
|
||||
self.updateDaysAgo,
|
||||
"status":
|
||||
self.status,
|
||||
self.status,
|
||||
"last_seen":
|
||||
self.last_seen,
|
||||
self.last_seen,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
listing_paths = sorted(list(pathlib.Path("data/rs").glob("*/listing.json")))
|
||||
listings = Listing.get_all_listings()
|
||||
print(listings[0].list_floorplans())
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue