perf(studio): parse a built PAC's clips once: the findings re-check asked every few seconds

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Author: sp00ktober

apps/studio/src/mhfu_studio/monster/clips.py

@@ -13,10 +13,12 @@
 
 from __future__ import annotations
 
+import functools
 import hashlib
 import re
 from collections.abc import Callable, Iterable, Mapping
 from dataclasses import dataclass, field
+from types import MappingProxyType
 
 from mhfu_port import layout as layouts
 from mhfu_port import motion, slots
@@ -77,8 +79,11 @@ def clip_table(anim: AnimPack) -> dict[int, Fingerprint]:
     return {s: (frames, bool(loop)) for s, (frames, loop) in slots.slot_prints(anim).items()}
 
 
-def pac_clip_table(pac: bytes) -> dict[int, Fingerprint]:
-    return clip_table(slots.anim_of(pac))
+@functools.lru_cache(maxsize=2)
+def pac_clip_table(pac: bytes) -> Mapping[int, Fingerprint]:
+    """`clip_table` of a built PAC, parsed once per PAC: the findings ask every few seconds.
+    Read-only, as the table is shared by every caller."""
+    return MappingProxyType(clip_table(slots.anim_of(pac)))
 
 
 def source_clip_table(moveset: bytes) -> dict[int, Fingerprint]:

apps/studio/src/mhfu_studio/monster/validate.py

@@ -208,7 +208,7 @@ def _pac(m: Manifest, pac: bytes | None, sources: Mapping[int, int]) -> list[Fin
         return []
     out = []
     try:
-        table: dict[int, clips.Fingerprint] | None = clips.pac_clip_table(pac)
+        table: Mapping[int, clips.Fingerprint] | None = clips.pac_clip_table(pac)
     except ValueError as e:
         out.append(_f("error", "PAC_UNREADABLE", "", f"no clip table in the PAC: {e}"))
         table = None
@@ -225,7 +225,7 @@ def _pac(m: Manifest, pac: bytes | None, sources: Mapping[int, int]) -> list[Fin
 
 
 def _clips(
-    m: Manifest, table: dict[int, clips.Fingerprint], sources: Mapping[int, int]
+    m: Manifest, table: Mapping[int, clips.Fingerprint], sources: Mapping[int, int]
 ) -> list[Finding]:
     out = []
     moved = {

apps/studio/src/mhfu_studio/monster/workspace.py

@@ -12,7 +12,7 @@
 
 import dataclasses
 import tomllib
-from collections.abc import Callable, Hashable, Sequence
+from collections.abc import Callable, Hashable, Mapping, Sequence
 from concurrent.futures import Future, ThreadPoolExecutor
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
@@ -924,7 +924,7 @@ def attack_session(self) -> AttackSession | None:
 
     # clips
 
-    def clip_table(self) -> dict[int, clips.Fingerprint]:
+    def clip_table(self) -> Mapping[int, clips.Fingerprint]:
         """The open build's fingerprints."""
         sc = self.scene
         if sc is None:

apps/studio/src/mhfu_studio/shell/studio.py

@@ -33,6 +33,12 @@
     from mhfu_studio.shell.viewport import Viewport
 
 
+SLOW_COST, SLOW_EVERY = 0.05, 60.0
+"""A timed re-check that cost more than this many seconds is not repeated for that many: a stall
+is the check's whole length, which the interval's share of the time does not bound. A change
+(`Findings.stale`) re-checks at once."""
+
+
 class Findings:
     """The active document's findings, re-checked on demand and, when `auto`, now and then.
 
@@ -55,7 +61,8 @@ def get(self, doc: Any, now: float | None = None) -> list[Finding]:
         if doc is None:
             self.found, self._doc = [], None
             return self.found
-        due = self.auto and now - self._at >= max(1.0, 10 * self._cost)
+        wait = SLOW_EVERY if self._cost > SLOW_COST else max(1.0, 10 * self._cost)
+        due = self.auto and now - self._at >= wait
         if self._stale or doc is not self._doc or due:
             t0 = time.monotonic()
             try:

apps/studio/tests/monster/test_clips.py

@@ -216,3 +216,52 @@ def test_zinogre(games, built, ports):
 def test_brute(built):
     table = C.pac_clip_table(built("brute_tigrex"))
     assert len(table) == 77 and table[61] == (382, False) and 24 not in table
+
+
+@pytest.fixture
+def parses(monkeypatch):
+    """The PACs `pac_clip_table` parsed, with a table that says which."""
+    seen: list[bytes] = []
+    monkeypatch.setattr(slots, "anim_of", lambda pac: seen.append(pac) or pac)
+    monkeypatch.setattr(C, "clip_table", lambda pack: {1: (len(pack), True)})
+    C.pac_clip_table.cache_clear()
+    yield seen
+    C.pac_clip_table.cache_clear()
+
+
+def test_a_pac_is_parsed_once_whoever_asks(parses):
+    pac = b"a pac"
+    first = C.pac_clip_table(pac)
+    assert C.pac_clip_table(pac) is first and C.pac_clip_table(bytes(bytearray(pac))) is first
+    assert parses == [pac]
+
+
+def test_other_bytes_are_another_table(parses):
+    one, two = C.pac_clip_table(b"one"), C.pac_clip_table(b"three")
+    assert one is not two and dict(one) == {1: (3, True)} and dict(two) == {1: (5, True)}
+    assert parses == [b"one", b"three"]
+
+
+def test_the_table_is_shared_so_no_caller_can_change_it(parses):
+    table = C.pac_clip_table(b"pac")
+    with pytest.raises(TypeError):
+        table[2] = (1, False)
+    with pytest.raises(TypeError):
+        del table[1]
+    assert dict(C.pac_clip_table(b"pac")) == {1: (3, True)}
+
+
+def test_an_unreadable_pac_is_not_remembered(monkeypatch):
+    tries = []
+
+    def broken(pac):
+        tries.append(pac)
+        raise ValueError("no clips")
+
+    monkeypatch.setattr(slots, "anim_of", broken)
+    C.pac_clip_table.cache_clear()
+    for _ in range(2):
+        with pytest.raises(ValueError, match="no clips"):
+            C.pac_clip_table(b"bad")
+    assert len(tries) == 2
+    C.pac_clip_table.cache_clear()