"""Drive progress over fixed widgets and print one key=value per line.

ONE INTERPRETER START DOES THE WHOLE OF IT. Every python3 start on this arm is a
fresh load of one large static image (99-python3 records the cost), so the suite
that drives this asserts on the OUTPUT rather than calling python3 once per case.

Every line is `key=value`; a key is emitted at most once. The suite reads the
keys it names and ignores the rest, so adding a fact here cannot break it.
Failure is reported as a key too -- `*_error' -- rather than only as an exit
status, because the case that reds should say which stage stopped and why.

WHAT IS BEING MEASURED IS A STREAM, AND IT IS EMITTED TWICE.
progress writes CARRIAGE RETURNS, not lines: `writeln' emits `\\r' + the redrawn
widget and never a newline, so a widget's whole output is one run of bytes whose
CR positions are the content. Each stream is therefore emitted as

  <name>_sha  -- a 16-hex digest of the EXACT characters the widget wrote,
                 trailing spaces and control bytes included; and
  <name>_vis  -- the same characters with `\\r' shown as `~', `\\n' as `<NL>' and
                 the two cursor escapes as `<HIDE>' / `<SHOW>'.

Neither is sufficient alone. The digest is exact and illegible; the visible form
is what a reviewer can check against the widget they expected, and it is the only
form that survives a shell literal at all -- a raw CR in a test file would be
eaten by every editor that touched it.

THE TTY GATE IS WHY A StringIO NEEDS `check_tty=False', AND IT IS MEASURED.
`Infinite.writeln' writes only `if self.file and self.is_tty()', and `is_tty'
calls `self.file.isatty()' unless `check_tty' is False. A StringIO HAS isatty and
answers False, so a widget handed one with the default `check_tty' writes NOTHING
-- not a truncated widget, nothing at all. `tty_gate_*' below measures both sides
of that gate, because it is the single fact that decides whether this package is
assertable off a terminal, and a suite that merely passed `check_tty=False'
without asserting why would leave the reason in a comment.

`hide_cursor' is False everywhere except the one `finish' fixture. It is not
tidiness: with it True, `__del__' prints SHOW_CURSOR to the stream at an
unspecified time, so a stream read after the widget became garbage would differ
from one read before it. The `finish' fixture keeps it True and calls `finish()',
which clears `_hidden_cursor' so `__del__' adds nothing.

THE CLOCK IS REPLACED, NOT TOLERATED. `eta' and `avg' are derived from
`monotonic()', which `progress/__init__.py' binds at module scope -- so the probe
rebinds `progress.monotonic' to a fixed sequence and the two quantities become
literals. Each `next()' consumes TWO ticks: one in `next' and one inside
`update_avg'. The sequence below is chosen so every `avg' is a whole number and
every `progress'/`percent' a dyadic rational, which is what makes them assertable
as exact decimal text rather than as a tolerance.

WHERE THE VALUES THE SUITE ASSERTS COME FROM. Every literal in
99-python3-progress.test was produced by running THIS file on x86-64 against the
sdist the recipe's SRC_URI sha256 pins (c9c86e98...34cd, verified by digest
before measuring). The control interpreter is CPython 3.10 and the image ships
3.12; that gap is stated in the suite header and the test plan rather than
papered over. The on-target result is not yet measured -- the batch verify is
this port's first run on wasm32-hwjs.
"""
import hashlib
import io
import sys

_seen = set()


def emit(key, value):
    if key in _seen:
        return
    _seen.add(key)
    # One line per fact: the suite greps by key, and an embedded newline would
    # land as a second, unkeyed line.
    print("%s=%s" % (key, str(value).replace("\n", "~").replace("\r", " ")))


def fail(key, exc):
    emit(key, "%s: %s" % (type(exc).__name__, exc))
    raise SystemExit(1)


def sha16(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]


def vis(text):
    """The stream as a reviewer can read it. See the module header."""
    return (text.replace("\x1b[?25l", "<HIDE>")
                .replace("\x1b[?25h", "<SHOW>")
                .replace("\r", "~")
                .replace("\n", "<NL>"))


