Add support to tapinfo.py for PZX files

skoolkit/tape.py

@@ -227,14 +227,15 @@ def __init__(self, blocks, version=None, warnings=()):
         self.warnings = warnings
 
 class TapeBlock:
-    def __init__(self, number, tape_data, timings=None, block_data=None, block_id=None, name=None, info=()):
+    def __init__(self, number, tape_data, timings=None, block_data=None, block_id=None, name=None, info=(), standard=True):
         self.number = number
         self.data = tape_data
         self.timings = timings
         self.block_data = block_data
         self.block_id = block_id
         self.name = name
         self.info = info
+        self.standard = standard
 
 class TapeBlockTimings:
     def __init__(self, pilot_len=0, pilot=0, sync=(), zero=0, one=0, pause=0, used_bits=8, pulses=(), error=None):
@@ -275,20 +276,110 @@ def _format_text(prefix, data, start, length, dump=False):
             lines.append(f'{indent}  {line}')
     return lines
 
+def _get_pzx_block(data, i, block_num, prev_rom_pilot):
+    # http://zxds.raxoft.cz/docs/pzx.txt
+    block_id = ''.join(chr(b) for b in data[i:i + 4])
+    block_len = get_dword(data, i + 4)
+    tape_data = None
+    standard = False
+    rom_pilot = False
+    info = []
+    if block_id == 'PZXT':
+        name = 'PZX header block'
+        info.append(f'Version: {data[i + 8]}.{data[i + 9]}')
+        pairs = ['Title', '']
+        j = i + 10
+        while j < i + 8 + block_len:
+            b = data[j]
+            if b:
+                pairs[-1] += chr(b)
+            else:
+                if len(pairs) == 2:
+                    info.append(f'{pairs[0]}: {pairs[1]}')
+                    pairs.clear()
+                pairs.append('')
+            j += 1
+        if len(pairs) == 2 and pairs[1]:
+            info.append(f'{pairs[0]}: {pairs[1]}')
+    elif block_id == 'PULS':
+        name = 'Pulse sequence'
+        pulses = []
+        j = i + 8
+        while j < i + 8 + block_len:
+            count = 1
+            duration = get_word(data, j)
+            j += 2
+            if duration > 0x8000:
+                count = duration % 0x8000
+                duration = get_word(data, j)
+                j += 2
+            if duration >= 0x8000:
+                duration = (duration % 0x8000) * 65536 + get_word(data, j)
+                j += 2
+            info.append(f'{count} x {duration} T-states')
+            pulses.append((count, duration))
+        rom_pilot = len(pulses) == 3 and pulses[0] in ((3223, 2168), (8063, 2168)) and pulses[1:] == [(1, 667), (1, 735)]
+    elif block_id == 'DATA':
+        name = 'Data block'
+        count = get_dword(data, i + 8)
+        bits = count % 0x80000000
+        num_bytes = bits // 8
+        used_bits = bits % 8
+        tail = get_word(data, i + 12)
+        p0, p1 = data[i + 14:i + 16]
+        j = i + 16
+        s0 = [get_word(data, k) for k in range(j, j + 2 * p0, 2)]
+        j += 2 * p0
+        s1 = [get_word(data, k) for k in range(j, j + 2 * p1, 2)]
+        j += 2 * p1
+        tape_data = data[j:j + num_bytes + int(used_bits > 0)]
+        standard = prev_rom_pilot and p0 == 2 and p1 == 2 and s0 == [855, 855] and s1 == [1710, 1710]
+        if used_bits:
+            info.append(f'Bits: {bits} ({num_bytes} bytes + {used_bits} bits)')
+        else:
+            info.append(f'Bits: {bits} ({num_bytes} bytes)')
+        info.extend((
+            f'Initial pulse level: {count >> 31}',
+            '0-bit pulse sequence: {} (T-states)'.format(', '.join(str(p) for p in s0)),
+            '1-bit pulse sequence: {} (T-states)'.format(', '.join(str(p) for p in s1)),
+            f'Tail pulse: {tail} T-states'
+        ))
+    elif block_id == 'PAUS':
+        name = 'Pause'
+        duration = get_dword(data, i + 8)
+        info.extend((
+            f'Duration: {duration % 0x80000000} T-states',
+            f'Initial pulse level: {duration >> 31}',
+        ))
+    elif block_id == 'BRWS':
+        name = 'Browse point'
+        info.append(''.join(chr(b) for b in data[i + 8:i + 8 + block_len]))
+    elif block_id == 'STOP':
+        name = 'Stop tape command'
+        flags = get_word(data, i + 8)
+        mode = ('Always', '48K only')[flags & 1]
+        info.append(f'Mode: {mode}')
+    else:
+        name = block_id
+    block = TapeBlock(block_num, tape_data, block_id=block_id, name=name, info=info, standard=standard)
+    return i + 8 + block_len, block, rom_pilot
+
 def _get_tzx_block(data, i, block_num, get_info, get_timings):
     # https://worldofspectrum.net/features/TZXformat.html
     block_id = data[i]
     block_data = None
     tape_data = None
     info = []
     timings = None
