Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Per-case cost of a self-hosted Text-to-Scene run from the token counts its agent logged (user, 2026-10-16: | |
| "Gemma的价格可以估算一下？"). | |
| The OpenRouter-loop agent writes every model call's usage to <attempt>/episode/openrouter_r*_raw.jsonl | |
| (`llm_request` items: input_tokens, cached_input_tokens, output_tokens, which include reasoning_output_tokens). | |
| A self-hosted model is not billed (cost_usd_micros 0), so the internal table has no cost for it; this prices the | |
| logged tokens at the model's OpenRouter list price. The kept attempt of each case is <run_dir>/cases/<case>/final, | |
| as for every other model. Read-only; JSON on stdout: | |
| ssh <host> python3 - <in $/M> <cached in $/M> <out $/M> <run_dir> ... < scripts/t2s_token_cost.py > ../t2s-src/<name>_cost.json | |
| """ | |
| import glob | |
| import json | |
| import os | |
| import statistics | |
| import sys | |
| P_IN, P_CACHED, P_OUT = (float(x) for x in sys.argv[1:4]) | |
| RUNS = sys.argv[4:] | |
| out, problems = {}, [] | |
| for rd in RUNS: | |
| for cdir in sorted(glob.glob(os.path.join(rd, "cases", "*-long"))): | |
| case = os.path.basename(cdir) | |
| att = os.path.realpath(os.path.join(cdir, "final")) | |
| streams = sorted(glob.glob(os.path.join(att, "episode", "openrouter_r*_raw.jsonl"))) | |
| if not streams: | |
| continue | |
| tok = {"input": 0, "cached": 0, "output": 0, "reasoning": 0} | |
| calls = 0 | |
| for path in streams: | |
| for line in open(path, errors="ignore"): | |
| if '"llm_request"' not in line: | |
| continue | |
| try: | |
| item = json.loads(line).get("item") or {} | |
| except ValueError: | |
| continue | |
| u = item.get("usage") if item.get("type") == "llm_request" else None | |
| if not u: | |
| continue | |
| inp, cached, outp, rsn = (int(u.get(k) or 0) for k in | |
| ("input_tokens", "cached_input_tokens", "output_tokens", "reasoning_output_tokens")) | |
| if cached > inp or rsn > outp: | |
| problems.append("%s: usage outside the checked semantics %s" % (case, u)) | |
| tok["input"] += inp; tok["cached"] += cached; tok["output"] += outp; tok["reasoning"] += rsn | |
| calls += 1 | |
| usd = ((tok["input"] - tok["cached"]) * P_IN + tok["cached"] * P_CACHED + tok["output"] * P_OUT) / 1e6 | |
| if not calls: | |
| problems.append("%s: no logged usage in %s" % (case, att)) | |
| out[case] = {"attempt": att, "calls": calls, "tokens": tok, "cost_usd": round(usd, 6) if calls else None} | |
| costs = [v["cost_usd"] for v in out.values() if v["cost_usd"] is not None] | |
| json.dump({"method": __doc__, "prices_per_million": [P_IN, P_CACHED, P_OUT], "cases": out, "problems": problems, | |
| "median_cost_usd": statistics.median(costs) if costs else None, "n": len(costs)}, sys.stdout, indent=1) | |