def emit_stream(name, text):
    """Both readings of one widget's output. See the module header."""
    emit("%s_sha" % name, sha16(text))
    emit("%s_vis" % name, vis(text))


try:
    import progress
    from progress.bar import (Bar, ChargingBar, FillingCirclesBar,
                              FillingSquaresBar, IncrementalBar, PixelBar,
                              ShadyBar)
    from progress.counter import Counter, Countdown, Pie, Stack
    from progress.spinner import (LineSpinner, MoonSpinner, PieSpinner,
                                  PixelSpinner, Spinner)
except Exception as exc:                                  # noqa: BLE001
    fail("import_error", exc)

emit("module_file", getattr(progress, "__file__", ""))
emit("version_attr", getattr(progress, "__version__", ""))

try:
    from importlib.metadata import version as _md_version
    emit("dist_version", _md_version("progress"))
except Exception as exc:                                  # noqa: BLE001
    emit("dist_version_error", "%s: %s" % (type(exc).__name__, exc))

# It is a PACKAGE, not a module. setup.py declares `packages=["progress"]', so
# what lands in site-packages is a DIRECTORY whose __init__.py carries the two
# base classes -- the opposite shape from python3-texttable's single file, and
# the reason `module_file' above names an __init__.py.
emit("is_package", "yes" if hasattr(progress, "__path__") else "no")
emit("submodules", ",".join(sorted(
    n.split(".", 1)[1] for n in sys.modules if n.startswith("progress."))))

# NO C ANYWHERE IN THE IMPORT PATH. This image's python3 is one static binary
# with an empty lib-dynload (BRINTOS-1476), so a module reached through a .so
# could not be imported here at all. Every module this package loaded is
# recorded with the file it came from, so a release that grew an accelerator
# announces itself here rather than as an ImportError nobody can attribute.
emit("submodule_suffixes", ",".join(sorted(set(
    (getattr(m, "__file__", "") or "").rsplit(".", 1)[-1]
    for n, m in sys.modules.items() if n == "progress" or
    n.startswith("progress.")))))

# The two cursor escapes are module-scope constants a caller may emit itself,
# and they are what the `finish' fixture below asserts coming out of the widget.
emit("hide_cursor_repr", repr(progress.HIDE_CURSOR))
emit("show_cursor_repr", repr(progress.SHOW_CURSOR))

# ------------------------------------------------------------ the hierarchy ---
# Two roots and nothing else: `Infinite' for the widgets that do not know where
# they are going, `Progress' for the ones that do. Which root a widget has is
# what decides whether `max', `remaining', `percent' and `eta' exist on it at
# all, so it is asserted per class rather than left to the import.
for _cls in (Bar, ChargingBar, FillingSquaresBar, FillingCirclesBar,
             IncrementalBar, PixelBar, ShadyBar, Spinner, PieSpinner,
             MoonSpinner, LineSpinner, PixelSpinner, Counter, Countdown,
             Stack, Pie):
    emit("base_%s" % _cls.__name__,
         ",".join(c.__name__ for c in _cls.__mro__[1:-1]))

# The class attributes a caller overrides. Asserted as VALUES because they are
# the defaults every rendering below is drawn at: a release that changed
# `Bar.width' would move every digest in this file at once, and this is the case
# that names the cause.
emit("bar_width", Bar.width)
emit("bar_suffix", Bar.suffix)
emit("bar_prefix_repr", repr(Bar.bar_prefix))
emit("bar_bar_suffix_repr", repr(Bar.bar_suffix))
emit("bar_fill", Bar.fill)
emit("bar_empty_fill_repr", repr(Bar.empty_fill))
emit("charging_suffix", ChargingBar.suffix)
emit("charging_fill", ChargingBar.fill)
emit("charging_empty_fill", ChargingBar.empty_fill)
emit("sma_window", progress.Infinite.sma_window)
emit("default_max", progress.Progress(file=None).max)

