feat(log): time every phase and fold stealth into one spawn-to-loot timeline

Extends the town timeline to the whole cycle and times every step:

    grep "TL>" log/log.txt

    TL> g2 r1 | game  | spawn             | ok    | at a5_town_start
    TL> g2 r1 | town  | inspect_inventory | ok    | took=3.0s   | in pack=2 keep=0 sell=2
    TL> g2 r1 | town  | repair            | ok    | took=19.3s  | at a5_larzuk
    TL> g2 r1 | town  | item_sell         | ok    | SOUL IMPALER @ (928, 465)
    TL> g2 r1 | town  | resurrect_merc    | fail  | took=113.6s | NPC not reachable
    TL> g2 r1 | town  | maintenance       | ok    | took=137.6s | at a5_larzuk
    TL> g2 r1 | run   | run_pindle        | start | from a5_larzuk
    TL> g2 r1 | run   | approach          | fail  | took=71.9s  | step=click_red_portal
    TL> g2 r2 | game  | end               | fail  | Approach failed [step: click_red_portal]

- phases: game (start / spawn / end), town (all maintenance steps), run (approach /
  battle / loot), stlth.
- every terminating line carries took=Ns; "start" stamps the clock in Bot._tl_starts
  keyed by (phase, step). The sample above pays for itself immediately: a failed game
  spent 113.6s of its 137.6s town visit on a merc resurrect that failed.
- stealth decisions are tracked: afk_break (timed across the sleep), skip_run and
  wrong_waypoint.

Bot.timeline() is a static entry point so utils/stealth.py and inventory/personal.py can
emit without importing Bot at module level (that would be circular). Both use a lazy
guarded import and it no-ops when no Bot is live. Item sells/stashes/drops now route
through it too, so they share the game/run counters and column widths instead of being a
separately formatted line.

