Appendix III: The Finished edisc.py

The whole tool as S26 leaves it: 618 lines, one file, nothing outside Python's own standard library. It runs on Python 3.9 and later. It is also in the course download as edisc.py.

"""edisc.py - list, read and write Tatung Einstein .dsk disc images.

Built section by section in the TatungBytes course "Inside the
Einstein: The Disc". One file, standard library only, Python 3.9 or later.
"""

import argparse
import os
import sys


def show_byte(b):
    if 32 <= b < 127:
        return chr(b)
    return "."


def dump(data, start, length):
    for offset in range(start, start + length, 16):
        row = data[offset:offset + 16]
        hex_part = ""
        text_part = ""
        for b in row:
            hex_part = hex_part + f"{b:02X} "
            text_part = text_part + show_byte(b)
        print(f"{offset:06X}  {hex_part:<48} {text_part}")


def read_image(filename):
    try:
        with open(filename, "rb") as f:
            disc = f.read()
    except FileNotFoundError:
        sys.exit(f"There is no file called {filename}")
    check_image(disc)
    return disc


def check_image(disc):
    if len(disc) < 256:
        sys.exit("This is not a disc image: it is too short")
    image_format(disc)
    if disc[0x30] != 40 or disc[0x31] != 1:
        sys.exit(f"This image has {disc[0x30]} tracks and {disc[0x31]} sides;"
                 " an Einstein disc has 40 tracks and 1 side")
    if 256 + sum(track_sizes(disc)) != len(disc):
        sys.exit("This image is damaged: its size does not match its header")
    offsets = track_offsets(disc)
    for track in range(40):
        if disc[offsets[track]:offsets[track] + 12] != b"Track-Info\r\n":
            sys.exit(f"This image is damaged: track {track} has no Track-Info header")


def image_format(disc):
    if disc[0:16] == b"EXTENDED CPC DSK":
        return "extended"
    if disc[0:8] == b"MV - CPC":
        return "standard"
    sys.exit("This is not a disc image: it does not start EXTENDED CPC DSK or MV - CPC")


def track_sizes(disc):
    form = image_format(disc)
    count = disc[0x30] * disc[0x31]
    sizes = []
    if form == "extended":
        for i in range(count):
            sizes.append(disc[0x34 + i] * 256)
    else:
        size = int.from_bytes(disc[0x32:0x34], "little")
        for i in range(count):
            sizes.append(size)
    return sizes


def track_offsets(disc):
    offsets = []
    offset = 256
    for size in track_sizes(disc):
        offsets.append(offset)
        offset = offset + size
    return offsets


def sector_ids(disc, track):
    offset = track_offsets(disc)[track]
    count = disc[offset + 0x15]
    ids = []
    for i in range(count):
        entry = offset + 0x18 + i * 8
        ids.append(disc[entry + 2])
    return ids


def find_sector(disc, track, sector_id):
    offset = track_offsets(disc)[track]
    ids = sector_ids(disc, track)
    if sector_id not in ids:
        sys.exit(f"This image is damaged: track {track} has no sector {sector_id}")
    position = ids.index(sector_id)
    start = offset + 0x100 + position * 512
    return start


def read_sector(disc, track, sector_id):
    start = find_sector(disc, track, sector_id)
    return disc[start:start + 512]


def read_directory(disc):
    directory = b""
    for sector_id in range(4):
        directory = directory + read_sector(disc, 2, sector_id)
    return directory


def entry_text(raw):
    text = ""
    for b in raw:
        text = text + show_byte(b)
    return text


def entry_name(entry):
    name = entry_text(entry[1:9]).rstrip()
    ext = entry_text(bytes([entry[9] & 0x7F]) + entry[10:12]).rstrip()
    return name + "." + ext