# THE PHASE ALPHABETS, in order. They are the whole content of a spinner and the
# last character of an incremental bar, so each is emitted as an ordered join --
# a set comparison would pass on an implementation that cycled them backwards.
for _n, _cls in (("spinner", Spinner), ("pie_spinner", PieSpinner),
                 ("moon_spinner", MoonSpinner), ("line_spinner", LineSpinner),
                 ("pixel_spinner", PixelSpinner), ("incremental", IncrementalBar),
                 ("pixel_bar", PixelBar), ("shady_bar", ShadyBar),
                 ("stack", Stack), ("pie", Pie)):
    emit("phases_%s" % _n, "".join(_cls.phases))
    emit("nphases_%s" % _n, len(_cls.phases))

# --------------------------------------------------------------- the tty gate ---
# See the module header. This is the fact that makes the package assertable off a
# terminal, and both sides of it are measured.
_g = io.StringIO()
_w = Bar(max=8, file=_g)                      # check_tty defaults True
_w.next()
_w.next()
emit("tty_gate_default_len", len(_g.getvalue()))
emit("tty_gate_default_index", _w.index)      # it COUNTED; it just did not write
emit("tty_gate_isatty", "yes" if _g.isatty() else "no")
del _w

_g2 = io.StringIO()
_w2 = Bar(max=8, file=_g2, check_tty=False, hide_cursor=False)
_w2.next()
emit("tty_gate_off_len", len(_g2.getvalue()))
del _w2

# `file=None' is the third way to silence a widget, and it is the one that needs
# no isatty at all -- `if self.file and ...' short-circuits before is_tty runs.
_w3 = Bar(max=8, file=None)
_w3.next()
emit("file_none_index", _w3.index)
del _w3


# A stream with no isatty and check_tty left True is the documented AttributeError,
# and the message names the remedy. `str(self)' is in it and carries the object's
# address, so only the part after it is emitted.
class _NoIsatty(object):
    def write(self, s):
        pass

    def flush(self):
        pass


try:
    Bar(max=8, file=_NoIsatty())
    emit("isatty_missing_raised", "no")
except AttributeError as exc:
    _m = str(exc)
    _i = _m.find(" has no attribute")
    emit("isatty_missing_raised", "yes")
    emit("isatty_missing_msg", _m[_i + 1:] if _i >= 0 else _m)
except Exception as exc:                                  # noqa: BLE001
    emit("isatty_missing_raised", "wrong-type")
    emit("isatty_missing_exc", type(exc).__name__)

# ------------------------------------------------------------- the renderings ---
# THE FIXTURE IS ONE ADVANCE SCHEDULE. Every bar below is a `max' of 8 advanced
# four times, so the four widgets differ only in their own alphabet and suffix
# and a red names the widget rather than the schedule. 8 is not arbitrary: it
# makes index/max a dyadic rational at every step, so `progress' and `percent'
# are exact decimal text (see the module header) and `int(32 * progress)' lands
# on a whole character rather than on a rounding.
MAXV = 8
STEPS = 4


def drive(factory, steps=STEPS, **kw):
    sio = io.StringIO()
    kw.setdefault("check_tty", False)
    kw.setdefault("hide_cursor", False)
    w = factory(file=sio, **kw)
    for _ in range(steps):
        w.next()
    return w, sio.getvalue()


_b, _s = drive(lambda **kw: Bar("Loading ", max=MAXV, **kw))
emit_stream("bar", _s)
emit("bar_index", _b.index)
_b, _s = drive(lambda **kw: ChargingBar("Charging ", max=MAXV, **kw))
emit_stream("charging", _s)
_b, _s = drive(lambda **kw: FillingSquaresBar("Squares ", max=MAXV, **kw))
emit_stream("squares", _s)
_b, _s = drive(lambda **kw: FillingCirclesBar("Circles ", max=MAXV, **kw))
emit_stream("circles", _s)

