Spaces:
Sleeping
Sleeping
File size: 4,808 Bytes
15eb4b9 55c930e 15eb4b9 55c930e 15eb4b9 64a7d89 15eb4b9 d9b42dc 15eb4b9 55c930e 15eb4b9 55c930e 15eb4b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """Cost / size estimation from the no-fetch estimate job's output."""
from __future__ import annotations
import math
import re
from dataclasses import dataclass
from typing import Iterable, Optional
from . import config
_ESTIMATE_RE = re.compile(r"ESTIMATE\s+n_records=(\d+)\s+total_bytes=(\d+)")
@dataclass(frozen=True)
class Estimate:
n_records: int
total_bytes: int
flavor: str
@property
def total_gib(self) -> float:
return self.total_bytes / (1024 ** 3)
@property
def rec_per_s(self) -> int:
return config.FLAVOR_REC_PER_S.get(self.flavor, config.FLAVOR_REC_PER_S[config.DEFAULT_FLAVOR])
@property
def fetch_seconds(self) -> float:
if self.n_records <= 0:
return 0.0
return self.n_records / self.rec_per_s
@property
def n_processes(self) -> int:
return config.compute_processes(self.n_records, self.flavor)
@property
def merge_seconds(self) -> float:
# Only multi-process runs produce shards that need merging.
if self.n_processes <= 1:
return 0.0
return config.MERGE_BASE_S + self.total_bytes / config.MERGE_THROUGHPUT_BYTES_PER_S
@property
def duration_seconds(self) -> float:
# Billed time = container/install overhead + fetch + (merge if multi-process).
return config.CONTAINER_OVERHEAD_S + self.fetch_seconds + self.merge_seconds
@property
def hourly_usd(self) -> float:
return config.FLAVOR_HOURLY_USD.get(self.flavor, config.FLAVOR_HOURLY_USD[config.DEFAULT_FLAVOR])
@property
def cost_usd(self) -> float:
return self.duration_seconds / 3600.0 * self.hourly_usd
@property
def n_output_files(self) -> int:
if self.total_bytes <= 0:
return 0
return max(1, math.ceil(self.total_bytes / config.WARC_TARGET_SIZE))
def parse_estimate(logs: Iterable[str], flavor: str) -> Optional[Estimate]:
"""Find the last `ESTIMATE n_records=.. total_bytes=..` line in the job logs."""
found: Optional[Estimate] = None
for line in logs:
m = _ESTIMATE_RE.search(line)
if m:
found = Estimate(int(m.group(1)), int(m.group(2)), flavor)
return found
def _fmt_cost(usd: float) -> str:
if usd <= 0:
return "$0.00"
if usd < 0.01:
return "<$0.01"
return f"${usd:.2f}"
def _fmt_duration(seconds: float) -> str:
seconds = int(round(seconds))
if seconds < 60:
return f"{seconds}s"
if seconds < 3600:
return f"{seconds // 60}m {seconds % 60}s"
return f"{seconds // 3600}h {(seconds % 3600) // 60}m"
def actual_cost_usd(running_secs: int, flavor: str) -> float:
hourly = config.FLAVOR_HOURLY_USD.get(flavor, config.FLAVOR_HOURLY_USD[config.DEFAULT_FLAVOR])
return max(0, running_secs) / 3600.0 * hourly
def format_cost_report(running_secs: int, flavor: str) -> str:
"""One-line actual-cost report shown after a job finishes (HF bills running time)."""
return (
f"💰 **Actual cost:** ~{_fmt_cost(actual_cost_usd(running_secs, flavor))} "
f"({running_secs}s of runtime on `{flavor}`)"
)
def format_estimate(est: Estimate) -> str:
"""Markdown summary shown to the user before launching the fetch job."""
if est.n_records == 0:
return (
"### ⚠️ No matching records\n"
"The index query matched **0 records** for these hostnames/domains in the "
"selected crawl(s), so there is nothing to repackage.\n\n"
"Things to check:\n"
"- **Hostnames match exactly** (e.g. `www.example.com`). To include all "
"subdomains of a site, use the **Registered domains** field instead "
"(e.g. `example.com`).\n"
"- Try a different / additional crawl."
)
runtime_parts = [f"~{_fmt_duration(config.CONTAINER_OVERHEAD_S)} setup",
f"~{_fmt_duration(est.fetch_seconds)} fetch"]
if est.merge_seconds > 0:
runtime_parts.append(f"~{_fmt_duration(est.merge_seconds)} merge ({est.n_processes} shards)")
return (
"### 💸 Cost estimate\n"
f"- **Records to fetch:** {est.n_records:,}\n"
f"- **Total data size:** {est.total_gib:.2f} GiB\n"
f"- **Output:** ~{est.n_output_files} WARC file(s) (≤1 GiB each)\n"
f"- **Hardware:** `{est.flavor}` (${est.hourly_usd:.2f}/hr, ~{est.rec_per_s:,} rec/s)\n"
f"- **Estimated runtime:** ~{_fmt_duration(est.duration_seconds)} "
f"({' + '.join(runtime_parts)})\n"
f"- **Estimated cost:** **~{_fmt_cost(est.cost_usd)}**\n\n"
"_Estimate assumes conservative benchmark throughput; actual cost is billed "
"per second of job runtime. The estimate job itself costs a few cents._"
)
|