+    standard = False
     i += 1
     if block_id == 0x10:
         # Standard speed data block
         header = 'Standard speed data'
         pause = get_word(data, i)
         length = get_word(data, i + 2)
         tape_data = data[i + 4:i + 4 + length]
+        standard = True
         if get_info:
             info.append(f'Pause: {pause}ms')
         if get_timings and tape_data:
@@ -571,7 +662,7 @@ def _get_tzx_block(data, i, block_num, get_info, get_timings):
         i += 9
     else:
         raise SkoolKitError(f'Unknown TZX block ID: 0x{block_id:02X}')
-    return i, TapeBlock(block_num, tape_data, timings, block_data, block_id, header, info)
+    return i, TapeBlock(block_num, tape_data, timings, block_data, block_id, header, info, standard)
 
 def hex_dump(data, row_size=16):
     lines = []
@@ -582,6 +673,21 @@ def hex_dump(data, row_size=16):
         lines.append(f'{index:04X}  {values_hex} {values_text}')
     return lines
 
+def parse_pzx(pzx):
+    if isinstance(pzx, str):
+        pzx = read_bin_file(pzx)
+    if pzx[:4] != b'PZXT':
+        raise SkoolKitError('Not a PZX file')
+    blocks = []
+    block_num = 1
+    rom_pilot = False
+    i = 0
+    while i < len(pzx):
+        i, block, rom_pilot = _get_pzx_block(pzx, i, block_num, rom_pilot)
+        blocks.append(block)
+        block_num += 1
+    return Tape(blocks)
+
 def parse_tap(tap, start=1, stop=0):
     if isinstance(tap, str):
         tap = read_bin_file(tap)

skoolkit/tapinfo.py

@@ -19,22 +19,24 @@
 
 from skoolkit import SkoolKitError, get_word, get_int_param, warn, VERSION
 from skoolkit.basic import BasicLister, TextReader
-from skoolkit.tape import hex_dump, parse_tap, parse_tzx
+from skoolkit.tape import hex_dump, parse_pzx, parse_tap, parse_tzx
 
 def _bytes_to_str(data):
     return ', '.join(str(b) for b in data)
 
 def _print_info(text):
     print('  ' + text)
 
-def _print_block(index, data, show_data, text_reader, info=(), block_id=None, header=None):
+def _print_block(index, data, show_data, text_reader, info, block_id, header, standard):
     if block_id is None:
-        print("{}:".format(index))
+        print(f'{index}:')
+    elif isinstance(block_id, int):
+        print(f'{index}: {header} (0x{block_id:02X})')
     else:
-        print("{}: {} (0x{:02X})".format(index, header, block_id))
+        print(f'{index}: {header}')
     for line in info:
         _print_info(line)
-    if data and block_id in (None, 16):
+    if data and standard:
         data_type = "Unknown"
         name_str = None
         line = 0xC000
@@ -101,14 +103,14 @@ def _analyse_tape(tape, basic_block, text_reader, show_data):
         if basic_block:
             _list_basic(block.number, block.data, *basic_block)
         else:
-            _print_block(block.number, block.data, show_data, text_reader, block.info, block.block_id, block.name)
+            _print_block(block.number, block.data, show_data, text_reader, block.info, block.block_id, block.name, block.standard)
     for msg in tape.warnings:
         warn(msg)
 
 def main(args):
     parser = argparse.ArgumentParser(
         usage="tapinfo.py FILE",
-        description="Show the blocks in a TAP or TZX file.",
+        description="Show the blocks in a PZX, TAP or TZX file.",
         add_help=False
     )
     parser.add_argument('infile', help=argparse.SUPPRESS, nargs='?')
@@ -125,7 +127,9 @@ def main(args):
     basic_block = _get_basic_block(namespace.basic)
     text_reader = TextReader()
     tape_type = namespace.infile.lower()[-4:]
-    if tape_type == '.tap':
+    if tape_type == '.pzx':
+        tape = parse_pzx(namespace.infile)
+    elif tape_type == '.tap':
         tape = parse_tap(namespace.infile)
     elif tape_type == '.tzx':
         tape = parse_tzx(namespace.infile)

sphinx/source/changelog.rst

@@ -3,6 +3,7 @@ Changelog
 
 9.3b1
 -----
+* Added support to :ref:`tapinfo.py` for PZX files
 * Added support to :ref:`tap2sna.py <tap2sna-conf>` for the ``m`` (memory)
   replacement field in the ``TraceLine`` configuration parameter
 * Added support to :ref:`trace.py <trace-conf>` for the ``m`` (memory)

sphinx/source/commands.rst

@@ -1927,7 +1927,7 @@ Configuration parameters may also be set on the command line by using the
 
 tapinfo.py
 ----------
-`tapinfo.py` shows information on the blocks in a TAP or TZX file. For
+`tapinfo.py` shows information on the blocks in a PZX, TAP or TZX file. For
 example::
 
   $ tapinfo.py game.tzx
@@ -1936,7 +1936,7 @@ To list the options supported by `tapinfo.py`, run it with no arguments::
 
   usage: tapinfo.py FILE
 
-  Show the blocks in a TAP or TZX file.
+  Show the blocks in a PZX, TAP or TZX file.
 
   Options:
     -b N[,A], --basic N[,A]
@@ -1948,8 +1948,9 @@ To list the options supported by `tapinfo.py`, run it with no arguments::
 +---------+-------------------------------------------------------------------+
 | Version | Changes                                                           |
 +=========+===================================================================+
-| 9.3     | Shows info for TZX block type 0x18 (CSW recording); recognises    |
-|         | deprecated TZX block types 0x16, 0x17, 0x34 and 0x40              |
+| 9.3     | Added support for PZX files; shows info for TZX block type 0x18   |
+|         | (CSW recording); recognises deprecated TZX block types 0x16,      |
+|         | 0x17, 0x34 and 0x40                                               |
 +---------+-------------------------------------------------------------------+
 | 9.2     | Shows info for TZX block type 0x15 (direct recording)             |
 +---------+-------------------------------------------------------------------+

sphinx/source/conf.py

@@ -250,7 +250,7 @@
     ('man/tap2sna.py', 'tap2sna.py',
      'convert a TAP or TZX file into a snapshot file', _authors, 1),
     ('man/tapinfo.py', 'tapinfo.py',
-     'show the blocks in a TAP or TZX file', _authors, 1),
+     'show the blocks in a PZX, TAP or TZX file', _authors, 1),
     ('man/trace.py', 'trace.py',
      'simulate code execution in a 48K or 128K memory snapshot', _authors, 1)
 ]

sphinx/source/man/tapinfo.py.rst

@@ -10,7 +10,7 @@ SYNOPSIS
 
 DESCRIPTION
 ===========
-``tapinfo.py`` shows information on the blocks in a TAP or TZX file.
+``tapinfo.py`` shows information on the blocks in a PZX, TAP or TZX file.
 
 OPTIONS
 =======

tests/skoolkittest.py

@@ -27,6 +27,15 @@
 def as_dword(num):
     return (num % 256, (num >> 8) % 256, (num >> 16) % 256, (num >> 24) % 256)
 