# THE INCREMENTAL BARS TAKE A DIFFERENT `max' ON PURPOSE. Their whole difference
# from `Bar' is the PARTIAL last character, and `max=8' -- which fills a whole
# number of the 32 columns at every step -- would leave `phase' zero throughout
# and render them identical to a plain bar with a different fill. `max=5' puts
# 32/5 = 6.4 columns in after one step, so the last character is phases[3] and
# the machinery is actually exercised.
_b, _s = drive(lambda **kw: PixelBar("Pixels ", max=5, **kw), steps=1)
emit_stream("pixel", _s)
_b, _s = drive(lambda **kw: ShadyBar("Shady ", max=5, **kw), steps=1)
emit_stream("shady", _s)
_b, _s = drive(lambda **kw: IncrementalBar("Incr ", max=5, **kw), steps=1)
emit_stream("incr", _s)

# THE SPINNERS DO NOT KNOW WHERE THEY ARE GOING -- they are `Infinite', so they
# have no `max' and no percentage. Driven a full cycle PLUS ONE (`len(phases)+1'
# advances) so the wrap back to phases[0] is in the stream: a spinner that ran
# off the end of its alphabet would pass a case that stopped at the last phase.
_b, _s = drive(lambda **kw: Spinner("Spin ", **kw), steps=5)
emit_stream("spinner", _s)
emit("spinner_wrapped", "yes" if _b.index == 5 else "no")
_b, _s = drive(lambda **kw: PieSpinner("Pie ", **kw), steps=5)
emit_stream("pie_spinner", _s)

# THE COUNTERS. `Counter' is `Infinite' and counts UP with no ceiling;
# `Countdown' is `Progress' and prints what is LEFT -- the same four advances
# give 1,2,3,4 from one and 7,6,5,4 from the other, which is the pair that says
# they are not the same widget with a different name.
_b, _s = drive(lambda **kw: Counter("Count ", **kw))
emit_stream("counter", _s)
_b, _s = drive(lambda **kw: Countdown("Left ", max=MAXV, **kw))
emit_stream("countdown", _s)
_b, _s = drive(lambda **kw: Stack("Stack ", max=MAXV, **kw))
emit_stream("stack", _s)
_b, _s = drive(lambda **kw: Pie("Pie ", max=MAXV, **kw))
emit_stream("pie", _s)

# ---------------------------------------------------------------- finish() ---
# The ONE fixture that keeps `hide_cursor' at its default True. It is what makes
# the terminating sequence assertable end to end: HIDE_CURSOR at construction,
# then `finish()' writing a newline and SHOW_CURSOR back. `finish' clears
# `_hidden_cursor', so `__del__' adds nothing afterwards and the stream is stable
# whenever it is read (see the module header).
_fs = io.StringIO()
_fb = Bar("Done ", max=MAXV, file=_fs, check_tty=False)
_fb.next()
_fb.finish()
emit_stream("finish", _fs.getvalue())
emit("finish_hidden_cursor_cleared", "no" if _fb._hidden_cursor else "yes")

# ...and the context manager is the same two calls with the widget deciding when.
# `__enter__' calls `start()', which on a `Progress' draws the widget at index 0
# -- so the with-block stream carries a leading render a bare constructor does
# not, and `__exit__' calls `finish()'.
_cs = io.StringIO()
with Bar("Ctx ", max=MAXV, file=_cs, check_tty=False, hide_cursor=False) as _cb:
    _cb.next()
emit_stream("ctx", _cs.getvalue())

# `iter' is the wrapper a caller actually reaches for, and on a `Progress' it
# takes `max' from the iterable's own length -- so a sized iterable needs no
# `max' at all. The yielded values are asserted beside the stream, because a
# generator that advanced the bar but dropped an item would still draw correctly.
_is = io.StringIO()
_ib = Bar("Iter ", file=_is, check_tty=False, hide_cursor=False)
emit("iter_values", ",".join(str(x) for x in _ib.iter(["a", "b", "c"])))
emit("iter_max", _ib.max)
emit_stream("iter", _is.getvalue())

# `goto' is `next' by absolute index rather than by delta, and it is the only way
# to move a bar BACKWARDS. Asserted because a caller resuming a transfer uses it.
_gs = io.StringIO()
_gb = Bar("Goto ", max=MAXV, file=_gs, check_tty=False, hide_cursor=False)
_gb.goto(6)
_gb.goto(2)
emit("goto_index", _gb.index)
emit_stream("goto", _gs.getvalue())