def entry_blocks(entry):
    field = entry[16:32]
    sixteen_bit = True
    for i in range(1, 16, 2):
        if field[i] != 0:
            sixteen_bit = False
    blocks = []
    if sixteen_bit:
        for i in range(0, 16, 2):
            block = int.from_bytes(field[i:i + 2], "little")
            if block != 0:
                blocks.append(block)
    else:
        for block in field:
            if block != 0:
                blocks.append(block)
    return blocks


def live_entries(disc):
    directory = read_directory(disc)
    entries = []
    for slot in range(64):
        entry = directory[slot * 32:slot * 32 + 32]
        if entry[0] != 0xE5:
            entries.append(entry)
    return entries


def block_sectors(block):
    places = []
    for i in range(4):
        logical = block * 4 + i
        places.append((logical // 10 + 2, logical % 10))
    return places


def read_block(disc, block):
    data = b""
    for track, sector_id in block_sectors(block):
        data = data + read_sector(disc, track, sector_id)
    return data


def find_file(disc, name):
    entries = []
    for entry in live_entries(disc):
        if entry_name(entry).upper() == name.upper():
            entries.append(entry)
    if len(entries) == 0:
        sys.exit(f"There is no file called {name} on this disc")
    return entries


def read_file(disc, name):
    entries = find_file(disc, name)
    data = b""
    for extent in range(len(entries)):
        entry = None
        for e in entries:
            if e[12] == extent:
                entry = e
        if entry is None:
            sys.exit(f"{name} is damaged: its extent {extent} is missing")
        part = b""
        for block in entry_blocks(entry):
            part = part + read_block(disc, block)
        data = data + part[:entry[15] * 128]
    return data


def host_name(name):
    safe = ""
    for ch in name:
        if ch.isalnum() or ch in "._-":
            safe = safe + ch
        else:
            safe = safe + "_"
    return safe


def save_image(filename, disc):
    temporary = filename + ".tmp"
    with open(temporary, "wb") as f:
        f.write(disc)
    os.replace(temporary, filename)


def blank_image():
    disc = bytearray(256)
    disc[0:34] = b"EXTENDED CPC DSK File\r\nDisk-Info\r\n"
    disc[0x22:0x27] = b"edisc"
    disc[0x30] = 40
    disc[0x31] = 1
    for track in range(40):
        disc[0x34 + track] = 0x15
    for track in range(40):
        header = bytearray(256)
        header[0:12] = b"Track-Info\r\n"
        header[0x10] = track
        header[0x11] = 0
        header[0x14] = 2
        header[0x15] = 10
        header[0x16] = 0x2A
        header[0x17] = 0xE5
        for sector_id in range(10):
            entry = 0x18 + sector_id * 8
            header[entry] = track
            header[entry + 1] = 0
            header[entry + 2] = sector_id
            header[entry + 3] = 2
            header[entry + 6:entry + 8] = (512).to_bytes(2, "little")
        disc = disc + header + bytes([0xE5]) * 5120
    return disc


def used_blocks(disc):
    used = []
    for entry in live_entries(disc):
        for block in entry_blocks(entry):
            used.append(block)
    return used


def free_blocks(disc):
    used = used_blocks(disc)
    free = []
    for block in range(1, 95):
        if block not in used:
            free.append(block)
    return free


def free_slots(disc):
    directory = read_directory(disc)
    free = []
    for slot in range(64):
        if directory[slot * 32] == 0xE5:
            free.append(slot)
    return free


def write_sector(disc, track, sector_id, data):
    start = find_sector(disc, track, sector_id)
    disc[start:start + 512] = data


def write_directory(disc, directory):
    for sector_id in range(4):
        write_sector(disc, 2, sector_id, directory[sector_id * 512:sector_id * 512 + 512])


def split_name(name):
    name = name.upper()
    parts = name.split(".")
    if len(parts) == 1:
        parts.append("")
    if len(parts) != 2:
        sys.exit(f"{name}: a disc name has at most one dot")
    base, ext = parts
    if len(base) < 1 or len(base) > 8 or len(ext) > 3:
        sys.exit(f"{name}: a disc name is 1 to 8 characters, a dot, and up to 3 more")
    for ch in base + ext:
        if ch <= " " or ch > "~" or ch in "*?:":
            sys.exit(f"{name}: the character {ch!r} cannot be used in a disc name")
    return base, ext


def make_entry(base, ext, extent, records, blocks):
    entry = bytearray(32)
    entry[0] = 0
    entry[1:9] = base.encode("ascii").ljust(8)
    entry[9:12] = ext.encode("ascii").ljust(3)
    entry[12] = extent
    entry[15] = records
    for i in range(len(blocks)):
        entry[16 + i * 2:18 + i * 2] = blocks[i].to_bytes(2, "little")
    return entry


def check_unlocked(disc, name):
    for entry in find_file(disc, name):
        if entry[9] & 0x80:
            sys.exit(f"{name.upper()} is locked: edisc will not change it")


def name_in_use(disc, name):
    for entry in live_entries(disc):
        if entry_name(entry).upper() == name.upper():
            return True
    return False


def write_block(disc, block, data):
    places = block_sectors(block)
    for i in range(4):
        track, sector_id = places[i]
        write_sector(disc, track, sector_id, data[i * 512:i * 512 + 512])


def add_file(disc, path, disc_name):
    base, ext = split_name(disc_name)
    if name_in_use(disc, base + "." + ext):
        sys.exit(f"{base}.{ext} is already on this disc")
    with open(path, "rb") as f:
        data = f.read()
    if len(data) == 0:
        sys.exit(f"{path} is empty: there is nothing to add")
    size = len(data)
    records = (size + 127) // 128
    data = data + bytes([0x1A]) * (records * 128 - size)
    count = (len(data) + 2047) // 2048
    extents = (len(data) + 16383) // 16384
    blocks = free_blocks(disc)[:count]
    slots = free_slots(disc)[:extents]
    if len(blocks) < count:
        sys.exit(f"The disc is full: {path} needs {count * 2}K")
    if len(slots) < extents:
        sys.exit(f"The disc's directory is full: {path} needs {extents} slot(s)")
    for i in range(count):
        chunk = data[i * 2048:i * 2048 + 2048]
        write_block(disc, blocks[i], chunk + bytes([0xE5]) * (2048 - len(chunk)))
    directory = bytearray(read_directory(disc))
    for extent in range(extents):
        part_blocks = blocks[extent * 8:extent * 8 + 8]
        part_records = min(128, records - extent * 128)
        slot = slots[extent]
        directory[slot * 32:slot * 32 + 32] = make_entry(base, ext, extent,
                                                         part_records, part_blocks)
    write_directory(disc, directory)
    print(f"Added {base}.{ext}: {size} bytes, {count} block(s), {extents} extent(s)")


def remove_file(disc, name):
    check_unlocked(disc, name)
    directory = bytearray(read_directory(disc))
    removed = 0
    for slot in range(64):
        entry = directory[slot * 32:slot * 32 + 32]
        if entry[0] != 0xE5 and entry_name(entry).upper() == name.upper():
            directory[slot * 32] = 0xE5
            removed = removed + 1
    write_directory(disc, directory)
    print(f"Removed {name.upper()}: {removed} directory entr{'y' if removed == 1 else 'ies'}")


def info(filename, disc):
    sizes = track_sizes(disc)
    expected = 256 + sum(sizes)
    print("Image:   ", filename)
    print("Format:  ", image_format(disc))
    print("Tracks:  ", disc[0x30])
    print("Sides:   ", disc[0x31])
    print("Tracks are", min(sizes), "to", max(sizes), "bytes")
    print("Expected size:", expected, "bytes")
    print("Actual size:  ", len(disc), "bytes")
    if expected == len(disc):
        print("The sizes add up.")
    else:
        print("The sizes DO NOT add up.")


def tracks(disc):
    print("Track  Starts at  Magic  Track Side Sectors  N  Gap3 Filler")
    offsets = track_offsets(disc)
    for i in range(len(offsets)):
        offset = offsets[i]
        header = disc[offset:offset + 256]
        if header[0:12] == b"Track-Info\r\n":
            magic = "yes"
        else:
            magic = "NO"
        print(f"{i:5}  {offset:9}  {magic:5}  {header[0x10]:5} {header[0x11]:4}"
              f" {header[0x15]:7} {header[0x14]:2}  {header[0x16]:02X}   {header[0x17]:02X}")


def sectors(disc, track):
    offset = track_offsets(disc)[track]
    count = disc[offset + 0x15]
    print("Track", track, "has", count, "sectors, stored in this order:")
    print("Position  C  H  ID  N  ST1 ST2  Length  Data starts at")
    for i in range(count):
        entry = disc[offset + 0x18 + i * 8:offset + 0x18 + i * 8 + 8]
        length = int.from_bytes(entry[6:8], "little")
        data_at = offset + 0x100 + i * 512
        print(f"{i:8}  {entry[0]:2} {entry[1]:2} {entry[2]:3} {entry[3]:2}"
              f"  {entry[4]:02X}  {entry[5]:02X}  {length:6}  {data_at}")


def dump_sector(disc, track, sector_id):
    start = find_sector(disc, track, sector_id)
    print(f"Track {track}, sector {sector_id}: bytes {start} to {start + 511} of the image")
    dump(disc[start:start + 512], 0, 512)


def show_directory(disc):
    directory = read_directory(disc)
    used = 0
    for slot in range(64):
        entry = directory[slot * 32:slot * 32 + 32]
        if entry[0] != 0xE5:
            print("Slot", slot)
            dump(entry, 0, 32)
            used = used + 1
    print(used, "of 64 directory slots in use")


def show_entries(disc):
    directory = read_directory(disc)
    for slot in range(64):
        entry = directory[slot * 32:slot * 32 + 32]
        if entry[0] == 0xE5:
            continue
        locked = ""
        if entry[9] & 0x80:
            locked = "  locked"
        print(f"Slot {slot:2}: {entry_name(entry):12}  status {entry[0]:02X}"
              f"  extent {entry[12]}  records {entry[15]:3}  blocks {entry_blocks(entry)}{locked}")


def list_files(disc):
    entries = live_entries(disc)
    names = []
    for entry in entries:
        name = entry_name(entry)
        if name.upper() not in names:
            names.append(name.upper())
    used = 0
    for upper_name in names:
        records = 0
        blocks = 0
        extents = 0
        locked = ""
        for entry in entries:
            if entry_name(entry).upper() == upper_name:
                shown = entry_name(entry)
                records = records + entry[15]
                blocks = blocks + len(entry_blocks(entry))
                extents = extents + 1
                if entry[9] & 0x80:
                    locked = "  locked"
        print(f"{shown:12} {records * 128:7} bytes {blocks * 2:4}K  {extents} extent(s){locked}")
        used = used + blocks * 2
    print(f"{len(names)} file(s): {used}k Size, {188 - used}k Free, 190k Total")


def show_block(disc, block):
    places = block_sectors(block)
    for track, sector_id in places:
        print(f"Block {block}: track {track}, sector {sector_id}")
    dump(read_block(disc, block), 0, 128)


def type_file(disc, name):
    text = ""
    for b in read_file(disc, name):
        if b == 0x1A:
            break
        if b == 10:
            text = text + "\n"
        elif b != 13:
            text = text + show_byte(b)
    print(text)


def extract(disc, name, out):
    data = read_file(disc, name)
    if out is None:
        out = host_name(name)
    with open(out, "wb") as f:
        f.write(data)
    print(f"Wrote {out}: {len(data)} bytes")


def create(filename):
    if os.path.exists(filename):
        sys.exit(f"{filename} already exists: edisc will not overwrite it")
    save_image(filename, blank_image())
    print(f"Created {filename}: a blank Einstein disc, 188K free")


def show_free(disc):
    used = used_blocks(disc)
    line = "D"
    for block in range(1, 95):
        if block in used:
            line = line + "#"
        else:
            line = line + "."
    print("Blocks 0-94:  D = directory, # = in use, . = free")
    for start in range(0, 95, 50):
        print(f"{start:3}  {line[start:start + 50]}")
    print(f"Free: {len(free_blocks(disc))} blocks ({len(free_blocks(disc)) * 2}K),"
          f" {len(free_slots(disc))} of 64 directory slots")


def rename_file(disc, old, new):
    check_unlocked(disc, old)
    base, ext = split_name(new)
    if name_in_use(disc, base + "." + ext):
        sys.exit(f"{base}.{ext} is already on this disc")
    directory = bytearray(read_directory(disc))
    for slot in range(64):
        entry = directory[slot * 32:slot * 32 + 32]
        if entry[0] != 0xE5 and entry_name(entry).upper() == old.upper():
            directory[slot * 32 + 1:slot * 32 + 9] = base.encode("ascii").ljust(8)
            directory[slot * 32 + 9:slot * 32 + 12] = ext.encode("ascii").ljust(3)
    write_directory(disc, directory)
    print(f"Renamed {old.upper()} to {base}.{ext}")


def main():
    parser = argparse.ArgumentParser(
        prog="edisc.py",
        description="List, read and write Tatung Einstein .dsk disc images.")
    commands = parser.add_subparsers(dest="command", required=True)

    def command(name, help_text):
        sub = commands.add_parser(name, help=help_text, description=help_text)
        sub.add_argument("image", help="the .dsk disc image")
        return sub

    command("info", "check an image's header and size")
    command("header", "dump the image's 256-byte disk-info header")
    command("tracks", "list every track's header")
    command("sectors", "list one track's sectors").add_argument("track", type=int)
    sub = command("dump", "dump one sector")
    sub.add_argument("track", type=int)
    sub.add_argument("sector", type=int)
    command("directory", "dump the directory slots in use")
    command("entries", "decode every directory entry in use")
    command("list", "list the files on the disc")
    command("block", "show where a block is, and its first bytes").add_argument("block", type=int)
    command("type", "show a text file").add_argument("name")
    sub = command("extract", "copy a file off the disc")
    sub.add_argument("name")
    sub.add_argument("out", nargs="?", help="the file to write (default: the disc name)")
    command("free", "show the free blocks and directory slots")
    command("create", "make a new, blank disc image")
    sub = command("add", "copy a file onto the disc")
    sub.add_argument("path", help="the file on your computer")
    sub.add_argument("name", nargs="?", help="the name on the disc (default: the file's own)")
    sub = command("rename", "rename a file on the disc")
    sub.add_argument("old")
    sub.add_argument("new")
    command("remove", "delete a file from the disc").add_argument("name")

    args = parser.parse_args()

    if args.command == "create":
        create(args.image)
        return
    disc = read_image(args.image)
    if args.command == "info":
        info(args.image, disc)
    elif args.command == "header":
        dump(disc, 0, 256)
    elif args.command == "tracks":
        tracks(disc)
    elif args.command == "sectors":
        sectors(disc, args.track)
    elif args.command == "dump":
        dump_sector(disc, args.track, args.sector)
    elif args.command == "directory":
        show_directory(disc)
    elif args.command == "entries":
        show_entries(disc)
    elif args.command == "list":
        list_files(disc)
    elif args.command == "block":
        show_block(disc, args.block)
    elif args.command == "type":
        type_file(disc, args.name)
    elif args.command == "extract":
        extract(disc, args.name, args.out)
    elif args.command == "free":
        show_free(disc)
    else:
        disc = bytearray(disc)
        if args.command == "add":
            add_file(disc, args.path, args.name or os.path.basename(args.path))
        elif args.command == "rename":
            rename_file(disc, args.old, args.new)
        elif args.command == "remove":
            remove_file(disc, args.name)
        save_image(args.image, disc)


main()