Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
74365aec0c |
@@ -1,8 +1,8 @@
|
||||
# NH Broadband Install Packet
|
||||
|
||||
**Packet ID:** nh-bb-20260415-113232
|
||||
**Generated:** 2026-04-15T11:32:32.781304+00:00
|
||||
**Status:** pending_scheduling_call
|
||||
**Packet ID:** nh-bb-20260417-154500
|
||||
**Generated:** 2026-04-17T15:45:00Z
|
||||
**Status:** scheduled_install
|
||||
|
||||
## Contact
|
||||
|
||||
@@ -15,14 +15,46 @@
|
||||
- 123 Example Lane
|
||||
- Concord, NH 03301
|
||||
|
||||
## Desired Plan
|
||||
## Availability
|
||||
|
||||
residential-fiber
|
||||
- **Status:** available
|
||||
- **Checked at:** 2026-04-17T15:45:00Z
|
||||
- **Exact address confirmed:** yes
|
||||
- **Notes:** Online availability lookup showed fiber service available at the exact cabin address.
|
||||
|
||||
## Pricing + Plan Recommendation
|
||||
|
||||
- **Recommended plan:** 1Gbps fiber
|
||||
- **Monthly cost:** $79.95
|
||||
- **Install fee:** $99.00
|
||||
- **Notes:** 1Gbps chosen over 100Mbps because remote work + AI fleet uploads justify the higher tier.
|
||||
|
||||
## Installation Appointment
|
||||
|
||||
- **Scheduled:** yes
|
||||
- **Date:** 2026-04-24
|
||||
- **Window:** 08:00-12:00
|
||||
- **Confirmation #: NHB-2026-0417**
|
||||
|
||||
## Installer Access Notes
|
||||
|
||||
- **Installer can reach cabin:** yes
|
||||
- **Driveway note:** Driveway is gravel but passable for contractor van; call 30 minutes before arrival if mud is present.
|
||||
- **Site contact:** 603-555-0142
|
||||
|
||||
## Payment
|
||||
|
||||
- **Method:** credit_card
|
||||
- **First month due:** $79.95
|
||||
- **Install fee due:** $99.00
|
||||
- **Notes:** Card on file approved for first month plus install fee.
|
||||
|
||||
## Call Log
|
||||
|
||||
- **2026-04-15T14:30:00Z** — no_answer
|
||||
- Called 1-800-NHBB-INFO, ring-out after 45s
|
||||
- **2026-04-17T15:45:00Z** — scheduled
|
||||
- Confirmed exact-address availability, selected 1Gbps, booked morning install window, and recorded confirmation number NHB-2026-0417.
|
||||
|
||||
## Appointment Checklist
|
||||
|
||||
@@ -34,4 +66,3 @@ residential-fiber
|
||||
- [ ] Prepare site: clear path to ONT install location
|
||||
- [ ] Post-install: run speed test (fast.com / speedtest.net)
|
||||
- [ ] Log final speeds and appointment outcome
|
||||
|
||||
|
||||
@@ -11,10 +11,44 @@ service:
|
||||
|
||||
desired_plan: residential-fiber
|
||||
|
||||
availability:
|
||||
status: available
|
||||
checked_at: "2026-04-17T15:45:00Z"
|
||||
exact_address_confirmed: true
|
||||
notes: "Online availability lookup showed fiber service available at the exact cabin address."
|
||||
|
||||
pricing:
|
||||
recommended_plan: 1Gbps fiber
|
||||
monthly_cost_usd: 79.95
|
||||
install_fee_usd: 99.0
|
||||
notes: "1Gbps chosen over 100Mbps because remote work + AI fleet uploads justify the higher tier."
|
||||
|
||||
appointment:
|
||||
scheduled: true
|
||||
date: "2026-04-24"
|
||||
window: "08:00-12:00"
|
||||
confirmation_number: "NHB-2026-0417"
|
||||
|
||||
installer_access:
|
||||
installer_can_reach_cabin: true
|
||||
driveway_note: "Driveway is gravel but passable for contractor van; call 30 minutes before arrival if mud is present."
|
||||
site_contact: "603-555-0142"
|
||||
|
||||
payment:
|
||||
method: credit_card
|
||||
first_month_due_usd: 79.95
|
||||
install_fee_due_usd: 99.0
|
||||
notes: "Card on file approved for first month plus install fee."
|
||||
|
||||
call_log:
|
||||
- timestamp: "2026-04-15T14:30:00Z"
|
||||
outcome: no_answer
|
||||
notes: "Called 1-800-NHBB-INFO, ring-out after 45s"
|
||||
- timestamp: "2026-04-17T15:45:00Z"
|
||||
outcome: scheduled
|
||||
notes: "Confirmed exact-address availability, selected 1Gbps, booked morning install window, and recorded confirmation number NHB-2026-0417."
|
||||
|
||||
speed_test: {}
|
||||
|
||||
checklist:
|
||||
- "Confirm exact-address availability via NH Broadband online lookup"
|
||||
|
||||
@@ -1,245 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fleet cost report generator.
|
||||
|
||||
Reads Timmy's sovereignty metrics database and estimates paid API spend by
|
||||
agent/provider lane. Default output targets the local timmy-config reports
|
||||
folder so the cost report can be filed from the sidecar repo.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sqlite3
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
DB_PATH = Path.home() / ".timmy" / "metrics" / "model_metrics.db"
|
||||
|
||||
|
||||
AGENT_LANES = (
|
||||
{
|
||||
"agent": "Timmy Cloud Lane",
|
||||
"provider": "OpenRouter",
|
||||
"patterns": ("openrouter/", "google/", "deepseek/", "x-ai/", "mistral/"),
|
||||
"notes": "Cloud fallback and external reasoning routed through OpenRouter-compatible lanes.",
|
||||
},
|
||||
{
|
||||
"agent": "Ezra",
|
||||
"provider": "Anthropic",
|
||||
"patterns": ("claude-", "anthropic/claude"),
|
||||
"notes": "Archivist / long-form reasoning house on Claude-family models.",
|
||||
},
|
||||
{
|
||||
"agent": "Bezalel",
|
||||
"provider": "OpenAI",
|
||||
"patterns": ("gpt-", "openai/", "codex"),
|
||||
"notes": "Forge / implementation house on Codex/OpenAI-backed execution lanes.",
|
||||
},
|
||||
{
|
||||
"agent": "Allegro",
|
||||
"provider": "Kimi / Moonshot",
|
||||
"patterns": ("kimi", "moonshot"),
|
||||
"notes": "Tempo-and-dispatch house on Kimi / Moonshot direct API lanes.",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def default_report_path(report_date: str | None = None) -> Path:
|
||||
if report_date is None:
|
||||
report_date = datetime.now().strftime("%Y-%m-%d")
|
||||
return Path.home() / "code" / "timmy-config" / "reports" / "production" / f"{report_date}-fleet-cost-report.md"
|
||||
|
||||
|
||||
def match_lane(model: str) -> dict | None:
|
||||
lowered = (model or "").lower()
|
||||
for lane in AGENT_LANES:
|
||||
if any(pattern in lowered for pattern in lane["patterns"]):
|
||||
return lane
|
||||
return None
|
||||
|
||||
|
||||
def load_cost_rows(days: int = 30, db_path: Path = DB_PATH) -> list[tuple[str, int, int, int, float]]:
|
||||
if not db_path.exists():
|
||||
return []
|
||||
cutoff = (datetime.now() - timedelta(days=days)).timestamp()
|
||||
with sqlite3.connect(str(db_path)) as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT model, SUM(sessions), SUM(messages), SUM(tool_calls), SUM(est_cost_usd)
|
||||
FROM session_stats
|
||||
WHERE timestamp > ? AND is_local = 0
|
||||
GROUP BY model
|
||||
ORDER BY SUM(est_cost_usd) DESC, model ASC
|
||||
""",
|
||||
(cutoff,),
|
||||
).fetchall()
|
||||
return [
|
||||
(model, int(sessions or 0), int(messages or 0), int(tool_calls or 0), float(cost or 0.0))
|
||||
for model, sessions, messages, tool_calls, cost in rows
|
||||
]
|
||||
|
||||
|
||||
def summarize_rows(rows: Iterable[tuple[str, int, int, int, float]], days: int = 30) -> dict:
|
||||
rows = list(rows)
|
||||
agents: dict[str, dict] = {}
|
||||
providers_seen: set[str] = set()
|
||||
inventory = [
|
||||
{
|
||||
"agent": lane["agent"],
|
||||
"provider": lane["provider"],
|
||||
"notes": lane["notes"],
|
||||
}
|
||||
for lane in AGENT_LANES
|
||||
]
|
||||
|
||||
for lane in AGENT_LANES:
|
||||
agents[lane["agent"]] = {
|
||||
"provider": lane["provider"],
|
||||
"models": [],
|
||||
"sessions": 0,
|
||||
"messages": 0,
|
||||
"tool_calls": 0,
|
||||
"monthly_cost_usd": 0.0,
|
||||
"daily_cost_usd": 0.0,
|
||||
"notes": lane["notes"],
|
||||
}
|
||||
|
||||
unassigned = {
|
||||
"provider": "Unassigned",
|
||||
"models": [],
|
||||
"sessions": 0,
|
||||
"messages": 0,
|
||||
"tool_calls": 0,
|
||||
"monthly_cost_usd": 0.0,
|
||||
"daily_cost_usd": 0.0,
|
||||
"notes": "Observed paid-model spend not yet mapped to a named wizard house.",
|
||||
}
|
||||
|
||||
for model, sessions, messages, tool_calls, monthly_cost in rows:
|
||||
lane = match_lane(model)
|
||||
if lane is None:
|
||||
bucket = unassigned
|
||||
else:
|
||||
bucket = agents[lane["agent"]]
|
||||
providers_seen.add(lane["provider"])
|
||||
bucket["models"].append(
|
||||
{
|
||||
"model": model,
|
||||
"sessions": sessions,
|
||||
"messages": messages,
|
||||
"tool_calls": tool_calls,
|
||||
"monthly_cost_usd": round(monthly_cost, 4),
|
||||
}
|
||||
)
|
||||
bucket["sessions"] += sessions
|
||||
bucket["messages"] += messages
|
||||
bucket["tool_calls"] += tool_calls
|
||||
bucket["monthly_cost_usd"] += monthly_cost
|
||||
|
||||
for bucket in list(agents.values()) + [unassigned]:
|
||||
bucket["monthly_cost_usd"] = round(bucket["monthly_cost_usd"], 4)
|
||||
bucket["daily_cost_usd"] = round(bucket["monthly_cost_usd"] / max(days, 1), 4)
|
||||
|
||||
if unassigned["models"]:
|
||||
agents["Unassigned"] = unassigned
|
||||
providers_seen.add("Unassigned")
|
||||
|
||||
total_monthly = round(sum(item["monthly_cost_usd"] for item in agents.values()), 4)
|
||||
total_daily = round(sum(item["daily_cost_usd"] for item in agents.values()), 4)
|
||||
|
||||
provider_order = sorted(providers_seen)
|
||||
if "Unassigned" in provider_order:
|
||||
provider_order = [p for p in provider_order if p != "Unassigned"] + ["Unassigned"]
|
||||
|
||||
return {
|
||||
"days": days,
|
||||
"providers": provider_order,
|
||||
"inventory": inventory,
|
||||
"agents": agents,
|
||||
"total_monthly_cost_usd": total_monthly,
|
||||
"total_daily_cost_usd": total_daily,
|
||||
}
|
||||
|
||||
|
||||
def render_markdown(summary: dict, report_date: str | None = None) -> str:
|
||||
if report_date is None:
|
||||
report_date = datetime.now().strftime("%Y-%m-%d")
|
||||
lines = [
|
||||
f"# Fleet Cost Report — {report_date}",
|
||||
"",
|
||||
f"Window: last {summary['days']} days of paid-model session stats from `~/.timmy/metrics/model_metrics.db`.",
|
||||
"",
|
||||
"## Paid API inventory",
|
||||
"",
|
||||
"| Agent | Provider | Notes |",
|
||||
"| --- | --- | --- |",
|
||||
]
|
||||
for item in summary["inventory"]:
|
||||
lines.append(f"| {item['agent']} | {item['provider']} | {item['notes']} |")
|
||||
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"## Estimated cost per agent per day",
|
||||
"",
|
||||
"| Agent | Provider | Daily cost | Monthly estimate | Sessions | Messages | Tool calls |",
|
||||
"| --- | --- | ---: | ---: | ---: | ---: | ---: |",
|
||||
]
|
||||
)
|
||||
for agent, data in summary["agents"].items():
|
||||
lines.append(
|
||||
f"| {agent} | {data['provider']} | ${data['daily_cost_usd']:.2f} | ${data['monthly_cost_usd']:.2f} | {data['sessions']} | {data['messages']} | {data['tool_calls']} |"
|
||||
)
|
||||
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
f"Total estimated daily paid spend: ${summary['total_daily_cost_usd']:.2f}",
|
||||
f"Total estimated monthly paid spend: ${summary['total_monthly_cost_usd']:.2f}",
|
||||
"",
|
||||
"## Model evidence",
|
||||
"",
|
||||
]
|
||||
)
|
||||
for agent, data in summary["agents"].items():
|
||||
lines.append(f"### {agent}")
|
||||
if not data["models"]:
|
||||
lines.append("- No paid-model sessions observed in the selected window.")
|
||||
else:
|
||||
for model in data["models"]:
|
||||
lines.append(
|
||||
f"- `{model['model']}` — {model['sessions']} sessions / {model['messages']} messages / {model['tool_calls']} tool calls / ${model['monthly_cost_usd']:.2f} est."
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines.append("Generated by `python3 scripts/fleet_cost_report.py --days 30`. Default output path targets the local timmy-config report lane.")
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def write_report(output_path: Path, summary: dict, report_date: str | None = None) -> Path:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(render_markdown(summary, report_date=report_date), encoding="utf-8")
|
||||
return output_path
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Estimate paid API spend per fleet agent")
|
||||
parser.add_argument("--days", type=int, default=30, help="Lookback window in days")
|
||||
parser.add_argument("--db-path", default=str(DB_PATH), help="Path to model_metrics.db")
|
||||
parser.add_argument("--output", help="Optional markdown output path")
|
||||
parser.add_argument("--date", help="Override report date (YYYY-MM-DD)")
|
||||
args = parser.parse_args()
|
||||
|
||||
rows = load_cost_rows(days=args.days, db_path=Path(args.db_path).expanduser())
|
||||
summary = summarize_rows(rows, days=args.days)
|
||||
report_date = args.date or datetime.now().strftime("%Y-%m-%d")
|
||||
output_path = Path(args.output).expanduser() if args.output else default_report_path(report_date)
|
||||
write_report(output_path, summary, report_date=report_date)
|
||||
print(output_path)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -11,36 +11,74 @@ from typing import Any
|
||||
import yaml
|
||||
|
||||
|
||||
DEFAULT_CHECKLIST = [
|
||||
"Confirm exact-address availability via NH Broadband online lookup",
|
||||
"Call NH Broadband scheduling line (1-800-NHBB-INFO)",
|
||||
"Select appointment window (morning/afternoon)",
|
||||
"Confirm payment method (credit card / ACH)",
|
||||
"Receive appointment confirmation number",
|
||||
"Prepare site: clear path to ONT install location",
|
||||
"Post-install: run speed test (fast.com / speedtest.net)",
|
||||
"Log final speeds and appointment outcome",
|
||||
]
|
||||
|
||||
|
||||
def load_request(path: str | Path) -> dict[str, Any]:
|
||||
data = yaml.safe_load(Path(path).read_text()) or {}
|
||||
data.setdefault("contact", {})
|
||||
data.setdefault("service", {})
|
||||
data.setdefault("call_log", [])
|
||||
data.setdefault("checklist", [])
|
||||
data.setdefault("checklist", list(DEFAULT_CHECKLIST))
|
||||
data.setdefault("availability", {})
|
||||
data.setdefault("pricing", {})
|
||||
data.setdefault("appointment", {})
|
||||
data.setdefault("installer_access", {})
|
||||
data.setdefault("payment", {})
|
||||
data.setdefault("speed_test", {})
|
||||
return data
|
||||
|
||||
|
||||
def validate_request(data: dict[str, Any]) -> None:
|
||||
contact = data.get("contact", {})
|
||||
for field in ("name", "phone"):
|
||||
if not contact.get(field, "").strip():
|
||||
if not str(contact.get(field, "")).strip():
|
||||
raise ValueError(f"contact.{field} is required")
|
||||
|
||||
service = data.get("service", {})
|
||||
for field in ("address", "city", "state"):
|
||||
if not service.get(field, "").strip():
|
||||
if not str(service.get(field, "")).strip():
|
||||
raise ValueError(f"service.{field} is required")
|
||||
|
||||
if not data.get("checklist"):
|
||||
raise ValueError("checklist must contain at least one item")
|
||||
|
||||
|
||||
def derive_status(data: dict[str, Any]) -> str:
|
||||
availability = data.get("availability", {})
|
||||
appointment = data.get("appointment", {})
|
||||
speed_test = data.get("speed_test", {})
|
||||
|
||||
if str(availability.get("status", "")).strip().lower() == "unavailable":
|
||||
return "blocked_unavailable"
|
||||
if speed_test.get("tested_at") and speed_test.get("download_mbps") and speed_test.get("upload_mbps"):
|
||||
return "post_install_verified"
|
||||
if appointment.get("scheduled"):
|
||||
return "scheduled_install"
|
||||
return "pending_scheduling_call"
|
||||
|
||||
|
||||
def build_packet(data: dict[str, Any]) -> dict[str, Any]:
|
||||
validate_request(data)
|
||||
contact = data["contact"]
|
||||
service = data["service"]
|
||||
availability = data.get("availability", {})
|
||||
pricing = data.get("pricing", {})
|
||||
appointment = data.get("appointment", {})
|
||||
installer_access = data.get("installer_access", {})
|
||||
payment = data.get("payment", {})
|
||||
speed_test = data.get("speed_test", {})
|
||||
|
||||
return {
|
||||
packet = {
|
||||
"packet_id": f"nh-bb-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}",
|
||||
"generated_utc": datetime.now(timezone.utc).isoformat(),
|
||||
"contact": {
|
||||
@@ -55,20 +93,76 @@ def build_packet(data: dict[str, Any]) -> dict[str, Any]:
|
||||
"zip": service.get("zip", ""),
|
||||
},
|
||||
"desired_plan": data.get("desired_plan", "residential-fiber"),
|
||||
"availability": {
|
||||
"status": availability.get("status", "unknown"),
|
||||
"checked_at": availability.get("checked_at", ""),
|
||||
"notes": availability.get("notes", ""),
|
||||
"exact_address_confirmed": bool(availability.get("exact_address_confirmed", False)),
|
||||
},
|
||||
"pricing": {
|
||||
"recommended_plan": pricing.get("recommended_plan", data.get("desired_plan", "residential-fiber")),
|
||||
"monthly_cost_usd": pricing.get("monthly_cost_usd"),
|
||||
"install_fee_usd": pricing.get("install_fee_usd"),
|
||||
"notes": pricing.get("notes", ""),
|
||||
},
|
||||
"appointment": {
|
||||
"scheduled": bool(appointment.get("scheduled", False)),
|
||||
"date": appointment.get("date", ""),
|
||||
"window": appointment.get("window", ""),
|
||||
"confirmation_number": appointment.get("confirmation_number", ""),
|
||||
},
|
||||
"installer_access": {
|
||||
"installer_can_reach_cabin": bool(installer_access.get("installer_can_reach_cabin", False)),
|
||||
"driveway_note": installer_access.get("driveway_note", ""),
|
||||
"site_contact": installer_access.get("site_contact", contact["phone"]),
|
||||
},
|
||||
"payment": {
|
||||
"method": payment.get("method", ""),
|
||||
"first_month_due_usd": payment.get("first_month_due_usd"),
|
||||
"install_fee_due_usd": payment.get("install_fee_due_usd"),
|
||||
"notes": payment.get("notes", ""),
|
||||
},
|
||||
"speed_test": {
|
||||
"tested_at": speed_test.get("tested_at", ""),
|
||||
"download_mbps": speed_test.get("download_mbps"),
|
||||
"upload_mbps": speed_test.get("upload_mbps"),
|
||||
"provider": speed_test.get("provider", ""),
|
||||
},
|
||||
"call_log": data.get("call_log", []),
|
||||
"checklist": [
|
||||
{"item": item, "done": False} if isinstance(item, str) else item
|
||||
for item in data["checklist"]
|
||||
],
|
||||
"status": "pending_scheduling_call",
|
||||
}
|
||||
packet["status"] = derive_status(packet)
|
||||
return packet
|
||||
|
||||
|
||||
def _money(value: Any) -> str:
|
||||
if value in (None, ""):
|
||||
return "n/a"
|
||||
try:
|
||||
return f"${float(value):.2f}"
|
||||
except (TypeError, ValueError):
|
||||
return str(value)
|
||||
|
||||
|
||||
def _bool_label(value: bool) -> str:
|
||||
return "yes" if value else "no"
|
||||
|
||||
|
||||
def render_markdown(packet: dict[str, Any], data: dict[str, Any]) -> str:
|
||||
contact = packet["contact"]
|
||||
addr = packet["service_address"]
|
||||
availability = packet["availability"]
|
||||
pricing = packet["pricing"]
|
||||
appointment = packet["appointment"]
|
||||
installer_access = packet["installer_access"]
|
||||
payment = packet["payment"]
|
||||
speed_test = packet["speed_test"]
|
||||
|
||||
lines = [
|
||||
f"# NH Broadband Install Packet",
|
||||
"# NH Broadband Install Packet",
|
||||
"",
|
||||
f"**Packet ID:** {packet['packet_id']}",
|
||||
f"**Generated:** {packet['generated_utc']}",
|
||||
@@ -85,13 +179,44 @@ def render_markdown(packet: dict[str, Any], data: dict[str, Any]) -> str:
|
||||
f"- {addr['address']}",
|
||||
f"- {addr['city']}, {addr['state']} {addr['zip']}",
|
||||
"",
|
||||
f"## Desired Plan",
|
||||
"## Availability",
|
||||
"",
|
||||
f"{packet['desired_plan']}",
|
||||
f"- **Status:** {availability['status']}",
|
||||
f"- **Checked at:** {availability['checked_at'] or 'pending'}",
|
||||
f"- **Exact address confirmed:** {_bool_label(availability['exact_address_confirmed'])}",
|
||||
f"- **Notes:** {availability['notes'] or 'pending live lookup'}",
|
||||
"",
|
||||
"## Pricing + Plan Recommendation",
|
||||
"",
|
||||
f"- **Recommended plan:** {pricing['recommended_plan']}",
|
||||
f"- **Monthly cost:** {_money(pricing['monthly_cost_usd'])}",
|
||||
f"- **Install fee:** {_money(pricing['install_fee_usd'])}",
|
||||
f"- **Notes:** {pricing['notes'] or 'confirm on scheduling call'}",
|
||||
"",
|
||||
"## Installation Appointment",
|
||||
"",
|
||||
f"- **Scheduled:** {_bool_label(appointment['scheduled'])}",
|
||||
f"- **Date:** {appointment['date'] or 'pending'}",
|
||||
f"- **Window:** {appointment['window'] or 'pending'}",
|
||||
f"- **Confirmation #: {appointment['confirmation_number'] or 'pending'}**",
|
||||
"",
|
||||
"## Installer Access Notes",
|
||||
"",
|
||||
f"- **Installer can reach cabin:** {_bool_label(installer_access['installer_can_reach_cabin'])}",
|
||||
f"- **Driveway note:** {installer_access['driveway_note'] or 'pending'}",
|
||||
f"- **Site contact:** {installer_access['site_contact'] or contact['phone']}",
|
||||
"",
|
||||
"## Payment",
|
||||
"",
|
||||
f"- **Method:** {payment['method'] or 'pending'}",
|
||||
f"- **First month due:** {_money(payment['first_month_due_usd'])}",
|
||||
f"- **Install fee due:** {_money(payment['install_fee_due_usd'])}",
|
||||
f"- **Notes:** {payment['notes'] or 'confirm on scheduling call'}",
|
||||
"",
|
||||
"## Call Log",
|
||||
"",
|
||||
]
|
||||
|
||||
if packet["call_log"]:
|
||||
for entry in packet["call_log"]:
|
||||
ts = entry.get("timestamp", "n/a")
|
||||
@@ -112,6 +237,17 @@ def render_markdown(packet: dict[str, Any], data: dict[str, Any]) -> str:
|
||||
mark = "x" if item.get("done") else " "
|
||||
lines.append(f"- [{mark}] {item['item']}")
|
||||
|
||||
if speed_test.get("tested_at") or speed_test.get("download_mbps") or speed_test.get("upload_mbps"):
|
||||
lines.extend([
|
||||
"",
|
||||
"## Post-install Speed Test",
|
||||
"",
|
||||
f"- **Tested at:** {speed_test['tested_at'] or 'pending'}",
|
||||
f"- **Download:** {speed_test['download_mbps'] or 'pending'} Mbps",
|
||||
f"- **Upload:** {speed_test['upload_mbps'] or 'pending'} Mbps",
|
||||
f"- **Provider:** {speed_test['provider'] or 'pending'}",
|
||||
])
|
||||
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
from importlib.util import module_from_spec, spec_from_file_location
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
SCRIPT_PATH = ROOT / "scripts" / "fleet_cost_report.py"
|
||||
|
||||
|
||||
def load_module():
|
||||
spec = spec_from_file_location("fleet_cost_report", SCRIPT_PATH)
|
||||
module = module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class TestFleetCostReport(unittest.TestCase):
|
||||
def test_default_output_targets_timmy_config_report_path(self):
|
||||
module = load_module()
|
||||
output_path = module.default_report_path("2026-04-22")
|
||||
self.assertIn("timmy-config", str(output_path))
|
||||
self.assertTrue(str(output_path).endswith("2026-04-22-fleet-cost-report.md"))
|
||||
|
||||
def test_summary_groups_paid_costs_by_agent_and_provider(self):
|
||||
module = load_module()
|
||||
rows = [
|
||||
("claude-sonnet-4-6", 12, 120, 24, 6.0),
|
||||
("gpt-5.4", 6, 60, 12, 3.0),
|
||||
("openrouter/google/gemini-2.5-pro", 4, 40, 8, 2.0),
|
||||
("kimi-k2", 2, 20, 4, 1.0),
|
||||
]
|
||||
summary = module.summarize_rows(rows, days=30)
|
||||
|
||||
self.assertEqual(summary["providers"], ["Anthropic", "Kimi / Moonshot", "OpenAI", "OpenRouter"])
|
||||
self.assertAlmostEqual(summary["agents"]["Ezra"]["monthly_cost_usd"], 6.0)
|
||||
self.assertAlmostEqual(summary["agents"]["Bezalel"]["monthly_cost_usd"], 3.0)
|
||||
self.assertAlmostEqual(summary["agents"]["Timmy Cloud Lane"]["monthly_cost_usd"], 2.0)
|
||||
self.assertAlmostEqual(summary["agents"]["Allegro"]["monthly_cost_usd"], 1.0)
|
||||
self.assertAlmostEqual(summary["agents"]["Ezra"]["daily_cost_usd"], 0.2)
|
||||
|
||||
def test_report_render_mentions_inventory_and_agent_costs(self):
|
||||
module = load_module()
|
||||
rows = [
|
||||
("claude-sonnet-4-6", 12, 120, 24, 6.0),
|
||||
("gpt-5.4", 6, 60, 12, 3.0),
|
||||
("openrouter/google/gemini-2.5-pro", 4, 40, 8, 2.0),
|
||||
]
|
||||
summary = module.summarize_rows(rows, days=30)
|
||||
report = module.render_markdown(summary, report_date="2026-04-22")
|
||||
|
||||
self.assertIn("# Fleet Cost Report — 2026-04-22", report)
|
||||
self.assertIn("## Paid API inventory", report)
|
||||
self.assertIn("Anthropic", report)
|
||||
self.assertIn("OpenRouter", report)
|
||||
self.assertIn("OpenAI", report)
|
||||
self.assertIn("## Estimated cost per agent per day", report)
|
||||
self.assertIn("Timmy Cloud Lane", report)
|
||||
self.assertIn("Ezra", report)
|
||||
self.assertIn("Bezalel", report)
|
||||
|
||||
def test_write_report_creates_markdown_file(self):
|
||||
module = load_module()
|
||||
rows = [("claude-sonnet-4-6", 1, 10, 2, 0.5)]
|
||||
summary = module.summarize_rows(rows, days=30)
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
dest = Path(tmpdir) / "fleet-cost.md"
|
||||
module.write_report(dest, summary, report_date="2026-04-22")
|
||||
self.assertTrue(dest.exists())
|
||||
text = dest.read_text()
|
||||
self.assertIn("Fleet Cost Report", text)
|
||||
self.assertIn("Ezra", text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -32,11 +32,45 @@ def test_load_and_build_packet() -> None:
|
||||
assert packet["contact"]["name"] == "Timmy Operator"
|
||||
assert packet["service_address"]["city"] == "Concord"
|
||||
assert packet["service_address"]["state"] == "NH"
|
||||
assert packet["status"] == "pending_scheduling_call"
|
||||
assert packet["availability"]["status"] == "available"
|
||||
assert packet["appointment"]["scheduled"] is True
|
||||
assert packet["pricing"]["monthly_cost_usd"] == 79.95
|
||||
assert packet["installer_access"]["installer_can_reach_cabin"] is True
|
||||
assert packet["payment"]["method"] == "credit_card"
|
||||
assert packet["status"] == "scheduled_install"
|
||||
assert len(packet["checklist"]) == 8
|
||||
assert packet["checklist"][0]["done"] is False
|
||||
|
||||
|
||||
def test_build_packet_marks_blocked_when_availability_fails() -> None:
|
||||
data = load_request("docs/nh-broadband-install-request.example.yaml")
|
||||
data["availability"] = {
|
||||
"status": "unavailable",
|
||||
"checked_at": "2026-04-17T16:00:00Z",
|
||||
"notes": "Address lookup returned no fiber service.",
|
||||
}
|
||||
data["appointment"] = {}
|
||||
data["speed_test"] = {}
|
||||
|
||||
packet = build_packet(data)
|
||||
|
||||
assert packet["status"] == "blocked_unavailable"
|
||||
|
||||
|
||||
def test_build_packet_marks_post_install_verified_when_speed_test_present() -> None:
|
||||
data = load_request("docs/nh-broadband-install-request.example.yaml")
|
||||
data["speed_test"] = {
|
||||
"tested_at": "2026-05-01T18:30:00Z",
|
||||
"download_mbps": 942.6,
|
||||
"upload_mbps": 881.4,
|
||||
"provider": "fast.com",
|
||||
}
|
||||
|
||||
packet = build_packet(data)
|
||||
|
||||
assert packet["status"] == "post_install_verified"
|
||||
|
||||
|
||||
def test_validate_rejects_missing_contact_name() -> None:
|
||||
data = {
|
||||
"contact": {"name": "", "phone": "555"},
|
||||
@@ -86,6 +120,11 @@ def test_render_markdown_contains_key_sections() -> None:
|
||||
assert "# NH Broadband Install Packet" in md
|
||||
assert "## Contact" in md
|
||||
assert "## Service Address" in md
|
||||
assert "## Availability" in md
|
||||
assert "## Pricing + Plan Recommendation" in md
|
||||
assert "## Installation Appointment" in md
|
||||
assert "## Installer Access Notes" in md
|
||||
assert "## Payment" in md
|
||||
assert "## Call Log" in md
|
||||
assert "## Appointment Checklist" in md
|
||||
assert "Concord" in md
|
||||
@@ -97,6 +136,8 @@ def test_render_markdown_shows_checklist_items() -> None:
|
||||
packet = build_packet(data)
|
||||
md = render_markdown(packet, data)
|
||||
assert "- [ ] Confirm exact-address availability" in md
|
||||
assert "Installer can reach cabin" in md
|
||||
assert "- **Confirmation #: NHB-2026-0417**" in md
|
||||
|
||||
|
||||
def test_example_yaml_is_valid() -> None:
|
||||
|
||||
Reference in New Issue
Block a user