calculate.py
scripts/calculate.py
#!/usr/bin/env python3
"""Calculate the Monday report from the normalized files. Every number in the report comes from here.
Usage: python3 scripts/calculate.py <normalized_dir> <thresholds.json> <out_dir>
Reads days.csv, groups.csv, maintenance.csv and meta.json (written by normalize.py).
Writes report.json (format: schemas/report.schema.json) and report.csv to <out_dir>.
Forward window: the business date through business date + 6 days (7 dates).
Last 7 days: business date - 7 through business date - 1.
"""
import csv
import json
import os
import sys
from datetime import date, timedelta
CSV_COLUMNS = [
"property_id", "week_start", "date", "occupancy_on_books", "rooms_available", "below_threshold",
"group_rooms_allocated", "group_rooms_picked_up", "rooms_with_open_maintenance", "source_file", "source_row",
]
def read_csv(path: str) -> list[dict]:
with open(path, newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
def number(text: str) -> float | None:
return float(text) if text != "" else None
def calculate(norm_dir: str, threshold: float) -> dict:
with open(os.path.join(norm_dir, "meta.json"), encoding="utf-8") as f:
meta = json.load(f)
days = {r["date"]: r for r in read_csv(os.path.join(norm_dir, "days.csv"))}
groups = read_csv(os.path.join(norm_dir, "groups.csv"))
maintenance = read_csv(os.path.join(norm_dir, "maintenance.csv"))
warnings = list(meta["warnings"])
business = date.fromisoformat(meta["business_date"])
forward = [(business + timedelta(days=i)).isoformat() for i in range(7)]
last = [(business - timedelta(days=i)).isoformat() for i in range(7, 0, -1)]
absent = [d for d in forward if d not in days or days[d]["rooms_available"] == ""]
if absent:
raise ValueError(f"no occupancy or rooms available for: {', '.join(absent)}")
derived = [d for d in forward if days[d]["availability_row"].startswith("derived")]
if derived:
warnings.append(
f"rooms_available for {', '.join(derived)} is derived from the inventory and the occupancy: "
"the export has no hotel-level availability row for that date."
)
# Group numbers per date: sum over every block in the grid. A blank pickup counts as 0.
allocated, picked_up, blank_pickup = {}, {}, 0
for r in groups:
allocated[r["date"]] = allocated.get(r["date"], 0) + int(number(r["allocated"]) or 0)
if r["picked_up"] == "":
blank_pickup += 1
picked_up[r["date"]] = picked_up.get(r["date"], 0) + int(number(r["picked_up"]) or 0)
if blank_pickup:
warnings.append(f"groups.csv has {blank_pickup} block/date rows with no pickup value. Counted as 0.")
blocks_in_grid = len({r["block"] for r in groups})
if blocks_in_grid < meta["blocks_listed"]:
warnings.append(
f"The group grid covers {blocks_in_grid} of {meta['blocks_listed']} blocks listed in the export. "
"Group numbers cover only the blocks in the grid."
)
# Open maintenance: distinct rooms with an UnResolved record. One count for all dates.
open_rooms = len({r["room"] for r in maintenance if r["status"].lower() == "unresolved"})
warnings.append(
"rooms_with_open_maintenance is one count for every date: the export has no end dates for maintenance records."
)
rows = []
for d in forward:
occ = round(float(days[d]["occupancy"]), 4)
rows.append({
"date": d,
"occupancy_on_books": occ,
"rooms_available": int(days[d]["rooms_available"]),
"below_threshold": occ < threshold,
"group_rooms_allocated": allocated.get(d, 0),
"group_rooms_picked_up": picked_up.get(d, 0),
"rooms_with_open_maintenance": open_rooms,
"source_file": days[d]["source_file"],
"source_row": f"{days[d]['occupancy_row']}; {days[d]['availability_row']}",
})
present = [float(days[d]["occupancy"]) for d in last if d in days]
if not present:
raise ValueError("no occupancy for the last 7 days")
if len(present) < 7:
warnings.append(f"The last 7 days summary uses {len(present)} days: the others are missing in the export.")
last_avg = round(sum(present) / len(present), 4)
next_avg = round(sum(r["occupancy_on_books"] for r in rows) / 7, 4)
return {
"business_date": meta["business_date"],
"property_id": meta["property_id"],
"threshold": threshold,
"rows": rows,
"last_7_days": {
"start_date": last[0],
"end_date": last[-1],
"days": len(present),
"occupancy_avg": last_avg,
"next_7_days_occupancy_avg": next_avg,
"change_vs_last_7_days": round(next_avg - last_avg, 4),
},
"warnings": warnings,
}
def main(argv: list[str]) -> int:
if len(argv) != 4:
print("Usage: python3 scripts/calculate.py <normalized_dir> <thresholds.json> <out_dir>", file=sys.stderr)
return 1
norm_dir, thresholds_path, out_dir = argv[1:]
try:
with open(thresholds_path, encoding="utf-8") as f:
threshold = float(json.load(f)["occupancy_on_books"])
report = calculate(norm_dir, threshold)
os.makedirs(out_dir, exist_ok=True)
with open(os.path.join(out_dir, "report.json"), "w", encoding="utf-8") as f:
json.dump(report, f, indent=2)
f.write("\n")
with open(os.path.join(out_dir, "report.csv"), "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
writer.writeheader()
for row in report["rows"]:
writer.writerow({
**row,
"property_id": report["property_id"],
"week_start": report["business_date"],
"below_threshold": str(row["below_threshold"]).lower(),
})
except (OSError, json.JSONDecodeError) as err:
print(f"Could not read or write files: {err}", file=sys.stderr)
return 1
except (KeyError, ValueError) as err:
print(f"Could not calculate the report: {err!r}", file=sys.stderr)
return 1
below = [r["date"] for r in report["rows"] if r["below_threshold"]]
print(f"Wrote report.json and report.csv to {out_dir}. Dates below {threshold}: {', '.join(below) or 'none'}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))