+def as_word(num):
+    return (num % 256, (num >> 8) % 256)
+
+def as_words(values):
+    data = []
+    for v in values:
+        data.extend(as_word(v))
+    return data
+
 def get_parity(data):
     parity = 0
     for b in data:
@@ -341,6 +350,106 @@ def data(self):
             data.extend(input_recording)
         return data
 
+class PZX:
+    def __init__(self, major=1, minor=0, info=(), null=True):
+        info_bytes = []
+        for key, value in info:
+            if key:
+                info_bytes.extend(ord(c) for c in key)
+                info_bytes.append(0)
+            info_bytes.extend(ord(c) for c in value)
+            info_bytes.append(0)
+        if info_bytes and info_bytes[-1] == 0 and not null:
+            info_bytes.pop()
+        length = 2 + len(info_bytes)
+        self.data = [
+            80, 90, 88, 84,    # PZXT
+            *as_dword(length), # Block length
+            major, minor,      # Version
+            *info_bytes        # Tape info
+        ]
+
+    def _encode_pulses(self, count, duration):
+        if count == 1 and duration < 0x8000:
+            return as_word(duration)
+        if duration < 0x8000:
+            return (*as_word(0x8000 + count), *as_word(duration))
+        return (*as_word(0x8000 + count), *as_word(0x8000 + (duration >> 16)), *as_word(duration % 65536))
+
+    def add_puls(self, standard=1, pulses=(), pulse_data=None):
+        if pulse_data is None:
+            pulse_data = []
+            if pulses:
+                count = 1
+                prev_p = pulses[0]
+                for p in pulses[1:] + [None]:
+                    if p == prev_p:
+                        count += 1
+                    else:
+                        while count > 0:
+                            pulse_data.extend(self._encode_pulses(min(count, 0x7FFF), prev_p))
+                            count -= 0x7FFF
+                        count = 1
+                    prev_p = p
+            else:
+                pulse_data.extend(self._encode_pulses(3223 + 4840 * (standard > 1), 2168))
+                pulse_data.extend(self._encode_pulses(1, 667))
+                pulse_data.extend(self._encode_pulses(1, 735))
+        length = len(pulse_data)
+        self.data.extend((
+            80, 85, 76, 83,    # PULS
+            *as_dword(length), # Block length
+            *pulse_data        # Pulses
+        ))
+
+    def add_data(self, data, s0=(855, 855), s1=(1710, 1710), tail=945, used_bits=0, polarity=0):
+        length = 8 + 2 * (len(s0) + len(s1)) + len(data)
+        bits = 8 * len(data)
+        if used_bits:
+            bits += used_bits - 8
+        bits += polarity * 0x80000000
+        self.data.extend((
+            68, 65, 84, 65,    # DATA
+            *as_dword(length), # Block length
+            *as_dword(bits),   # Polarity and bit count
+            *as_word(tail),    # Tail pulse
+            len(s0),           # p0
+            len(s1),           # p1
+            *as_words(s0),     # s0
+            *as_words(s1),     # s1
+            *data              # Data
+        ))
+
+    def add_paus(self, duration=3500000, polarity=0):
+        value = polarity * 0x80000000 + (duration % 0x80000000)
+        self.data.extend((
+            80, 65, 85, 83,   # PAUS
+            *as_dword(4),     # Block length
+            *as_dword(value)  # Duration
+        ))
+
+    def add_brws(self, text):
+        text_bytes = [ord(c) for c in text]
+        length = len(text)
+        self.data.extend((
+            66, 82, 87, 83,    # BRWS
+            *as_dword(length), # Block length
+            *text_bytes        # Text
+        ))
+
+    def add_stop(self, always=True):
+        flags = int(not always)
+        self.data.extend((
+            83, 84, 79, 80, # STOP
+            *as_dword(2),   # Block length
+            *as_word(flags) # Flags
+        ))
+
+    def add_block(self, name, data):
+        block = [32, 32, 32, 32, *as_dword(len(data)), *data]
+        block[:len(name)] = [ord(c) for c in name[:4]]
+        self.data.extend(block)
+
 class SkoolKitTestCase(TestCase):
     stdout_binary = False