from __future__ import annotations import math import time from dataclasses import dataclass from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from typing import Any, Dict, List, Optional, Tuple NUM_COLUMNS = 8 MIN_DISPLAY_POWER = 2 CurvePoint = Tuple[int, float] def format_dollars(pennies: int) -> str: """Format integer pennies as `$1,234.56`.""" value = max(0, int(pennies)) dollars = value // 100 cents = value % 100 return f"${dollars:,}.{cents:02d}" def parse_display_to_pennies(value: Any, default: int = 0) -> int: """Parse an int-pennies or dollar-string value into pennies.""" if value is None: return max(0, int(default)) if isinstance(value, int): return max(0, int(value)) if isinstance(value, float): cents = Decimal(str(value)) * Decimal("100") return max(0, int(cents.quantize(Decimal("1"), rounding=ROUND_HALF_UP))) text = str(value).strip() if not text: return max(0, int(default)) text = text.replace("$", "").replace(",", "") try: parsed = Decimal(text) except (InvalidOperation, ValueError): return max(0, int(default)) if "." in text: cents = parsed * Decimal("100") return max(0, int(cents.quantize(Decimal("1"), rounding=ROUND_HALF_UP))) return max(0, int(parsed)) @dataclass class OdometerConfig: max_lag_pennies: int = 1000 penny_max_lag_pennies: int = 500 dime_continuous_lag_pennies: int = 0 dime_roll_max_lag_pennies: int = 900 penny_roll_max_lag_pennies: int = 0 min_rate: float = 0.15 max_rate: float = 0.5 dime_max_rate: float = 5.0 idle_catchup_sec: float = 25.0 contribution_idle_sec: float = 3.0 penny_roll_sec: float = 2.5 penny_roll_sec_min: float = 0.0 carry_roll_sec: float = 2.5 carry_roll_sec_min: float = 0.0 initial_display_pennies: int = 0 rate_curve: Tuple[CurvePoint, ...] = () roll_curve_sec: Tuple[CurvePoint, ...] = () chase_rate_scale: float = 1.0 catchup_lag_threshold: int = 20 catchup_lag_bonus_per_cent: float = 0.0 # When display lag exceeds threshold (pennies), snap to this lag behind live target. snap_behind_threshold_pennies: int = 1000 snap_behind_to_lag_pennies: int = 200 # Exponential smoothing time constant (seconds) for roll-rate transitions. # Prevents per-penny speed jerks; ~90% convergence in ~2.3*tau seconds. roll_smooth_tau: float = 0.4 # Never scroll pennies faster than this (seconds per 1¢ digit roll). penny_roll_sec_floor: float = 0.20 # Lag-gated dime-driven catch-up (hysteresis). 0 enter = disabled. dime_catchup_enter_pennies: int = 100 dime_catchup_exit_pennies: int = 80 @classmethod def from_mapping( cls, data: Optional[Dict[str, Any]] = None, **kwargs: Any ) -> "OdometerConfig": merged: Dict[str, Any] = {} if data: merged.update(data) merged.update(kwargs) cfg = cls() for key in ("initial_display", "initial_dollars", "initial_pennies"): if key in merged: cfg.initial_display_pennies = parse_display_to_pennies( merged[key], default=cfg.initial_display_pennies ) break key_map = { "max_lag_pennies": "max_lag_pennies", "penny_max_lag_pennies": "penny_max_lag_pennies", "penny_max_lag": "penny_max_lag_pennies", "penny_roll_max_lag_pennies": "penny_roll_max_lag_pennies", "penny_roll_max_lag": "penny_roll_max_lag_pennies", "dime_continuous_lag_pennies": "dime_continuous_lag_pennies", "dime_continuous_lag": "dime_continuous_lag_pennies", "dime_roll_max_lag_pennies": "dime_roll_max_lag_pennies", "dime_roll_max_lag": "dime_roll_max_lag_pennies", "min_rate": "min_rate", "max_rate": "max_rate", "dime_max_rate": "dime_max_rate", "idle_catchup_sec": "idle_catchup_sec", "contribution_idle_sec": "contribution_idle_sec", "max_lag": "max_lag_pennies", "pennies_min_rate": "min_rate", "pennies_max_rate": "max_rate", "penny_roll_sec": "penny_roll_sec", "penny_roll_sec_min": "penny_roll_sec_min", "carry_roll_sec": "carry_roll_sec", "carry_roll_sec_min": "carry_roll_sec_min", "dime_roll_sec": "carry_roll_sec", "dime_roll_sec_min": "carry_roll_sec_min", "chase_rate_scale": "chase_rate_scale", "catchup_lag_threshold": "catchup_lag_threshold", "catchup_lag_bonus_per_cent": "catchup_lag_bonus_per_cent", "snap_behind_threshold_pennies": "snap_behind_threshold_pennies", "snap_behind_to_lag_pennies": "snap_behind_to_lag_pennies", "progressive_snap_behind_threshold": "snap_behind_threshold_pennies", "progressive_snap_behind_lag": "snap_behind_to_lag_pennies", "roll_smooth_tau": "roll_smooth_tau", "penny_roll_sec_floor": "penny_roll_sec_floor", "dime_catchup_enter_pennies": "dime_catchup_enter_pennies", "dime_catchup_enter": "dime_catchup_enter_pennies", "dime_catchup_exit_pennies": "dime_catchup_exit_pennies", "dime_catchup_exit": "dime_catchup_exit_pennies", } for src, dst in key_map.items(): if src in merged: setattr(cfg, dst, merged[src]) curve_map = ( ("rate_curve", "rate_curve", "rate"), ("chase_rate_curve", "rate_curve", "rate"), ("pennies_per_second_curve", "rate_curve", "rate"), ("roll_curve_sec", "roll_curve_sec", "sec"), ("penny_roll_curve_sec", "roll_curve_sec", "sec"), ("roll_sec_curve", "roll_curve_sec", "sec"), ) for src, dst, value_key in curve_map: if src in merged: setattr(cfg, dst, _parse_curve_points(merged[src], value_key)) return cfg def _penny_roll_ramp_lag(cfg: OdometerConfig) -> int: if cfg.roll_curve_sec: return max(point[0] for point in cfg.roll_curve_sec) if cfg.penny_roll_max_lag_pennies > 0: return cfg.penny_roll_max_lag_pennies return cfg.penny_max_lag_pennies def slowest_roll_sec(cfg: OdometerConfig) -> float: if cfg.roll_curve_sec: return max(point[1] for point in cfg.roll_curve_sec) fast = max(cfg.penny_roll_sec, 1e-6) return cfg.penny_roll_sec_min if cfg.penny_roll_sec_min > 0 else fast * 5.0 def carry_roll_sec_for_power( power: int, cfg: OdometerConfig, lag_pennies: int = -1 ) -> float: del power lag = lag_pennies if lag_pennies >= 0 else _penny_roll_ramp_lag(cfg) return _penny_roll_sec_at_lag(lag, cfg) def _penny_roll_sec_at_lag(delta: int, cfg: OdometerConfig) -> float: if cfg.roll_curve_sec: sec = _curve_value_at_lag(delta, cfg.roll_curve_sec) else: fast = max(cfg.penny_roll_sec, 1e-6) slow = cfg.penny_roll_sec_min if cfg.penny_roll_sec_min > 0 else fast * 5.0 cap = max(_penny_roll_ramp_lag(cfg), 1) t = min(max(0, delta), cap) / cap sec = slow + t * (fast - slow) floor = float(cfg.penny_roll_sec_floor or 0.0) if floor > 0.0: sec = max(sec, floor) return sec def _pennies_to_digits(pennies: int) -> List[int]: pennies = max(0, int(pennies)) return [(pennies // (10**i)) % 10 for i in range(NUM_COLUMNS)] def leftmost_significant_power(digits: List[int]) -> int: for power in range(NUM_COLUMNS - 1, -1, -1): if digits[power] != 0: return power return MIN_DISPLAY_POWER def format_odometer_config_summary(cfg: OdometerConfig) -> str: roll_cap = _penny_roll_ramp_lag(cfg) roll_slow = _penny_roll_sec_at_lag(0, cfg) roll_fast = _penny_roll_sec_at_lag(roll_cap, cfg) if cfg.rate_curve or cfg.roll_curve_sec: chase_desc = ( f"curve[{len(cfg.rate_curve)}]" if cfg.rate_curve else f"{cfg.min_rate}->{cfg.max_rate}+{cfg.dime_max_rate}c/s" ) roll_desc = ( f"curve[{len(cfg.roll_curve_sec)}]" if cfg.roll_curve_sec else f"{roll_slow:.2f}s->{roll_fast:.2f}s" ) dime_desc = ( f"dime_catchup@{cfg.dime_catchup_enter_pennies}c" if cfg.dime_catchup_enter_pennies > 0 else "dimes=intermittent" ) return ( f"start={format_dollars(cfg.initial_display_pennies)} " f"chase={chase_desc} roll={roll_desc} " f"{dime_desc} @ {cfg.max_lag_pennies}c" ) dime_desc = ( f"dime_catchup@{cfg.dime_catchup_enter_pennies}c" if cfg.dime_catchup_enter_pennies > 0 else "dimes=intermittent" ) return ( f"start={format_dollars(cfg.initial_display_pennies)} " f"chase@{cfg.penny_max_lag_pennies}c roll@{roll_cap}c " f"{roll_slow:.2f}s->{roll_fast:.2f}s " f"{dime_desc} {cfg.min_rate}->{cfg.max_rate}+{cfg.dime_max_rate}c/s " f"@ {cfg.max_lag_pennies}c" ) def format_odometer_lag_map(cfg: OdometerConfig) -> str: engine = OdometerEngine(cfg, initial_target_pennies=0) lines = ["lag($) lag(c) chase roll_s roll_cap EFFECTIVE"] for dollars in ( 0.25, 0.5, 1.0, 2.0, 3.0, 4.0, 4.5, 4.75, 4.99, 5.0, 6.0, 7.0, 7.5, 8.0, 8.5, 8.99, 9.0, 10.0, ): lag = int(round(dollars * 100)) chase = engine._pennies_per_second(lag) roll = _penny_roll_sec_at_lag(lag, cfg) rcap = 1.0 / roll eff = min(chase, rcap) lines.append( f" ${dollars:4.2f} {lag:5d} {chase:5.2f} {roll:5.2f} {rcap:5.2f} {eff:5.2f}" ) return "\n".join(lines) def _parse_curve_points(raw: Any, value_key: str) -> Tuple[CurvePoint, ...]: if not isinstance(raw, (list, tuple)): return () points: Dict[int, float] = {} for entry in raw: lag: Optional[int] = None value: Optional[float] = None if isinstance(entry, dict): lag_raw = entry.get("lag_pennies", entry.get("lag", entry.get("x"))) value_raw = entry.get( value_key, entry.get("value", entry.get("y")), ) if lag_raw is None or value_raw is None: continue try: lag = max(0, int(lag_raw)) value = float(value_raw) except (TypeError, ValueError): continue elif isinstance(entry, (list, tuple)) and len(entry) >= 2: try: lag = max(0, int(entry[0])) value = float(entry[1]) except (TypeError, ValueError): continue if lag is None or value is None or value <= 0.0: continue points[lag] = value return tuple(sorted(points.items())) def _curve_value_at_lag(delta: int, curve: Tuple[CurvePoint, ...]) -> float: if not curve: return 0.0 if delta <= curve[0][0]: return curve[0][1] prev_lag, prev_value = curve[0] for lag, value in curve[1:]: if delta <= lag: span = max(lag - prev_lag, 1) t = (delta - prev_lag) / span return prev_value + t * (value - prev_value) prev_lag, prev_value = lag, value return curve[-1][1] class OdometerEngine: def __init__( self, config: Optional[OdometerConfig] = None, initial_target_pennies: Optional[int] = None, ): self.config = config or OdometerConfig() start = ( self.config.initial_display_pennies if initial_target_pennies is None else max(0, int(initial_target_pennies)) ) self.target_pennies: int = start self.display_float: float = float(start) self._last_target_change_time: float = time.perf_counter() self._carry_scroll_phase: List[float] = [0.0] * NUM_COLUMNS self._carry_finishing: List[bool] = [False] * NUM_COLUMNS self._carry_from_digit: List[Optional[int]] = [None] * NUM_COLUMNS self._carry_was_in: List[bool] = [False] * NUM_COLUMNS self._last_maintained_lag: Optional[int] = None self._penny_scroll_visual: float = 0.0 self._smoothed_roll_sec: float = slowest_roll_sec(self.config) # True while lag-gated dime-driven catch-up owns +10¢ advances. self._dime_driven: bool = False # Free-spin digit while dime-driven (settled pennies stay on a dime boundary). self._cosmetic_penny_digit: Optional[int] = None def set_target_pennies(self, target_pennies: Optional[int]) -> None: if target_pennies is None: return t = max(0, int(target_pennies)) if t != self.target_pennies: self.target_pennies = t self._last_target_change_time = time.perf_counter() self.snap_if_too_far_behind() def display_immediate_pennies(self, target_pennies: Optional[int]) -> None: """Snap target and display immediately to a known starting/live value.""" if target_pennies is None: return t = max(0, int(target_pennies)) self.target_pennies = t self._last_target_change_time = time.perf_counter() self._snap_display_to_settled_cents(t) def set_target(self, target_pennies: Optional[int]) -> None: self.set_target_pennies(target_pennies) def bump_target_pennies(self, delta_pennies: int) -> None: self.set_target_pennies(self.target_pennies + int(delta_pennies)) def bump_target(self, delta_pennies: int) -> None: self.bump_target_pennies(delta_pennies) @property def display_pennies(self) -> int: return self._settled_cents() def is_animating(self) -> bool: return ( self.lag_pennies() > 0 or self._dime_driven or self._penny_scroll_visual > 1e-6 or any(p > 1e-6 for p in self._carry_scroll_phase) or any(self._carry_finishing) ) @property def catchup_mode(self) -> str: """HUD/debug: who owns catch-up math (`dime` or `penny`).""" return "dime" if self._dime_driven else "penny" def _settled_cents(self) -> int: return int(self.display_float) def snap_if_won(self) -> bool: overshoot = int(self.display_float) - self.target_pennies if overshoot <= self.config.max_lag_pennies: return False self._snap_display_to_settled_cents(self.target_pennies) return True def snap_if_too_far_behind(self) -> bool: """If lag exceeds threshold, jump display to a fixed lag behind target.""" lag = self.lag_pennies() threshold = max(0, int(self.config.snap_behind_threshold_pennies)) target_lag = max(0, int(self.config.snap_behind_to_lag_pennies)) if threshold <= 0 or lag <= threshold: return False snap_display = max(0, self.target_pennies - target_lag) if self._settled_cents() == snap_display: return False self._snap_display_to_settled_cents(snap_display) return True def _contributions_active(self) -> bool: if self.lag_pennies() > 0: return True return ( time.perf_counter() - self._last_target_change_time ) < self.config.contribution_idle_sec def pennies_per_second(self, delta: Optional[int] = None) -> float: if delta is None: delta = self.lag_pennies() return self._pennies_per_second(delta) def effective_pennies_per_second(self, delta: Optional[int] = None) -> float: if delta is None: delta = self.lag_pennies() if delta <= 0: return 0.0 if self._dime_driven or ( self.config.dime_catchup_enter_pennies > 0 and delta >= int(self.config.dime_catchup_enter_pennies) ): # Decoupled dimes: one digit roll = +10¢ at the lag-tapered roll period. roll = max(self._smoothed_roll_sec, 1e-6) return 10.0 / roll chase = min(self._pennies_per_second(delta), self._chase_animation_cap(delta)) ceil = self._intermittent_display_ceiling() headroom = max(0.0, ceil - float(self._settled_cents())) * 60.0 return min(chase, headroom) def _pennies_per_second(self, delta: int) -> float: cfg = self.config if delta <= 0: return 0.0 if not self._contributions_active(): return max(cfg.min_rate, delta / cfg.idle_catchup_sec) if cfg.rate_curve: rate = _curve_value_at_lag(delta, cfg.rate_curve) else: penny_cap = max(cfg.penny_max_lag_pennies, 1) if delta <= penny_cap: t = delta / penny_cap rate = cfg.min_rate + t * (cfg.max_rate - cfg.min_rate) else: penny_rate = cfg.max_rate band = max(cfg.max_lag_pennies - cfg.penny_max_lag_pennies, 1) excess = min(delta - cfg.penny_max_lag_pennies, band) t = excess / band rate = penny_rate + t * cfg.dime_max_rate scale = max(0.05, float(cfg.chase_rate_scale)) rate = rate * scale threshold = max(0, int(cfg.catchup_lag_threshold)) bonus = max(0.0, float(cfg.catchup_lag_bonus_per_cent)) if ( threshold > 0 and bonus > 0 and delta > threshold and self._contributions_active() ): rate += (delta - threshold) * bonus return rate def _chase_animation_cap(self, delta: int) -> float: return 1.0 / _penny_roll_sec_at_lag( min(max(0, delta), _penny_roll_ramp_lag(self.config)), self.config ) def _column_fraction(self, display_float: float, power: int) -> float: unit = 10**power return (display_float % unit) / unit def _snap_display_to_settled_cents(self, cents: Optional[int] = None) -> None: value = self._settled_cents() if cents is None else max(0, int(cents)) self.display_float = float(value) self._penny_scroll_visual = 0.0 self._smoothed_roll_sec = slowest_roll_sec(self.config) self._dime_driven = False self._cosmetic_penny_digit = None self._clear_all_carry_scroll() def _discard_sub_cent_slop(self) -> None: settled = self._settled_cents() if abs(self.display_float - settled) > 1e-9: self.display_float = float(settled) if self.lag_pennies() <= 0: self._penny_scroll_visual = 0.0 self._dime_driven = False self._cosmetic_penny_digit = None self._clear_all_carry_scroll() def _smooth_roll_sec(self, dt: float) -> float: """Exponentially smooth roll duration toward the lag-derived target. One-sided: deceleration (longer roll = slower speed) is immediate so the odometer always approaches lag=0 at the slowest scroll speed. Acceleration (shorter roll = faster speed) eases in over roll_smooth_tau to prevent jerks. """ # Dime-driven uses the same lag→roll curve as pennies so speed tapers # as the gap closes (floor still caps max speed inside _penny_roll_sec_at_lag). target = _penny_roll_sec_at_lag( min(max(1, self.lag_pennies()), _penny_roll_ramp_lag(self.config)), self.config, ) if target >= self._smoothed_roll_sec: self._smoothed_roll_sec = target else: tau = max(self.config.roll_smooth_tau, 1e-6) alpha = 1.0 - math.exp(-dt / tau) self._smoothed_roll_sec += (target - self._smoothed_roll_sec) * alpha return self._smoothed_roll_sec def _penny_roll_step(self, dt: float) -> float: return dt / max(self._smoothed_roll_sec, 1e-6) def _chase_needs_penny_carry_chain(self) -> bool: settled = self._settled_cents() gap = self.target_pennies - settled return gap > 0 and (settled % 10) == 9 def _update_dime_mode_state(self) -> None: """Enter/exit lag-gated dime-driven catch-up (hysteresis).""" enter = max(0, int(self.config.dime_catchup_enter_pennies)) if enter <= 0: if self._dime_driven and self._carry_scroll_phase[1] <= 1e-6: self._dime_driven = False self._cosmetic_penny_digit = None self._clear_carry_scroll(1) return exit_lag = max(0, int(self.config.dime_catchup_exit_pennies)) if exit_lag >= enter: exit_lag = max(0, enter // 2) lag = self.lag_pennies() dime_inflight = self._carry_scroll_phase[1] > 1e-6 if not self._dime_driven: if lag >= enter: self._dime_driven = True # Match lag taper immediately (same curve pennies would use). self._smoothed_roll_sec = _penny_roll_sec_at_lag( min(max(1, lag), _penny_roll_ramp_lag(self.config)), self.config, ) self._cosmetic_penny_digit = self._settled_cents() % 10 return # Hand back one dime before the exit band so pennies own the final # approach. Wait until cosmetic pennies land on the settled digit with # scroll idle — no mid-roll digit/scroll reset on mode switch. handoff = exit_lag + 10 if lag < handoff and not dime_inflight: settled_digit = self._settled_cents() % 10 cosmetic = self._cosmetic_penny_digit aligned = cosmetic is None or ( int(cosmetic) % 10 == settled_digit and self._penny_scroll_visual <= 1e-6 ) if not aligned: return self._dime_driven = False self._cosmetic_penny_digit = None self._clear_carry_scroll(1) def _chase_penny_cosmetic(self, dt: float) -> None: """Free-spin pennies 0-9 at current roll rate without advancing value.""" if self._cosmetic_penny_digit is None: self._cosmetic_penny_digit = self._settled_cents() % 10 step = self._penny_roll_step(dt) self._penny_scroll_visual = min(1.0, self._penny_scroll_visual + step) if self._penny_scroll_visual >= 1.0 - 1e-9: self._penny_scroll_visual = 0.0 self._cosmetic_penny_digit = (int(self._cosmetic_penny_digit) + 1) % 10 def _next_dime_boundary(self, settled: Optional[int] = None) -> int: value = self._settled_cents() if settled is None else max(0, int(settled)) return (value // 10) * 10 + 10 @staticmethod def _digit_changes_crossing(before: int, after: int, power: int) -> bool: """True when this decimal column's glyph changes between two whole-cent values.""" unit = 10**power return ((max(0, int(before)) // unit) % 10) != ((max(0, int(after)) // unit) % 10) def _powers_scrolling_with_next_dime(self, settled: Optional[int] = None) -> List[int]: """Dollar+ powers whose digits wrap when the next independent +10¢ lands.""" value = self._settled_cents() if settled is None else max(0, int(settled)) nxt = min(self._next_dime_boundary(value), self.target_pennies) if nxt <= value: return [] return [ power for power in range(2, NUM_COLUMNS) if self._digit_changes_crossing(value, nxt, power) ] def _higher_columns_pending_for_next_dime(self) -> bool: """True while a dollar+ cascade for the pending next-dime boundary is unfinished.""" for power in self._powers_scrolling_with_next_dime(): phase = self._carry_scroll_phase[power] if phase < 1.0 - 1e-9: return True return False def _clear_dime_cascade_columns(self) -> None: """Clear dimes + any dollar+ phases that ride the independent +10¢ land.""" self._clear_carry_scroll(1) for power in range(2, NUM_COLUMNS): if self._carry_scroll_phase[power] > 1e-6 or self._carry_finishing[power]: self._clear_carry_scroll(power) def _land_independent_dime(self) -> None: """Complete a decoupled dime roll: next dime boundary only (never snap to target).""" settled = self._settled_cents() next_dime = self._next_dime_boundary(settled) if next_dime > self.target_pennies: # Remainder is pennies — abort; penny chase finishes without a jump. self._clear_dime_cascade_columns() return self.display_float = float(next_dime) # Keep cosmetic penny free-spin; do not force digit back to 0 each land. self._clear_dime_cascade_columns() def _update_dime_driven_catchup(self, dt: float) -> None: """Dimes own +10¢ on their own schedule; pennies are cosmetic only. When the next +10¢ changes dollars (and higher), those columns scroll in sync via ``_update_intermittent_scroll`` — do not land until they finish. """ self._chase_penny_cosmetic(dt) settled = self._settled_cents() gap = self.target_pennies - settled phase = self._carry_scroll_phase[1] step = self._penny_roll_step(dt) exit_lag = max(0, int(self.config.dime_catchup_exit_pennies)) # Leave the exit band (+ one dime) for penny settle — avoids a fast final # +10¢ land then an abrupt mode switch. min_gap_to_start = max(10, exit_lag + 10) if phase <= 1e-6: if gap < min_gap_to_start: return self._carry_from_digit[1] = (settled // 10) % 10 self._carry_was_in[1] = True self._carry_finishing[1] = False self._carry_scroll_phase[1] = min(1.0, step) return if gap < 10: # Target moved / remainder too small — cancel in-flight, no value snap. self._clear_dime_cascade_columns() return phase = min(1.0, phase + step) self._carry_scroll_phase[1] = phase # Land after dollar+ intermittent updates this frame (see update()). def _try_land_independent_dime(self) -> None: """Apply +10¢ land when dime roll is complete and higher digits finished.""" if not self._dime_driven: return if self._carry_scroll_phase[1] < 1.0 - 1e-9: return if self._higher_columns_pending_for_next_dime(): return self._land_independent_dime() def _chase_cent_up(self, dt: float) -> None: settled = self._settled_cents() if self.target_pennies <= settled: return if self._chase_needs_penny_carry_chain(): step = self._penny_roll_step(dt) self._penny_scroll_visual = min(1.0, self._penny_scroll_visual + step) return step = self._penny_roll_step(dt) self._penny_scroll_visual = min(1.0, self._penny_scroll_visual + step) if self._penny_scroll_visual >= 1.0 - 1e-9: self._penny_scroll_visual = 0.0 self.display_float = float(min(settled + 1, self.target_pennies)) def _chase_cent_down(self, dt: float) -> None: settled = self._settled_cents() if settled <= self.target_pennies: return step = self._penny_roll_step(dt) if self._penny_scroll_visual > 1e-6: self._penny_scroll_visual = max(0.0, self._penny_scroll_visual - step) if self._penny_scroll_visual <= 1e-9: self._penny_scroll_visual = 0.0 self.display_float = float(max(settled - 1, self.target_pennies)) return self.display_float = float(max(settled - 1, self.target_pennies)) def update(self, dt: float) -> None: if dt <= 0: return self.snap_if_won() self.snap_if_too_far_behind() self._update_dime_mode_state() self._smooth_roll_sec(dt) settled = self._settled_cents() if self.target_pennies > settled: if self._dime_driven: # Start/advance independent dimes, then dollars that change on # that land, then apply the +10¢ only when higher cascades done. self._update_dime_driven_catchup(dt) self._update_intermittent_scroll(dt, min_power=2) self._try_land_independent_dime() else: self._chase_cent_up(dt) self._update_intermittent_scroll(dt, min_power=1) elif self.target_pennies < settled: self._dime_driven = False self._cosmetic_penny_digit = None self._chase_cent_down(dt) self._sync_intermittent_phases_to_display() self._update_intermittent_scroll(dt, min_power=1) else: self._discard_sub_cent_slop() self._update_intermittent_scroll(dt, min_power=1) def _clear_carry_scroll(self, power: int) -> None: self._carry_scroll_phase[power] = 0.0 self._carry_finishing[power] = False self._carry_from_digit[power] = None self._carry_was_in[power] = False def _clear_all_carry_scroll(self) -> None: for power in range(NUM_COLUMNS): self._clear_carry_scroll(power) def _sync_intermittent_phases_to_display(self) -> None: df = self.display_float for power in range(1, NUM_COLUMNS): phase = self._carry_scroll_phase[power] if phase <= 1e-6: continue if self._column_cascade_active(power): target = self._intermittent_scroll_target(power) elif self._carry_finishing[power]: target = self._column_fraction(df, power) else: continue if phase > target + 1e-6: self._carry_scroll_phase[power] = target @staticmethod def _digit_at_power(settled_cents: int, power: int) -> int: return (settled_cents // (10**power)) % 10 def _display_is_chasing(self) -> bool: return self.lag_pennies() > 0 def _carry_roll_in_progress(self, power: int) -> bool: phase = self._carry_scroll_phase[power] return (phase > 1e-6 and phase < 1.0 - 1e-9) or self._carry_finishing[power] def _column_cascade_active(self, power: int) -> bool: if power <= 0: return False # Dime column is owned by independent catch-up while dime-driven. if power == 1 and self._dime_driven: return False if self._carry_roll_in_progress(power): return True if not self._display_is_chasing(): return False settled = self._settled_cents() if power == 1: return self._column_in_carry(settled, 1) # Dime-driven +10¢: scroll every dollar+ column whose digit changes on the # pending next-dime land (e.g. $2.90→$3.00, $9.90→$10.00) — not only xx.99. if self._dime_driven: nxt = min(self._next_dime_boundary(settled), self.target_pennies) if nxt <= settled: return False return self._digit_changes_crossing(settled, nxt, power) if not self._column_in_carry(settled, power): return False if self._digit_at_power(settled, power) >= 9: return True lower = power - 1 return ( self._carry_roll_in_progress(lower) or self._carry_finishing[lower] or self._column_cascade_active(lower) ) def _intermittent_scroll_target(self, power: int) -> float: if not self._column_cascade_active(power): return 0.0 return 1.0 @staticmethod def _carry_window_cap(display_float: float, power: int) -> float: unit = 10**power tail = (int(display_float) // unit) * unit return float(tail + unit - 1) + 0.999999 def _carry_display_hold_cap(self, display_float: float) -> Optional[float]: # Independent dime catch-up must not freeze value at …x9 while dimes lead. if self._dime_driven: return None caps: List[float] = [] for power in range(1, NUM_COLUMNS): if self._carry_roll_in_progress(power): caps.append(self._carry_window_cap(display_float, power)) elif self._column_cascade_active(power): caps.append(self._carry_window_cap(display_float, power)) if not caps: return None return min(caps) def _bump_display_after_carry(self, power: int) -> None: for higher in range(power + 1, NUM_COLUMNS): if self._carry_roll_in_progress(higher) or self._carry_finishing[higher]: return settled = self._settled_cents() if settled >= self.target_pennies: return gap = self.target_pennies - settled if gap == 1 and (settled % 10) == 9: self._snap_display_to_settled_cents(self.target_pennies) return unit = 10**power if (settled % unit) < unit - 1: return bumped = min((settled // unit) * unit + unit, self.target_pennies) if bumped > settled: self.display_float = float(bumped) self._penny_scroll_visual = 0.0 def _intermittent_display_ceiling(self) -> float: ceiling = float(self.target_pennies) cap = self._carry_display_hold_cap(self.display_float) if cap is not None: ceiling = min(ceiling, cap) return ceiling def _update_intermittent_scroll(self, dt: float, min_power: int = 1) -> None: if dt <= 0: return start = max(1, int(min_power)) pending_dime_powers = ( set(self._powers_scrolling_with_next_dime()) if self._dime_driven else set() ) for power in range(start, NUM_COLUMNS): settled = (self._settled_cents() // (10**power)) % 10 phase = self._carry_scroll_phase[power] step = dt / max(self._smoothed_roll_sec, 1e-6) if phase >= 1.0 - 1e-9: # Under dime catch-up, hold finished dollar+ glyphs until the # independent +10¢ land clears the chain (do not restart). if self._dime_driven and power in pending_dime_powers: self._carry_scroll_phase[power] = 1.0 continue self._clear_carry_scroll(power) self._bump_display_after_carry(power) continue cascading = self._column_cascade_active(power) was_in = self._carry_was_in[power] if cascading: if not was_in: self._carry_from_digit[power] = settled self._carry_was_in[power] = True self._carry_finishing[power] = False target = self._intermittent_scroll_target(power) if target > phase: self._carry_scroll_phase[power] = min(target, phase + step) continue self._carry_was_in[power] = False if phase > 1e-6 and self._carry_from_digit[power] is not None: self._carry_finishing[power] = True phase = min(1.0, phase + step) self._carry_scroll_phase[power] = phase if phase >= 1.0 - 1e-9: if self._dime_driven and power in pending_dime_powers: self._carry_scroll_phase[power] = 1.0 continue self._clear_carry_scroll(power) self._bump_display_after_carry(power) continue self._clear_carry_scroll(power) @staticmethod def _column_in_carry(settled_cents: int, power: int) -> bool: if power <= 0: return False mod = 10**power return (settled_cents % mod) >= mod - 1 def get_display_digits(self) -> List[int]: return _pennies_to_digits(int(self.display_float)) def get_column_draw(self, power: int) -> Tuple[int, float]: if power < 0 or power >= NUM_COLUMNS: return 0, 0.0 if power == 0: if self._cosmetic_penny_digit is not None: return int(self._cosmetic_penny_digit) % 10, self._penny_scroll_visual return self._settled_cents() % 10, self._penny_scroll_visual unit = 10**power settled = (self._settled_cents() // unit) % 10 phase = self._carry_scroll_phase[power] if phase <= 1e-6: return settled, 0.0 from_digit = self._carry_from_digit[power] if from_digit is None: from_digit = settled return from_digit, min(phase, 1.0) def get_column_scroll(self, power: int) -> float: return self.get_column_draw(power)[1] def lag_pennies(self) -> int: return max(0, self.target_pennies - self._settled_cents()) def maintain_lag(self, lag_pennies: int) -> None: lag = min(max(0, int(lag_pennies)), self.config.max_lag_pennies) prev_lag = self._last_maintained_lag if prev_lag is not None and lag < prev_lag: self._sync_intermittent_phases_to_display() self._penny_scroll_visual = 0.0 self.target_pennies = max(0, self._settled_cents() + lag) self._last_maintained_lag = lag self._last_target_change_time = time.perf_counter() def reset_display_to_target(self) -> None: self._snap_display_to_settled_cents(self.target_pennies) self._last_maintained_lag = None