Prefix moved TOWN> -> TL> now that it spans more than town.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
alexpolo1
2026-08-27 18:07:33 +02:00
co-authored by Claude Opus 5
parent f01610e53f
commit 8da4c8eb9d
4 changed files with 119 additions and 34 deletions
+37 -25
View File
@@ -659,39 +659,51 @@ paths; don't unify them.
---
## The town timeline — `grep "TOWN>"`
## The run timeline — `grep "TL>"`
Every town-maintenance step reports start and outcome in one stable, machine-readable
shape, so a whole town visit reads as a sequence:
One fixed-width, machine-readable line per step, covering the whole cycle from spawn to
loot. Every step is **timed**.
```bash
grep "TOWN>" log/log.txt
grep "TL>" log/log.txt
```
```
TOWN> g2 r2 | maintenance | start | at a5_town_start
TOWN> g2 r2 | town_heal | start
TOWN> g2 r2 | inspect_inventory | ok | in pack=0 keep=0 sell=0 gold_full=False
TOWN> g2 r2 | buy_consumables | start | needs id=0 tp=0 hp=4 mana=0 rejuv=0 | sell_pending=0
TOWN> g2 r2 | buy_consumables | ok | at a4_jamella | after: Consumables(...)
TOWN> g2 r2 | stash_items | skip | nothing kept and gold not full
TOWN> g2 r2 | repair | skip | no repair due and nothing to sell
TOWN> g2 r2 | resurrect_merc | start
TOWN> g2 r2 | gamble | skip | stash not full / gambling not configured
TOWN> g2 r2 | maintenance | ok | done in 21s | at a4_jamella
TL> g2 r1 | game | start | start | char=fohdin difficulty=nightmare routes=['run_pindle']
TL> g2 r1 | game | spawn | ok | at a5_town_start (act a5_town_start)
TL> g2 r1 | town | maintenance | start | at a5_town_start
TL> g2 r1 | town | inspect_inventory | ok | took=3.0s | in pack=2 keep=0 sell=2 gold_full=False
TL> g2 r1 | town | stash_items | skip | nothing kept and gold not full
TL> g2 r1 | town | repair | ok | took=19.3s | at a5_larzuk
TL> g2 r1 | town | item_sell | ok | SOUL IMPALER @ (928, 465)
TL> g2 r1 | town | resurrect_merc | fail | took=113.6s | NPC not reachable — continuing mercless
TL> g2 r1 | town | maintenance | ok | took=137.6s | at a5_larzuk
TL> g2 r1 | run | run_pindle | start | from a5_larzuk
TL> g2 r1 | run | approach | fail | took=71.9s | step=click_red_portal
TL> g2 r2 | game | end | fail | Approach failed for run_pindle [step: click_red_portal]
```
- `status` is one of **start | ok | skip | fail**. `skip` says *why* it was skipped, so a
step that silently did nothing is now distinguishable from one that never ran.
- Individual item transfers mirror into the same stream from
`inventory/personal.py::transfer_items` as `item_sell` / `item_stash` / `item_drop`, so
vendoring a rare and stashing a rune appear inline with the steps around them.
- The `maintenance ok` line carries total duration — the fastest way to spot a slow town
visit. Example from the first run after this landed: `done in 225s` with
`buy_consumables | fail` versus `done in 21s` with `ok` the next game.
**Columns:** `game rN | phase | step | status | took= | detail`
Emitted by `_step()` inside `Bot.on_maintenance` (`src/bot.py`). **Keep the format stable** —
it is meant to be grepped, not read as prose. The `>` in the prefix matters: a bare `TOWN`
also matches template names like `A5_TOWN_0`.
- **phase**: `game` | `town` | `run` | `stlth`
- **status**: `start` | `ok` | `skip` | `fail`. `skip` states *why*, so a step that did
nothing is distinguishable from one that never ran.
- **took=**: emitted on every terminating line. `start` stamps the clock in
`Bot._tl_starts`, keyed by `(phase, step)`. This is what makes the slow phase findable —
the example above shows a failed game spending **113.6s** of its 137.6s town visit on a
merc resurrect that failed.
**Emitters:**
- `Bot.tl()` — the instance method (`src/bot.py`).
- `Bot.timeline()` — a **static** entry point for modules that cannot import `Bot` without
a circular import. `utils/stealth.py` and `inventory/personal.py` both use it via a lazy
guarded import, so item sells/stashes/drops and stealth decisions land in the same
stream. It is a no-op when no `Bot` is live.
**Stealth** appears as phase `stlth`: `afk_break` (start/ok around the sleep, so the break
duration is timed), `skip_run`, and `wrong_waypoint`.
**Keep the format stable** — it is meant to be grepped, not read as prose. The `>` in the
prefix matters: a bare `TL`/`TOWN` also matches template names like `A5_TOWN_0`.
---
+56 -7
View File
@@ -60,6 +60,16 @@ class Bot:
_MERC_RESURRECT_FAIL_LIMIT = 2
_MERC_RESURRECT_SKIP_GAMES = 15
# Set to the live Bot so utils.stealth can log into the same timeline without
# importing Bot (which would be a circular import).
_tl_active = None
@staticmethod
def timeline(phase: str, step: str, status: str = "start", detail: str = "") -> None:
"""Module-safe entry point for the timeline; a no-op if no Bot is running."""
if Bot._tl_active is not None:
Bot._tl_active.tl(phase, step, status, detail)
def __init__(self, game_stats: GameStats):
self._game_stats = game_stats
self._messenger = Messenger()
@@ -154,6 +164,8 @@ class Bot:
self._countess = Countess(self._pather, self._town_manager, self._char, self._pickit, self._do_runs)
# Create member variables
self._picked_up_items = False
self._tl_starts: dict = {}
Bot._tl_active = self
self._curr_loc: bool | Location = None
# Act the character spawned in this game — detected reliably at spawn (town
# markers are visible there). Used as the act-of-record when mid-town marker
@@ -314,6 +326,33 @@ class Bot:
result[asset] = key
return result
def tl(self, phase: str, step: str, status: str = "start", detail: str = "") -> None:
"""One line of the whole-cycle timeline: spawn -> town -> run -> loot.
grep "TL>" log/log.txt
phase : game | town | run | stlth
status : start | ok | skip | fail
Every step is timed: "start" stamps the clock, and the terminating line reports
took=Ns. That is what turns the timeline from a trace into something you can find
the slow phase in. Format is deliberately fixed-width and machine-readable — do
not prettify it.
"""
key = (phase, step)
if status == "start":
self._tl_starts[key] = time.time()
took = ""
else:
t0 = self._tl_starts.pop(key, None)
took = f"took={time.time() - t0:.1f}s" if t0 is not None else ""
gs = self._game_stats
head = (f"TL> g{getattr(gs, '_game_counter', 0)} r{getattr(gs, '_run_counter', 0)} "
f"| {phase:<5} | {step:<17} | {status:<5}")
bits = [b for b in (took, detail) if b]
line = f"{head} | {' | '.join(bits)}" if bits else head
(Logger.warning if status == "fail" else Logger.info)(line)
Bot._tl_active = self
def on_init(self):
self._game_stats.log_start_game()
self._town_manager.reset_wp_budget()
@@ -322,6 +361,7 @@ class Bot:
difficulty = Config().general.get("difficulty", "unknown")
char_type = Config().char.get("type", "unknown")
Logger.info(f"=== BOT START === char={char_type} | difficulty={difficulty} | routes={active_routes}")
self.tl("game", "start", "start", f"char={char_type} difficulty={difficulty} routes={active_routes}")
# Force D2R client area to stable position to prevent offset drift
from utils.misc import enforce_d2r_window
enforce_d2r_window(5, 98)
@@ -400,7 +440,8 @@ class Bot:
Logger.warning("Could not detect town spawn — defaulting to A5_TOWN_START (route home act)")
self._curr_loc = Location.A5_TOWN_START
self._spawn_act = TownManager.get_act_from_location(self._curr_loc)
self.tl("game", "spawn", "ok", f"at {self._curr_loc} (act {self._spawn_act})")
# Handle picking up corpse in case of death
if (corpse_present := is_visible(ScreenObjects.Corpse)):
self._previous_run_failed = True
@@ -468,16 +509,13 @@ class Bot:
Every step reports start and outcome in the same shape so a whole town visit
can be read (or grepped) as a sequence:
grep "TOWN>" log/log.txt
grep "TL>" log/log.txt
status is one of: start | ok | skip | fail. Keep the format stable — it is
meant to be machine-readable, not prose.
"""
if status == "start":
self._maintenance_step = step
gs = self._game_stats
head = f"TOWN> g{getattr(gs, '_game_counter', 0)} r{getattr(gs, '_run_counter', 0)} | {step:<16} | {status:<5}"
line = f"{head} | {detail}" if detail else head
(Logger.warning if status == "fail" else Logger.info)(line)
self.tl("town", step, status, detail)
self._step_fn = _step
_step("maintenance", "start", f"at {self._curr_loc}")
@@ -873,7 +911,7 @@ class Bot:
self._curr_loc = self._verify_town_location()
break
_step("maintenance", "ok", f"done in {time.time() - _maint_start:.0f}s | at {self._curr_loc}")
_step("maintenance", "ok", f"at {self._curr_loc}")
# Start a new run
started_run = False
@@ -894,6 +932,8 @@ class Bot:
self._game_stats.set_failure_reason("Game ended without completing runs")
if Config().general["info_screenshots"] and failed:
safe_imwrite("./log/screenshots/info/info_failed_game_" + time.strftime("%Y%m%d_%H%M%S") + ".png", grab())
self.tl("game", "end", "fail" if failed else "ok",
(self._game_stats.get_failure_reason() or "") if failed else "")
self._curr_loc = False
self._pre_buffed = False
if view.save_and_exit() == False:
@@ -1090,12 +1130,15 @@ class Bot:
res = False
self._do_runs[run_name] = False
self._game_stats.log_run_started(run_name)
self.tl("run", run_name, "start", f"from {self._curr_loc}")
set_pause_state(False)
self.tl("run", "approach", "start")
self._curr_loc = run_obj.approach(self._curr_loc, *approach_args)
if not self._curr_loc:
# Approach failed — couldn't reach the boss
step = getattr(run_obj, "approach_fail_step", None)
reason = f"Approach failed for {run_name}" + (f" [step: {step}]" if step else "")
self.tl("run", "approach", "fail", f"step={step or '?'}")
Logger.error(reason)
self._game_stats.set_failure_reason(reason)
self._save_error_screenshot(run_name, reason)
@@ -1103,10 +1146,13 @@ class Bot:
loot = self._pickit.consume_run_loot() if hasattr(self, "_pickit") and self._pickit else []
if loot:
Logger.info(f"Loot from {run_name} (approach failed): {', '.join(loot)}")
self.tl("run", "loot", "ok", f"(approach failed) {', '.join(loot)}")
self._game_stats.log_run_finished(run_name, True, picked, loot=loot)
self._record_run_result(run_name, True)
self._ending_run_helper(res)
return
self.tl("run", "approach", "ok", f"at {self._curr_loc}")
self.tl("run", "battle", "start")
try:
res = run_obj.battle(*battle_args)
except Exception as e:
@@ -1127,6 +1173,7 @@ class Bot:
battle_succeeded = bool(res) if not isinstance(res, np.ndarray) else bool(res.any())
else:
battle_succeeded = bool(res) if res is not None else False
self.tl("run", "battle", "ok" if battle_succeeded else "fail")
if not battle_succeeded:
# Battle failed — boss fight didn't complete
if not self._game_stats.get_failure_reason():
@@ -1140,8 +1187,10 @@ class Bot:
counted[name] = counted.get(name, 0) + 1
summary = ", ".join(f"{n}x {name}" if n > 1 else name for name, n in counted.items())
Logger.info(f"Loot from {run_name}: {summary}")
self.tl("run", "loot", "ok", summary)
else:
Logger.info(f"Loot from {run_name}: nothing picked up")
self.tl("run", "loot", "ok", "nothing picked up")
self._game_stats.log_run_finished(run_name, not battle_succeeded, picked, loot=loot)
self._record_run_result(run_name, not battle_succeeded)
self._ending_run_helper(res)
+5 -1
View File
@@ -543,7 +543,11 @@ def transfer_items(items: list, action: str = "drop", img: np.ndarray = None) ->
Logger.debug(f"Confirmed {action}{item_label} at position {item.pos}")
# Mirror into the TOWN timeline so sells/stashes/drops of individual items
# appear in the same greppable sequence as the maintenance steps.
Logger.info(f"TOWN> | item_{action:<11} | ok | {item.name or '?'} @ {item.pos}")
try:
from bot import Bot
Bot.timeline("town", f"item_{action}", "ok", f"{item.name or '?'} @ {item.pos}")
except Exception:
pass
for cnt, o_item in enumerate(items):
if o_item.pos == item.pos:
items.pop(cnt)
+21 -1
View File
@@ -6,6 +6,20 @@ from logger import Logger
from utils.misc import wait
def _tl(step: str, status: str = "ok", detail: str = "") -> None:
"""Mirror stealth decisions into the run timeline (grep "TL>").
Imported lazily and guarded: utils.stealth is imported BY bot.py, so a module-level
import of Bot here would be circular, and stealth helpers are also called from
contexts with no live Bot.
"""
try:
from bot import Bot
Bot.timeline("stlth", step, status, detail)
except Exception:
pass
def maybe_afk_break():
"""Call after each run. Randomly takes an unscheduled AFK break."""
try:
@@ -15,7 +29,9 @@ def maybe_afk_break():
if random.randint(1, 100) <= cfg["afk_break_chance"]:
minutes = random.uniform(cfg["afk_break_min_m"], cfg["afk_break_max_m"])
Logger.info(f"[Stealth] Taking unscheduled AFK break for {minutes:.1f} minutes")
_tl("afk_break", "start", f"planned {minutes:.1f}m (chance {cfg['afk_break_chance']}%)")
wait(minutes * 60, minutes * 60 * 1.5)
_tl("afk_break", "ok", "resuming")
Logger.info("[Stealth] AFK break over, resuming")
@@ -27,6 +43,7 @@ def should_skip_run() -> bool:
return False
if cfg["skip_run_chance"] > 0 and random.randint(1, 100) <= cfg["skip_run_chance"]:
Logger.info("[Stealth] Randomly skipping this run for anti-detection")
_tl("skip_run", "ok", f"rolled skip (chance {cfg['skip_run_chance']}%)")
return True
return False
@@ -184,7 +201,10 @@ def should_wrong_waypoint() -> bool:
chance = cfg.get("wrong_waypoint_chance", 0.025)
except Exception:
chance = 0.025
return random.random() < chance
hit = random.random() < chance
if hit:
_tl("wrong_waypoint", "ok", f"deliberate misclick (chance {chance})")
return hit
def skill_rotation_hesitation() -> float: