Protocols¶

unidesa ¶

Unidesa/Cirsa simulator.

Binary frames, XOR55 checksums, counters read by numeric id. Cirsa speaks the same thing, so it gets the same code under a different name. Saves money on translators.

UnidesaSimulator `dataclass` ¶

UnidesaSimulator(
    name: str = "unidesa",
    state: MachineState = MachineState(),
    selected_counters: list[int] = (
        lambda: list(DEFAULT_SELECTED_COUNTER_IDS)
    )(),
    reply_sd: int = 255,
    reply_so: int = 0,
)

Simulates Unidesa/Cirsa framing and counter reads.

The readers build frames as STX SD SO NM CMD ND DATA CHK EXT, where CHK is the XOR55 checksum. The counter response payload is: group (1) | first (1) | received (1) | [id (1) | value (4 LE)] * received

Supported: access (0x0B), logout (0x06), custom counter selection (0x2E) and counter reads (0x16). Anything else gets error 0x01. That's enough to keep the reader scripts happy, and keeping the reader scripts happy is the entire point of my existence here.

handle ¶

handle(data: bytes) -> bytes

Answer exactly one frame. A bad frame raises ValueError instead of getting a reply.

Source code in src/morpheus_vlt/protocols/unidesa.py

def handle(self, data: bytes) -> bytes:
    """Answer exactly one frame. A bad frame raises ValueError instead of getting a reply."""
    # One frame in, one frame out. TCP can glue several together, and splitting them is the
    # caller's job (the CLI does it). Stay strict here so chaos actually has teeth.
    msg = UnidesaMessage.parse(data)
    msg.validate()

    if msg.cmd == CMD_ACCESS:
        return self._reply(msg, CMD_ACCESS, b"\x00")

    if msg.cmd == CMD_LOGOUT:
        return self._reply(msg, CMD_LOGOUT, b"\x00")

    if msg.cmd == CMD_CUSTOM_SELECT:
        # Payload is (group, counter_id) pairs. The group gets read and ignored, same as my
        # emails. An odd trailing byte gets dropped. An empty list keeps the old selection.
        ids: list[int] = []
        for i in range(0, len(msg.data), 2):
            if i + 1 >= len(msg.data):
                break
            _grp = msg.data[i]
            cid = msg.data[i + 1]
            ids.append(cid)
        if ids:
            self.selected_counters = ids
        return self._reply(msg, CMD_CUSTOM_SELECT, b"\x00")

    if msg.cmd == CMD_COUNTERS:
        return self._handle_counters(msg)

    # Never heard of it. Echo the command back with error 0x01 and move on with our lives.
    return self._reply(msg, msg.cmd, b"\x01")

sas ¶

SAS simulator, at the "shape of the bytes" level.

Real SAS runs on a 9-bit serial line where the ninth bit marks the address byte. A PTY or a TCP socket has eight bits and no opinion on the matter, so the electrical part is not emulated. What is emulated: correctly sized replies, BCD meters at the offsets the readers slice, and a CRC that saspy agrees with. The readers count bytes and cut at fixed offsets, so getting the sizes right is 90% of the job. The other 10% is also getting the sizes right.

SASSimulator `dataclass` ¶

SASSimulator(
    name: str = "sas",
    state: MachineState = MachineState(),
    addr: int = 1,
    idle_announce_interval: float = 0.5,
    _announce_address_until_activity: bool = True,
    _last_idle_announce: float = 0.0,
    _aft_registered: bool = False,
    _last_aft: tuple[int, int, int, int] = (0, 0, 0, 0),
    _last_transaction_id: bytes = (
        lambda: bytes.fromhex(
            "20202020202020202020202020202021"
        )
    )(),
)

SAS protocol-shape simulator for serial readers.

One request in, one reply out, no buffering. Meters come out of state.meters, mostly as BCD. AFT transfers actually move the "credits" meter, so a test can push money in and watch it land. Everything else is read-only and boringly consistent, which is the dream.

handle ¶

handle(data: bytes) -> bytes

Answer one SAS request. Empty data means the line is idle (see the PTY transport).

Source code in src/morpheus_vlt/protocols/sas.py

def handle(self, data: bytes) -> bytes:
    """Answer one SAS request. Empty `data` means the line is idle (see the PTY transport)."""
    if not data:
        return self._idle_address_broadcast()

    self._announce_address_until_activity = False
    req = self._strip_host_prefix(data)
    if not req:
        return b""

    cmd = req[0]

    # General poll. Answer 0x00, "no events", which is the most honest thing a machine says.
    if cmd in (CMD_EVENT_POLL, (CMD_EVENT_POLL + self.addr) & 0xFF):
        return b"\x00"

    if cmd in ACK_ONLY_COMMANDS:
        return bytes([self.addr & 0xFF])

    if cmd == CMD_SELECT_METERS:
        return self._reply_select_meters()

    if cmd == CMD_CUSTOM_METERS:
        return self._reply_custom_meters()

    if cmd in SINGLE_METER_COMMANDS:
        return self._reply_single_meter(cmd)

    if cmd == 0x18:
        return self._reply_powerup_meters()

    if cmd == 0x19:
        return self._reply_meters_11_15()

    if cmd == 0x1C:
        return self._reply_meters_1c()

    if cmd == 0x1E:
        return self._reply_bill_meters()

    if cmd == 0x1F:
        return self._reply_gaming_machine_id()

    if cmd == 0x1B:
        return self._reply_handpay_info()

    if cmd == 0x21:
        return self._reply_rom_signature()

    if cmd == 0x3D:
        return self._reply_cashout_ticket_info()

    if cmd == 0x48:
        return self._reply_last_accepted_bill_info()

    if cmd == 0x4C:
        return self._reply_secure_validation_id()

    if cmd == 0x4D:
        return self._reply_enhanced_validation_info()

    if cmd == 0x4F:
        return self._reply_current_hopper_status()

    if cmd == 0x50:
        return self._reply_validation_meters(req)

    if cmd == 0x51:
        return self._reply_total_games()

    if cmd == 0x52:
        return self._reply_game_meters(req)

    if cmd == 0x53:
        return self._reply_game_configuration(req)

    if cmd == 0x54:
        return self._reply_sas_version_serial()

    if cmd == 0x55:
        return self._reply_selected_game()

    if cmd == 0x56:
        return self._reply_enabled_games()

    if cmd == 0x57:
        return self._reply_pending_cashout_info()

    if cmd == 0x58:
        return self._reply_receive_validation_number()

    if cmd == 0x6E:
        return self._reply_authentication_info()

    if cmd == 0x70:
        return self._reply_ticket_validation_data()

    if cmd == 0x71:
        return self._reply_redeem_ticket_response()

    if cmd == 0x72:
        return self._reply_aft(req)

    if cmd == 0x73:
        return self._reply_aft_register(req)

    if cmd == 0x74:
        return self._reply_aft_lock_status(req)

    if cmd == 0x7B:
        return self._reply_extended_validation_status()

    if cmd == 0x7E:
        return self._reply_current_datetime()

    if cmd == 0xA0:
        return self._reply_enabled_features(req)

    if cmd == 0x9A:
        return self._reply_legacy_bonus_meters(req)

    return self._reply_default(cmd)

vdai ¶

VDAI simulator. The machine prints its stats like a receipt, the reader regexes the receipt.

Nobody had a real printout lying around, so these layouts are rebuilt backwards from the reader's own parsing code (oblivion's lib/protocol/vdai.py): which lines its regex keeps, how it pulls the numbers out of them, and which positions it reads. The labels are made up. The positions, the number formats and the line shapes are exactly what the reader needs, and the tests run the reader's own expressions against them to keep it that way.

Money meters are in cents and get printed in euros. Counts get printed as counts.

VDAISimulator `dataclass` ¶

VDAISimulator(
    name: str = "vdai",
    state: MachineState = MachineState(),
    protocol_type: str = "m",
)

ASCII-style "printer" protocol simulator.

The reader sends an ESC/P-ish "stats" request ending in a line feed, then reads whatever comes back until the line goes quiet and digs the numbers out with regexes, one set per machine family. Every complete request line gets one stats page. Partial requests wait for the rest. Silence gets silence, which matters on a PTY where the transport pokes the protocol with empty input every 100 ms.

Meters used: coin_in, bill_in, coin_out, bill_out, plays, play and win for the generic and anniversario flavors, plus the in_/in_s_/in_h_/out_ denomination counts for anniversario and bally. tot_in and tot_out are computed, not read.

handle ¶

handle(data: bytes) -> bytes

Return one stats page per complete request line. What the request says is not consulted.

Source code in src/morpheus_vlt/protocols/vdai.py

def handle(self, data: bytes) -> bytes:
    """Return one stats page per complete request line. What the request says is not consulted."""
    if not data:
        return b""
    self._rx.extend(data)
    requests = self._rx.count(b"\n")
    if not requests:
        if len(self._rx) > MAX_REQUEST_BYTES:
            self._rx.clear()
        return b""
    del self._rx[: self._rx.rfind(b"\n") + 1]
    # Whatever it asked, the answer is the stats page. Like a support hotline.
    return self._page() * requests

rfranco ¶

R. Franco simulator. Plain ASCII lines, a password prompt, and Spanish. Lots of Spanish.

RFrancoSimulator `dataclass` ¶

RFrancoSimulator(
    name: str = "rfranco",
    state: MachineState = MachineState(),
    key: str = "RF",
    require_key: bool = True,
    echo_commands: bool = True,
)

ASCII line-oriented R. Franco simulator for the legacy reader flow.

Commands, one per line, case doesn't matter: C -> "CLAVE:" then the next line is the key ("RF" unless you change it) ESTADO -> "ESTADO=" plus 7 status bits TOTALES -> play, win and plays totals REGISTROS -> the detailed registers, only after the key, plus both door counters Anything else gets "ERR COMANDO". Each command gets echoed back first unless echo_commands is off. Lines end in CR, LF or CRLF, and they can arrive in as many pieces as TCP feels like.

authenticated `property` ¶

authenticated: bool

True once the right key went through. Wrong key flips it back to False.

handle ¶

handle(data: bytes) -> bytes

Buffer data, answer every complete line, keep the leftovers for next time.

Source code in src/morpheus_vlt/protocols/rfranco.py

def handle(self, data: bytes) -> bytes:
    """Buffer `data`, answer every complete line, keep the leftovers for next time."""
    if not data:
        return b""

    self._rx.extend(data)
    out = bytearray()

    while True:
        line = self._pop_line()
        if line is None:
            break

        line = line.strip()
        if not line:
            continue

        if self.echo_commands:
            out.extend(line)
            out.extend(b"\r\n")
        out.extend(self._dispatch_line(line))

    return bytes(out)

dex ¶

DEX/UCS simulator, playing the machine (VMC) side.

DEX is how vending machines hand over their audit data: two handshakes, then an EVA-DTS text file sent in DLE-framed blocks with a CRC-16 on each one. It's a protocol from the days when a handheld terminal was cutting-edge, and it has been feeling its age ever since.

The dance, from the reader's point of view: 1. reader ENQ -> sim DLE0 2. reader DLE SOH id/op/rev ... -> sim DLE1 (or NAK if the CRC is junk) 3. reader EOT -> sim ENQ (roles swap, the machine drives now) 4. reader DLE0/DLE1 -> sim second handshake (response code, id, revision) 5. reader DLE1 -> sim EOT ENQ 6. reader ACK -> sim first data block 7. reader ACK / NAK -> next block / same block again. After the last one, EOT.

DexState ¶

Bases: Enum

What the simulator is waiting for next. See the module docstring for the full dance.

DexSimulator `dataclass` ¶

DexSimulator(
    name: str = "dex",
    state: MachineState = MachineState(),
    communication_id: bytes = DEFAULT_COMMUNICATION_ID,
    revision_level: bytes = DEFAULT_REVISION_LEVEL,
    response_code: bytes = DEFAULT_RESPONSE_CODE,
    evadts_lines: list[str] = list(),
)

DEX/UCS simulator (VMC side).

Does the two handshakes and the data transfer, then resets and waits for the next reader. Bytes can arrive in any size chunks. Junk between messages gets thrown away. The EVA-DTS payload comes from a file, a list of lines, or a tiny built-in default if you gave it nothing. state is carried around for consistency with the other protocols but nothing reads it yet.