# ----------------------------------------------- the computed properties ------
# Read DIRECTLY rather than through a rendering. `Bar''s own suffix shows only
# `index' and `max', so a `progress' or `percent' that had drifted would not move
# any digest above -- these are the cases that cover them.
_ps = io.StringIO()
_pb = Bar(max=MAXV, file=_ps, check_tty=False, hide_cursor=False)
_prog, _pct, _idx, _rem = [], [], [], []
for _ in range(STEPS):
    _pb.next()
    _idx.append(str(_pb.index))
    _rem.append(str(_pb.remaining))
    _prog.append(repr(_pb.progress))
    _pct.append(repr(_pb.percent))
emit("prop_index", ",".join(_idx))
emit("prop_remaining", ",".join(_rem))
emit("prop_progress", ",".join(_prog))
emit("prop_percent", ",".join(_pct))

# THE TWO CLAMPS, which are the reason `progress' and `remaining' are properties
# rather than arithmetic at the call site. Overshoot the max and `progress'
# saturates at 1 while `remaining' floors at 0 -- a bar driven past its end draws
# full rather than overflowing its 32 columns.
_os = io.StringIO()
_ob = Bar(max=MAXV, file=_os, check_tty=False, hide_cursor=False)
_ob.next(MAXV + 4)
emit("overshoot_index", _ob.index)
emit("overshoot_progress", repr(_ob.progress))
emit("overshoot_remaining", _ob.remaining)
emit_stream("overshoot", _os.getvalue())

# `max=0' is the division this would otherwise be. It is guarded in `progress'
# and the widget draws empty rather than raising.
_zb = progress.Progress(max=0, file=None)
emit("zero_max_progress", repr(_zb.progress))
emit("zero_max_percent", repr(_zb.percent))

# `__getitem__' is what `message % self' and `suffix % self' go through, and its
# one rule is that a key beginning with `_' answers None -- so a format string
# cannot reach into the widget's private state. An unknown key answers None too
# rather than raising, which is why a bad `%(typo)s' renders `None'.
emit("getitem_public", _pb["index"])
emit("getitem_private", _pb["_ts"])
emit("getitem_unknown", _pb["nosuchkey"])

# --------------------------------------------------- avg and eta, on a fixed clock ---
# See the module header: `progress.monotonic' is rebound so the two time-derived
# quantities become literals. The ticks are consumed TWO per `next()' -- one in
# `next', one inside `update_avg' -- and the deltas they produce are 1, 3, 2, 2
# seconds, which give whole-number averages at every step.
TICKS = [1000.0,
         1001.0, 1001.0,
         1004.0, 1004.0,
         1006.0, 1006.0,
         1008.0, 1008.0]
_tick = {"i": 0}


def _fake_monotonic():
    i = _tick["i"]
    if i < len(TICKS):
        _tick["i"] = i + 1
        return TICKS[i]
    return TICKS[-1]                      # exhausted: the clock stands still


_real_monotonic = progress.monotonic
progress.monotonic = _fake_monotonic
try:
    _cs2 = io.StringIO()
    _cb2 = Bar(max=MAXV, file=_cs2, check_tty=False, hide_cursor=False)
    _avgs, _etas = [], []
    for _ in range(STEPS):
        _cb2.next()
        _avgs.append(repr(_cb2.avg))
        _etas.append(str(_cb2.eta))
    emit("clock_avg", ",".join(_avgs))
    emit("clock_eta", ",".join(_etas))
    emit("clock_ticks_used", _tick["i"])
    # `elapsed' is read after the sequence is exhausted, so the standing clock
    # makes it the whole span of the fixture rather than a race with the host.
    emit("clock_elapsed", _cb2.elapsed)
    emit("clock_elapsed_td", _cb2.elapsed_td)
    emit("clock_eta_td", _cb2.eta_td)
    emit("clock_xput_len", len(_cb2._xput))
finally:
    progress.monotonic = _real_monotonic

emit("probe_complete", "yes")