Files
KittyMacro/kittymacro/macros/piano.py
T
capyb2222 bc09d70516 feat(init): KittyMacro
Roblox macros that play from outside the game: Anime Vanguards, and a
falling-notes piano event.
2026-09-05 12:22:28 -07:00

1193 lines
47 KiB
Python

"""A falling-notes event - the lane lights up, the key goes down."""
from __future__ import annotations
import time
from collections import Counter, deque
from .. import config, screen, winput
from ..runner import Runner
NAME = "piano"
DEFAULTS = {
"hotkeys": {"pause": "f7", "stop": "f8", "capture": "f6"},
# One per lane, left to right: its key, its point, and the colours it learnt.
"lanes": [],
"detect": {
"threshold": 45, # how far from empty counts as a note, per channel
"offset_ms": 0, # wait this long after seeing one before pressing
"min_hold_ms": 35, # a tap still has to be long enough to register
"gap_ms": 45, # ignore a second note in a lane sooner than this
"let_go_ms": 12, # a held key waits this long for the note to come back
"hold_notes": True, # keep the key down while the note is still there
"max_hold_ms": 4000, # let go anyway after this, so a lane cannot jam
},
"finish": {
# When a song counts as over.
"quiet_seconds": 4.0, # a track that stops changing for this long is done
"gone_seconds": 1.5, # no lane on its own track for this long: not the event
"track_tolerance": 24, # and how near its own colour a lane has to be
"min_seconds": 20.0, # never call it done sooner than this into a song
"max_seconds": 420.0, # and never sit through more than this
"movement": 8, # per-channel change that counts as the track moving
"clock": None, # {"at": [x, y]} on the song timer — see `Timer`
"probe": None, # {"at": [x, y], "color": [r, g, b]} on the results panel
"tolerance": 26, # colour slack on that pixel
"check_every": 0.4, # how often to look at it
},
# Picked once each, then it drives itself in and back out again.
"routes": {
"into": [
{"do": "click", "at": None, "after": 1.5, "note": "Events"},
{"do": "click", "at": None, "after": 1.5, "note": "Special Events"},
{"do": "scroll", "at": None, "amount": -12, "after": 1.0,
"note": "scroll the list to the bottom"},
{"do": "click", "at": None, "after": 2.0, "note": "the Piano event"},
{"do": "click", "at": None, "after": 4.0, "note": "Play"},
],
"back": [
{"do": "click", "at": None, "after": 2.5, "note": "leave the results screen"},
],
},
"session": {
# Ten songs at four difficulties: 40 runs, resumed where it left off.
"songs": {
# One [x, y] per row of the list.
"points": [],
"first": None,
"last": None,
"count": 0, # how many songs there are; 0 means "the rows picked"
"visible": 0, # how many fit on screen; 0 means all of them
"scroll_at": None, # where to put the cursor to scroll the list
"top_clicks": 15, # wheel clicks that get it back to the top
"page_clicks": -3, # and how many move it on one screenful
},
# `clicks` is how many times to press this one's point.
"difficulties": [{"name": "easy", "at": None, "clicks": 0},
{"name": "medium", "at": None, "clicks": 1},
{"name": "hard", "at": None, "clicks": 2},
{"name": "expert", "at": None, "clicks": 3}],
"play": None, # [x, y] on the button that starts the song
# Where a finished song leaves you: back on the song "list".
"after_song": "list",
"pick_seconds": 1.2, # after each menu click
"settle_seconds": 4.0, # after pressing play, before watching for notes
"between_seconds": 2.5, # a breath between runs
},
}
# ------------------------------------------------------------------ lanes
def spread(keys: str, left, right) -> list[dict]:
"""Lane points evenly spaced between the two ends of the hit line."""
keys = list(keys)
if not keys:
return []
return [{"key": key, "at": at, "empty": None}
for key, at in zip(keys, between(left, right, len(keys)))]
def between(first, last, count: int) -> list[list[int]]:
"""`count` points evenly spaced from one to the other, both ends included."""
count = max(1, int(count))
if count == 1 or not last:
return [[int(first[0]), int(first[1])]]
return [[round(first[0] + (last[0] - first[0]) * index / (count - 1)),
round(first[1] + (last[1] - first[1]) * index / (count - 1))]
for index in range(count)]
def _median(values):
ordered = sorted(values)
return ordered[len(ordered) // 2]
def _apart(a, b) -> int:
"""How far two colours are, on their furthest channel."""
return max(abs(int(p) - int(q)) for p, q in zip(a, b))
def study(watched: list[list[tuple]], noise_gate: int = 10) -> dict:
"""Work out what each lane looks like empty, from a lane that is in use."""
lanes = []
for colours in watched:
colours = [colour for colour in colours if colour]
if not colours:
lanes.append({"empty": None, "note": None, "covered": 0.0,
"noise": 0, "distance": 0})
continue
# Quantised, so noise cannot split one background into a hundred near colours.
common = Counter((r >> 3, g >> 3, b >> 3) for r, g, b in colours)
bucket = common.most_common(1)[0][0]
empty_like = [c for c in colours if (c[0] >> 3, c[1] >> 3, c[2] >> 3) == bucket]
empty = tuple(_median([c[channel] for c in empty_like]) for channel in range(3))
deviations = [_apart(colour, empty) for colour in colours]
noise = max((d for d in deviations if d <= noise_gate), default=0)
notes = [colour for colour, drift in zip(colours, deviations)
if drift > max(noise_gate, noise * 2)]
note = (tuple(_median([c[channel] for c in notes]) for channel in range(3))
if len(notes) >= max(3, len(colours) // 100) else None)
covered = len(notes) / len(colours)
lanes.append({
"empty": empty,
"note": note,
"covered": covered,
"noise": noise,
"distance": _apart(note, empty) if note else 0,
# More than half of a watch under a note means the majority colour was the note.
"suspect": covered > 0.45,
"repaired": False,
})
_turn_suspects_around(lanes)
seen = [lane for lane in lanes if lane["note"] and not lane["suspect"]]
noise = max((lane["noise"] for lane in lanes), default=0)
nearest = min((lane["distance"] for lane in seen), default=0)
separable = bool(seen) and nearest > noise * 1.5 + 6
return {
"lanes": lanes,
"noise": noise,
"nearest": nearest,
"separable": separable,
# Halfway between the worst noise and the faintest note.
"threshold": max(8, min(200, round((noise + nearest) / 2))) if separable else None,
}
def _turn_suspects_around(lanes: list[dict]) -> None:
"""A lane read backwards, put right by what the other lanes agree on."""
settled = [lane["empty"] for lane in lanes
if lane["empty"] and not lane["suspect"]]
if not settled:
return
for lane in lanes:
if not (lane["suspect"] and lane["note"]):
continue
agree = sum(1 for background in settled if _apart(lane["note"], background) <= 12)
if agree * 2 >= len(settled):
lane["empty"], lane["note"] = lane["note"], lane["empty"]
lane["covered"] = 1.0 - lane["covered"]
lane["distance"] = _apart(lane["note"], lane["empty"])
lane["suspect"] = False
lane["repaired"] = True
def timeline(colours, empty, note, threshold: int, min_run: int = 2):
"""What a lane did, as runs of track / note / something else."""
runs = []
for colour in colours:
if not colour:
what = "unreadable"
elif empty and screen.close_enough(colour, empty, threshold):
what = "track"
elif note and screen.close_enough(colour, note, threshold):
what = "note"
else:
what = "other"
if runs and runs[-1][0] == what:
runs[-1][1] += 1
runs[-1][2].append(colour)
else:
runs.append([what, 1, [colour]])
kept = []
for what, count, seen in runs:
if count < min_run and kept: # a single odd scan is noise, not a run
kept[-1][1] += count
continue
seen = [colour for colour in seen if colour]
middle = (tuple(_median([c[channel] for c in seen]) for channel in range(3))
if seen else None)
kept.append([what, count, middle])
return [(what, count, middle) for what, count, middle in kept]
def find_lanes(columns, movement, want: int = 0, gate: float = 0.35,
min_width: int = 6) -> list[dict]:
"""Where along a row did the picture actually change?"""
if not movement:
return []
top = max(movement)
if top <= 0:
return []
step = (columns[1] - columns[0]) if len(columns) > 1 else 1
floor = max(top * gate, 12)
groups, current = [], None
for x, value in zip(columns, movement):
if value < floor:
continue
# A gap of a couple of samples is the same lane, not two of them.
if current is not None and x - current[-1][0] <= step * 3:
current.append((x, value))
else:
if current:
groups.append(current)
current = [(x, value)]
if current:
groups.append(current)
found = []
for group in groups:
left, right = group[0][0], group[-1][0]
if right - left + step < min_width:
continue
found.append({
"centre": (left + right) // 2,
"left": left,
"right": right,
"score": round(sum(value for _x, value in group) / len(group)),
})
if want and len(found) > want:
found.sort(key=lambda lane: lane["score"], reverse=True)
found = found[:want]
found.sort(key=lambda lane: lane["centre"])
return found
def pick_row(rows, columns, movements, want: int, gate: float = 0.35,
min_width: int = 6):
"""Of every row watched, the lowest one that shows all the lanes at once."""
qualifying = []
for row, movement in zip(rows, movements):
found = find_lanes(columns, movement, want=want, gate=gate, min_width=min_width)
if len(found) != want:
continue
qualifying.append((row, found, min(lane["score"] for lane in found)))
if not qualifying:
return None
strongest = max(strength for _row, _found, strength in qualifying)
solid = [item for item in qualifying if item[2] >= strongest * 0.5]
row, found, _strength = max(solid, key=lambda item: item[0])
return row, found
def profile(columns, movement, width: int = 96) -> str:
"""The same numbers as a picture, because a picture is read at a glance."""
if not columns:
return ""
span = max(columns[-1] - columns[0], 1)
buckets = [0] * width
for x, value in zip(columns, movement):
index = min(width - 1, (x - columns[0]) * width // (span + 1))
buckets[index] = max(buckets[index], value)
top = max(buckets) or 1
blocks = " \u2581\u2582\u2583\u2584\u2585\u2586\u2587\u2588"
return "".join(blocks[min(8, value * 8 // top)] for value in buckets)
def marks(columns, found, width: int = 96) -> str:
"""A row of carets under the profile, one per lane it settled on."""
if not columns:
return ""
span = max(columns[-1] - columns[0], 1)
row = [" "] * width
for lane in found:
index = min(width - 1, (lane["centre"] - columns[0]) * width // (span + 1))
row[index] = "^"
return "".join(row)
class Lane:
"""One lane's key, and whether it is currently down."""
def __init__(self, spec: dict):
self.key = spec.get("key") or ""
self.at = (int(spec["at"][0]), int(spec["at"][1]))
self.empty = tuple(spec.get("empty") or ())
self.note = tuple(spec.get("note") or ())
self.clicks = spec.get("do") == "click"
self.hits = 0
self.stuck = 0
self.lit = False
self.down = False
self.down_at = 0.0
self.armed_at = None
self.free_at = None
# Long ago, not zero, so the first note is never "too soon after the last".
self.freed_at = float("-inf")
@property
def ready(self) -> bool:
return bool(self.empty or self.note)
def on_track(self, colour, tolerance: int) -> bool:
"""Is this the empty track - the thing this lane is *usually* looking at?"""
return bool(colour and self.empty
and screen.close_enough(colour, self.empty, tolerance))
def looks_lit(self, colour, threshold: int) -> bool:
if not colour:
return False
if self.note:
return screen.close_enough(colour, self.note, threshold)
if self.empty:
return not screen.close_enough(colour, self.empty, threshold)
return False
# ------------------------------------------------------------ the key
def press(self, now: float) -> None:
self.armed_at = None
self.down = True
self.down_at = now
self.hits += 1
if self.clicks:
winput.click(*self.at)
else:
winput.key_down(self.key)
def release(self, now: float) -> None:
self.down = False
self.free_at = None
self.freed_at = now
if not self.clicks:
winput.key_up(self.key)
def forget(self) -> None:
"""Whatever we thought we were holding, we are not holding it now."""
self.down = False
self.armed_at = None
self.free_at = None
self.lit = False
def rest(self, now: float) -> None:
"""Nothing out there is moving - let go, and do not press again."""
if self.down:
self.release(now)
self.armed_at = None
self.free_at = None
self.lit = False
def update(self, colour, now: float, tuning: dict) -> None:
self.lit = self.looks_lit(colour, tuning["threshold"])
if self.down:
held = now - self.down_at
if self.lit and tuning["hold_notes"]:
self.free_at = None # still under the note
elif self.free_at is None:
# The note has left the sampler - which is above the line.
self.free_at = now + tuning["offset"] + tuning["let_go"]
if tuning["max_hold"] and held >= tuning["max_hold"]:
# The lane never went back to looking empty.
self.stuck += 1
self.release(now)
elif (self.free_at is not None and now >= self.free_at
and held >= tuning["min_hold"]):
self.release(now)
return
if self.armed_at is not None:
if now >= self.armed_at:
self.press(now)
return
if self.lit and now - self.freed_at >= tuning["gap"]:
self.armed_at = now + tuning["offset"]
if tuning["offset"] <= 0:
self.press(now)
# -------------------------------------------------------------- calibrate
def calibrate(cfg: dict) -> dict:
"""Set the lanes up from a terminal. The window does this better."""
from .. import routine
keys = cfg["hotkeys"]
print("\ncalibrating a falling-notes event")
print("The window (python macro.py gui) shows you what it is reading while")
print("you do this, which is worth a lot here. Carrying on anyway:\n")
order = input(" the key for each lane, left to right, no spaces (e.g. asdfg): ").strip()
if not order:
print(" nothing to do without lanes.")
return cfg
print("\nPut the sampling points ON the hit line — where you would press.")
left = routine.capture_point("the LEFTMOST lane, on the hit line", keys)
right = routine.capture_point("the RIGHTMOST lane, on the hit line", keys)
if not left or not right:
print(" need both ends to space the lanes out. Nothing changed.")
return cfg
cfg["lanes"] = spread(order, left["at"], right["at"])
for lane in cfg["lanes"]:
print(f" {lane['key']} at {lane['at']}")
print("\nNow let the track run empty — no notes crossing the line.")
input(" press Enter when it is clear: ")
with screen.around([lane["at"] for lane in cfg["lanes"]]) as region:
region.read()
for lane in cfg["lanes"]:
colour = region.at(*lane["at"])
lane["empty"] = list(colour or ())
print(f" {lane['key']} empty track {colour}")
return cfg
# -------------------------------------------------------------- the end
def probe_matches(point: dict | None, tolerance: int) -> bool:
"""Is the pixel that only the results screen paints showing it right now?"""
if not point or not point.get("at") or not point.get("color"):
return False
x, y = point["at"]
return screen.close_enough(screen.pixel(x, y), point["color"], tolerance)
class Timer:
"""The song's own clock, watched only for whether it is still counting."""
WIDE, TALL = 110, 34 # enough to cover a clock, small enough to be cheap
def __init__(self, at, every: float = 0.2):
x, y = int(at[0]), int(at[1])
self.region = screen.Region(x - self.WIDE // 2, y - self.TALL // 2,
self.WIDE, self.TALL)
self.every = float(every)
self.was = None
self.looked_at = 0.0
self.ticks = 0
def look(self, now: float) -> bool:
"""Did the clock change since the last look?"""
if now - self.looked_at < self.every:
return False
self.looked_at = now
if not self.region.read():
return False
picture = bytes(self.region.pixels)
changed = self.was is not None and picture != self.was
self.was = picture
self.ticks += bool(changed)
return changed
def close(self) -> None:
self.region.close()
class Ending:
"""Whether the song is over, and which of three signals says so."""
def __init__(self, settings: dict | None = None):
settings = settings or {}
self.quiet = float(settings.get("quiet_seconds", 4.0))
self.gone = float(settings.get("gone_seconds", 1.5))
self.track_tolerance = int(settings.get("track_tolerance", 24))
self.least = float(settings.get("min_seconds", 20.0))
self.most = float(settings.get("max_seconds", 420.0))
self.movement = int(settings.get("movement", 8))
self.probe = settings.get("probe")
self.clock = settings.get("clock") or {}
self.tolerance = int(settings.get("tolerance", 26))
self.check_every = float(settings.get("check_every", 0.4))
self.began = self.moved_at = self.looked_at = 0.0
self.was = None
self.probe_since = None
# When the music actually started, and the recent movements that say so.
self.started = None
self.moves = deque()
# And the same question asked of the clock, if there is one.
self.clock_at = None
self.clock_ticks = 0
# When a lane last looked at its own empty track, when half of them last did.
self.on_track_at = None
self.most_track_at = None
self.can_see_track = False
# movements inside `WINDOW` seconds that mean a song is under way.
BUSY = 12
WINDOW = 3.0
@property
def has_clock(self) -> bool:
return bool(self.clock.get("at"))
@property
def has_probe(self) -> bool:
return bool(self.probe and self.probe.get("at") and self.probe.get("color"))
def begin(self, now: float) -> None:
self.began = self.moved_at = self.looked_at = now
self.was = None
self.probe_since = None
self.started = None
self.moves.clear()
self.clock_at = None
self.clock_ticks = 0
self.on_track_at = None
self.most_track_at = None
def skipped(self, now: float, gap: float) -> None:
"""Time the macro was not watching - paused, or behind another window."""
self.began += gap
if self.started is not None:
self.started += gap
self.moved_at = self.looked_at = now
self.was = None
self.probe_since = None
self.moves.clear()
def tracked(self, on: int, lanes: int, now: float) -> None:
"""How many lanes are looking at their own empty track right now?"""
if on:
self.on_track_at = now
if lanes and on * 2 >= lanes:
self.most_track_at = now
def on_the_event(self, now: float) -> bool:
"""Is the track on screen right now?"""
if not self.can_see_track:
return True
return self.on_track_at is not None and now - self.on_track_at < self.gone
def clocked(self, ticked: bool, now: float) -> None:
"""The song timer, as of this look."""
if not ticked:
return
self.clock_ticks += 1
self.clock_at = now
if self.started is None and self.clock_ticks >= 2 and self.on_the_event(now):
self.started = now
def saw(self, colours, now: float) -> None:
"""The strip as of this scan."""
if self.was is not None and len(self.was) == len(colours):
for before, after in zip(self.was, colours):
if before and after and _apart(before, after) > self.movement:
self.moved_at = now
self.moves.append(now)
break
self.was = list(colours)
while self.moves and now - self.moves[0] > self.WINDOW:
self.moves.popleft()
if self.started is None and len(self.moves) >= self.BUSY \
and self.on_the_event(now):
self.started = self.moves[0]
def still_for(self, now: float) -> float:
return now - self.moved_at
def verdict(self, now: float) -> str | None:
"""Why the song is over - or None, meaning it is not."""
if self.started is None:
return None
if now - self.started >= self.most:
return "timeout"
if now - self.started < self.least:
return None
if self.has_probe and now - self.looked_at >= self.check_every:
self.looked_at = now
if probe_matches(self.probe, self.tolerance):
# Twice, a beat apart: one frame of an effect crossing that pixel is not a results screen.
if self.probe_since is None:
self.probe_since = now
elif now - self.probe_since >= 0.5:
return "results"
else:
self.probe_since = None
if self.on_track_at is not None and now - self.on_track_at >= self.gone:
return "gone"
if self.most_track_at is not None \
and now - self.most_track_at >= self.gone * 3:
return "gone"
if self.has_clock and self.clock_at is not None \
and now - self.clock_at >= self.quiet:
return "clock"
# A quiet stretch of chart stops the track moving without stopping the numbers counting.
patience = self.quiet * (3 if self.has_clock else 1)
if now - self.moved_at >= patience:
return "quiet"
return None
# --------------------------------------------------------------- routes
class Look:
"""The screen, coarsely, for asking whether a click did anything."""
STEP = 8
REACTED = 0.06 # the least a click has to move, however still it is
HOPELESS = 0.25 # and the point past which nothing can be told apart
def __init__(self):
width, height = screen.screen_size()
self.region = screen.Region(0, 0, width, height)
self.stride = width * 4
self.rows = range(0, height, self.STEP)
def take(self):
if not self.region.read():
return None
picture = bytes(self.region.pixels)
return b"".join(picture[row * self.stride:(row + 1) * self.stride:self.STEP * 4]
for row in self.rows)
@staticmethod
def apart(before, after) -> float:
"""How much of it changed, 0 to 1. An unreadable screen counts as all."""
if not before or not after or len(before) != len(after):
return 1.0
return sum(1 for a, b in zip(before, after) if a != b) / len(before)
def close(self) -> None:
self.region.close()
def sure_click(runner: Runner, at, label: str = "", tries: int = 3,
settle: float = 0.15, patience: float = 1.6,
look: "Look | None" = None) -> bool:
"""Click, and click again if the screen did not react."""
x, y = int(at[0]), int(at[1])
mine = look is None
look = look or Look()
try:
winput.move_to(x, y)
runner.wait(settle)
# What the screen does with nobody touching it.
steady = look.take()
runner.wait(0.4) # the same sort of span the click is judged over
steady, ambient = look.take(), look.apart(steady, look.take())
if ambient > look.HOPELESS:
runner.log(f" {label or 'that click'}: this screen moves "
f"{ambient:.0%} on its own, so a click cannot be told apart "
f"— clicking once and hoping")
winput.mouse_down()
runner.wait(0.05)
winput.mouse_up()
runner.wait(settle)
return True
bar = max(look.REACTED, ambient * 2.5)
for attempt in range(1, max(1, tries) + 1):
if attempt > 1:
# Off the button and back on, so the game gets a fresh hover before the press - some UI.
winput.move_to(x + 4, y + 4)
runner.wait(0.05)
winput.move_to(x, y)
runner.wait(0.08)
winput.mouse_down()
runner.wait(0.05)
winput.mouse_up()
waited, moved, after = 0.0, 0.0, steady
while waited < patience:
runner.wait(settle)
waited += settle
after = look.take()
moved = look.apart(steady, after)
if moved >= bar:
if attempt > 1:
runner.log(f" (that one needed {attempt} goes)")
return True
steady = after
runner.log(f" {label or 'that click'}: the screen moved {moved:.0%} after "
f"{tries} goes, against {bar:.0%} needed — carrying on anyway")
return False
finally:
if mine:
look.close()
def wheel(runner: Runner, clicks: int) -> None:
"""Wheel clicks one at a time, because a list scrolled in one jump does not."""
step = 1 if clicks > 0 else -1
for _ in range(abs(int(clicks))):
winput.scroll(step)
runner.wait(0.03)
def walk(runner: Runner, steps: list[dict], label: str = "") -> int:
"""Do a route: the picked steps, in order."""
if label:
runner.log(label)
done = 0
# A beat before the first step.
runner.wait(0.4)
look = Look()
for index, step in enumerate(steps or [], start=1):
kind = step.get("do", "click")
note = step.get("note") or kind
at = step.get("at")
if kind in ("click", "scroll") and not at:
runner.log(f" {index}. {note} — not picked yet, skipped")
continue
runner.gate()
if kind == "click":
if step.get("sure") is False or step.get("button", "left") != "left":
winput.click(int(at[0]), int(at[1]), button=step.get("button", "left"))
else:
sure_click(runner, at, note, look=look)
elif kind == "scroll":
winput.move_to(int(at[0]), int(at[1]))
runner.wait(0.12)
wheel(runner, int(step.get("amount", -5)))
elif kind == "key":
winput.tap(step.get("key", "e"))
elif kind != "wait":
runner.log(f" {index}. {note}: no idea what {kind!r} means, skipped")
continue
runner.log(f" {index}. {note}")
runner.wait(float(step.get("after", 0.8)))
done += 1
look.close()
return done
def picked(steps: list[dict] | None) -> bool:
"""Is there anything in this route worth walking?"""
return any(step.get("at") or step.get("do") in ("key", "wait")
for step in steps or [])
# ---------------------------------------------------------------- chart
def tuning_of(cfg: dict) -> dict:
"""Milliseconds are what a person thinks in; seconds are what a loop uses."""
detect = cfg["detect"]
return {
"threshold": int(detect.get("threshold", 45)),
"offset": float(detect.get("offset_ms", 0)) / 1000.0,
"min_hold": float(detect.get("min_hold_ms", 35)) / 1000.0,
"gap": float(detect.get("gap_ms", 45)) / 1000.0,
"let_go": float(detect.get("let_go_ms", 12)) / 1000.0,
"hold_notes": bool(detect.get("hold_notes", True)),
# Zero means never give up, which is a real answer for a game that genuinely holds a note.
"max_hold": float(detect.get("max_hold_ms", 2500)) / 1000.0,
}
class Chart:
"""The lanes, the strip they live in, and one song at a time."""
REST_AFTER = 1.2 # a picture this still is one to let go of and watch
WATCH_EVERY = 8 # px between the columns watched for "is it moving?"
def __init__(self, cfg: dict, runner: Runner):
self.cfg = cfg
self.runner = runner
self.lanes = [Lane(spec) for spec in cfg.get("lanes") or []]
self.tuning = tuning_of(cfg)
self.ending = Ending(cfg.get("finish"))
self.region = screen.around([lane.at for lane in self.lanes]) if self.lanes else None
# A row of the strip watched only for movement.
if self.region is not None:
self.watch_row = self.region.top + self.region.height // 2
self.watch_at = list(range(self.region.left,
self.region.left + self.region.width,
self.WATCH_EVERY))
clock = (cfg.get("finish") or {}).get("clock") or {}
self.timer = Timer(clock["at"]) if clock.get("at") else None
self.ending.can_see_track = any(lane.empty for lane in self.lanes)
self.songs = 0
self.notes = 0
self.scans = 0
self.last = {"notes": 0, "seconds": 0.0, "why": "none"}
def close(self) -> None:
if self.region is not None:
self.region.close()
self.region = None
if self.timer is not None:
self.timer.close()
self.timer = None
def look(self):
"""One read: a colour per lane, and a row across the whole strip."""
if not self.region.read():
return None, None
return ([self.region.at(*lane.at) for lane in self.lanes],
[self.region.at(x, self.watch_row) for x in self.watch_at])
# ----------------------------------------------------------- playing
def play(self, label: str = "") -> str:
"""Play until the song ends."""
runner = self.runner
for lane in self.lanes:
lane.hits = lane.stuck = 0
lane.forget()
started = time.perf_counter()
self.ending.begin(started)
last_scan = started
told = started
scans = 0
waiting = True
why = "stopped"
runner.stats.update(lit=[False] * len(self.lanes), hits=[0] * len(self.lanes),
song=label or "a song")
try:
while True:
runner.gate()
colours, strip = self.look()
if colours is None:
runner.wait(0.05)
continue
now = time.perf_counter()
if now - last_scan > 0.25:
# We were paused or behind another window, and the runner has already dropped every key.
for lane in self.lanes:
lane.forget()
self.ending.skipped(now, now - last_scan)
last_scan = now
scans += 1
self.scans += 1
near = self.ending.track_tolerance
self.ending.tracked(
sum(1 for lane, colour in zip(self.lanes, colours)
if lane.on_track(colour, near)), len(self.lanes), now)
if self.timer is not None:
self.ending.clocked(self.timer.look(now), now)
self.ending.saw(strip, now)
if self.ending.started is None:
runner.status(f"waiting for {label or 'a song'} to start")
elif waiting:
waiting = False
runner.log(f"{label or 'a song'} is under way")
# Nothing is pressed at a picture that has stopped.
resting = (self.ending.still_for(now) >= self.REST_AFTER
or not self.ending.on_the_event(now))
for lane, colour in zip(self.lanes, colours):
if resting:
lane.rest(now)
else:
lane.update(colour, now, self.tuning)
if now - told >= 0.5:
elapsed = now - started
runner.stats.update(
notes=self.notes + sum(lane.hits for lane in self.lanes),
scans=self.scans, rate=scans / elapsed if elapsed else 0.0,
lit=[lane.lit for lane in self.lanes],
hits=[lane.hits for lane in self.lanes])
told = now
verdict = self.ending.verdict(now)
if verdict:
why = verdict
break
finally:
for lane in self.lanes:
lane.rest(time.perf_counter())
winput.release_all()
played = sum(lane.hits for lane in self.lanes)
elapsed = time.perf_counter() - (self.ending.started or started)
self.notes += played
self.songs += 1
self.last = {"notes": played, "seconds": elapsed, "why": why}
runner.stats.update(notes=self.notes)
runner.log(f"{label or 'song'}: {SAID.get(why, why)} after "
f"{elapsed / 60:.1f} min — {played} notes, "
f"{scans / elapsed if elapsed else 0:.0f} scans a second")
for lane in self.lanes:
if lane.stuck:
runner.log(f" {lane.key}: {lane.stuck} let go on the time limit — "
f"that lane is not seeing its track come back")
return why
def report(self) -> None:
runner = self.runner
runner.log(f"{self.songs} songs, {self.notes} notes, {self.scans} scans")
for lane in self.lanes:
runner.log(f" {lane.key or 'click'}: {lane.hits} in the last song")
SAID = {
"gone": "the track went off screen",
"clock": "the timer stopped counting",
"results": "the results screen came up",
"quiet": "the track went still",
"timeout": "gave up on the clock",
"stopped": "stopped",
}
# -------------------------------------------------------------- session
PROGRESS = "piano_progress"
PROGRESS_DEFAULTS = {"done": {}, "runs": 0, "notes": 0}
def load_progress() -> dict:
return config.load(PROGRESS, PROGRESS_DEFAULTS)
def save_progress(data: dict) -> None:
config.save(PROGRESS, data)
def reset_progress() -> None:
save_progress({"done": {}, "runs": 0, "notes": 0})
def song_rows(songs: dict) -> list:
"""Every row of one screenful of the song list, top to bottom."""
points = [list(point) for point in (songs.get("points") or []) if point]
if points:
return points
if songs.get("first"):
count = max(1, int(songs.get("visible") or 0) or int(songs.get("count") or 1))
return between(songs.get("first"), songs.get("last"), count)
return []
def song_count(songs: dict) -> int:
"""How many songs there are: the rows picked, unless the list scrolls."""
rows = song_rows(songs)
if not int(songs.get("visible") or 0):
return len(rows)
return max(len(rows), int(songs.get("count") or 0))
def song_point(songs: dict, index: int):
"""Where song `index` is, and which screenful of the list it is on."""
rows = song_rows(songs)
if not rows:
return None, 0
visible = int(songs.get("visible") or 0) or len(rows)
visible = max(1, min(visible, len(rows)))
return rows[index % visible], index // visible
class Session:
"""Ten songs at four difficulties, and the memory to pick up where it left off."""
def __init__(self, cfg: dict, runner: Runner, chart: "Chart"):
self.cfg = cfg
self.runner = runner
self.settings = cfg.get("session") or {}
self.routes = cfg.get("routes") or {}
self.chart = chart
self.progress = load_progress()
self.announced = False
# Some events drop you back on the song list when a song ends and some throw you out.
self.exits = (self.settings.get("after_song") or "list") == "out"
self.needs_way_in = True
self.played = 0
# ------------------------------------------------------------- plan
def jobs(self) -> list[tuple]:
"""Every run there is to do, song by song, easiest difficulty first."""
songs = song_count(self.settings.get("songs") or {})
levels = self.settings.get("difficulties") or [{"name": "as it is", "at": None}]
return [(song, index, level)
for song in range(songs) for index, level in enumerate(levels)]
@staticmethod
def key_of(song: int, index: int) -> str:
# By position, not by name: renaming "expert" should not lose a night's progress.
return f"{song}:{index}"
def pending(self) -> list[tuple]:
done = self.progress.get("done") or {}
return [job for job in self.jobs() if self.key_of(job[0], job[1]) not in done]
@staticmethod
def name_of(song: int, index: int, level: dict) -> str:
return f"song {song + 1} {level.get('name') or f'difficulty {index + 1}'}"
# ------------------------------------------------------------ menus
CYCLE_GAP = 0.35 # between clicks on a difficulty button that cycles
def choose(self, song: int, index: int, level: dict) -> bool:
"""Click the song, set the difficulty, press play. False if it cannot."""
songs = self.settings.get("songs") or {}
point, page = song_point(songs, song)
if point is None:
self.runner.log("no song rows picked — Event tab, pick one point per row "
"of the song list")
return False
pause = float(self.settings.get("pick_seconds", 1.2))
if page:
self.scroll_to(page, songs)
self.runner.log(f" song {song + 1} at {point[0]},{point[1]}")
sure_click(self.runner, point, f"song {song + 1}")
self.runner.wait(pause)
name = level.get("name") or f"difficulty {index + 1}"
at = level.get("at")
clicks = 1 if level.get("clicks") is None else int(level["clicks"])
if not at:
self.runner.log(f" {name}: no point picked — playing whatever difficulty "
f"is already chosen")
elif clicks <= 0:
self.runner.log(f" {name}: nothing to click, it is where the panel starts")
else:
for click in range(clicks):
winput.click(*at)
self.runner.wait(self.CYCLE_GAP if click + 1 < clicks else 0)
self.runner.log(f" {name}: {clicks} clicks at {at[0]},{at[1]}"
if clicks > 1 else
f" {name}: one click at {at[0]},{at[1]}")
self.runner.wait(pause)
if self.settings.get("play"):
# Not the difficulty clicks above: those cycle a label through four values.
sure_click(self.runner, self.settings["play"], "play")
self.runner.wait(float(self.settings.get("settle_seconds", 4.0)))
return True
def scroll_to(self, page: int, songs: dict) -> None:
at = songs.get("scroll_at") or songs.get("first")
if not at:
return
winput.move_to(int(at[0]), int(at[1]))
self.runner.wait(0.15)
wheel(self.runner, int(songs.get("top_clicks", 15)))
self.runner.wait(0.4)
wheel(self.runner, page * int(songs.get("page_clicks", -3)))
self.runner.wait(0.4)
self.runner.log(f" wound the list down to screenful {page + 1}")
# ------------------------------------------------------------- runs
def remember(self, song: int, index: int, level: dict, why: str) -> None:
run = self.chart.last
done = self.progress.setdefault("done", {})
done[self.key_of(song, index)] = {
"song": song + 1,
"difficulty": level.get("name") or index + 1,
"notes": run["notes"],
"seconds": round(run["seconds"], 1),
"why": why,
"when": time.strftime("%Y-%m-%d %H:%M"),
}
self.progress["runs"] = len(done)
self.progress["notes"] = sum(int(item.get("notes", 0)) for item in done.values())
save_progress(self.progress)
def announce(self) -> None:
total = len(self.jobs())
done = len(self.progress.get("done") or {})
if done:
self.runner.log(f"picking up where it left off — {done} of {total} runs done")
else:
self.runner.log(f"{total} runs to do")
if self.exits:
self.runner.log("this event drops you out after a song, so the way in "
"gets walked again before every run")
def way_in(self) -> None:
if picked(self.routes.get("into")):
walk(self.runner, self.routes["into"], "walking the way in —")
else:
self.runner.log("no way in picked — carrying on from wherever you are, "
"which wants to be the song list")
def tick(self) -> None:
if not self.announced:
self.announced = True
self.announce()
if self.needs_way_in:
self.needs_way_in = False
self.way_in()
left = self.pending()
total = len(self.jobs())
self.runner.stats.update(runs=total - len(left), total=total)
if not left:
self.report()
self.runner.stopped = True
return
song, index, level = left[0]
label = self.name_of(song, index, level)
self.runner.log(f"{label} — run {total - len(left) + 1} of {total}")
if not self.choose(song, index, level):
self.runner.stopped = True
return
why = self.chart.play(label=label)
self.remember(song, index, level, why)
self.played += 1
self.runner.stats.update(runs=len(self.progress["done"]), total=total)
if picked(self.routes.get("back")):
walk(self.runner, self.routes["back"], " back to the list —")
if self.exits:
self.needs_way_in = True
self.runner.wait(float(self.settings.get("between_seconds", 2.5)))
def report(self) -> None:
done = self.progress.get("done") or {}
notes = sum(int(run.get("notes", 0)) for run in done.values())
minutes = sum(float(run.get("seconds", 0)) for run in done.values()) / 60
self.runner.log(f"all {len(done)} runs done — {notes} notes over "
f"{minutes:.0f} minutes of song, {self.played} of them today")
odd = sorted(f"{run['song']} {run['difficulty']}"
for run in done.values() if run.get("why") == "timeout")
if odd:
self.runner.log(f" ended on the clock rather than on an ending: "
f"{', '.join(odd)} — worth replaying by hand")
# -------------------------------------------------------------------- run
def migrate(cfg: dict) -> dict:
"""An older config kept the end of a song at the top level."""
cfg.pop("quiet_seconds", None)
return cfg
def _lanes_ready(cfg: dict, runner: Runner) -> bool:
lanes = cfg.get("lanes") or []
if not lanes:
runner.log("no lanes set up — calibrate first (the window makes this easy)")
return False
unknown = sorted({lane.get("key") for lane in lanes
if lane.get("do") != "click" and lane.get("key") not in winput.KEYS})
if unknown:
runner.log(f"lanes named {', '.join(repr(key) for key in unknown)} are not keys "
f"this can press — fix them before starting")
return False
unset = [lane.get("key") or "?" for lane in lanes
if not (lane.get("empty") or lane.get("note"))]
if unset:
runner.log(f"lanes {''.join(unset)!r} have no empty-track colour — "
f"they will never fire until you sample one")
return True
def run(once: bool = False, runner: Runner | None = None, mode: str | None = None) -> None:
"""Play: mode is "loop", "once" or "session". `once` is the older spelling."""
mode = mode or ("once" if once else "loop")
cfg = migrate(config.load(NAME, DEFAULTS))
config.save(NAME, cfg)
runner = runner or Runner("Piano", cfg["hotkeys"])
if not _lanes_ready(cfg, runner):
return
chart = Chart(cfg, runner)
ending = chart.ending
runner.log(f"{len(chart.lanes)} lanes, reading "
f"{chart.region.width}x{chart.region.height} px at "
f"{[lane.at for lane in chart.lanes]}")
knows = ["the track goes off screen"]
if chart.timer is not None:
knows.insert(0, "the timer stops")
if ending.has_probe:
knows.insert(0, "the results pixel shows")
runner.log(f"threshold {chart.tuning['threshold']}, offset "
f"{cfg['detect']['offset_ms']}ms — a song ends when "
f"{' or '.join(knows)} for {ending.quiet:.0f}s")
runner.stats.update(notes=0, scans=0, rate=0.0, started=time.monotonic(),
lit=[False] * len(chart.lanes), hits=[0] * len(chart.lanes))
if mode == "session":
tick = Session(cfg, runner, chart).tick
elif mode == "once":
def tick():
chart.play(label="the song")
runner.stopped = True
else:
def tick():
# play() waits for the next song by itself.
chart.play()
try:
runner.loop(tick)
finally:
chart.close()
chart.report()