Files
psp-preservation-tool/app/ps1_convert.py
T
quoije ea08f52d44 ps1: general STARTDAT lookup + warn on old base.pbp loader
- extract_startdat: locate STARTDAT at PSAR + p1 (the u32 at PSAR+0x0C)
  instead of popstation's p1 + 0x50000 absolute offset, which only worked
  for bases whose PSAR was aligned to 0x50000. Any PS1 classic can now be
  used as a base.
- _warn_old_base: print a warning when the base PBP is a 3.xx-era classic,
  since its POPS loader often fails to link on 6.60/6.61 CFW (error
  80020148 'game could not be started').
- README: document the 6.60/6.61 base preference and DEFLATE compression.
2026-09-07 17:36:16 -04:00

1132 lines
41 KiB
Python

"""PS1 -> PSP EBOOT.PBP conversion (pure Python).
Converts a PlayStation 1 disc image (``.iso``, ``.img``, or a ``.cue`` +
``.bin`` set) into the ``EBOOT.PBP`` "PS1 classic" container that PSP POPS
firmware can launch.
This is an independent, MIT-clean reimplementation of the popstation-md
algorithm (same format that PSXPackager/RupertAvery writes), implemented from
the published container format rather than by copying any GPL source.
Format reference (offsets are byte-exact):
* PBP header - 40 bytes: b"\\x00PBP" magic + version + 8 uint32 section offsets
(SFO, ICON0, ICON1, PIC0, PIC1, SND0, DATA.PSP, DATA.PSAR).
* PSAR - the DATA.PSAR section holding the disc image. Starts with
b"PSISOIMG0000", game id at +0x400, TOC at +0x800,
ISO block index at +0x4000, disc data at +0x100000.
A ``base.pbp`` (any PSN PS1 classic ``EBOOT.PBP``) is REQUIRED to supply two
blobs that we cannot synthesise:
* DATA.PSP - the encrypted POPS loader (header section 8).
* STARTDAT - the trailing encrypted block (boot.png + PGD).
Without a base.pbp the resulting EBOOT will not boot on real hardware; the
converter refuses to run without one.
Compression: the disc data is split into 0x9300-byte blocks and deflated
(raw DEFLATE, no zlib wrapper) at the configured level (default 1). POPS on
real PSP hardware refuses to boot uncompressed (level-0) PSARs, so level 0
is kept only for diagnostics; a valid conversion always uses level >= 1.
"""
from __future__ import annotations
import argparse
import hashlib
import io
import os
import re
import struct
import zlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterator, List, Optional, Tuple
# --------------------------------------------------------------------------
# Constants
# --------------------------------------------------------------------------
PBP_MAGIC = b"\x00PBP"
PBP_HEADER_SIZE = 0x28
PBP_HEADER_U32 = 10 # magic + version + 8 section offsets
# Section index -> header offset (in bytes) within the 40-byte header.
SECTION_OFFSET = {
"sfo": 0x08,
"icon0": 0x0C,
"icon1": 0x10,
"pic0": 0x14,
"pic1": 0x18,
"snd0": 0x1C,
"data_psp": 0x20,
"psar": 0x24,
}
SECTION_ORDER = ["sfo", "icon0", "icon1", "pic0", "pic1", "snd0", "data_psp", "psar"]
# PSAR (DATA.PSAR) layout.
PSAR_MAGIC = b"PSISOIMG0000"
PSAR_GAMEID_OFFSET = 0x400
PSAR_TOC_OFFSET = 0x800
PSAR_INDEX_OFFSET = 0x4000
PSAR_ISO_OFFSET = 0x100000
# popstation aligns the DATA.PSAR section to a 0x50000 boundary (the base PBP
# also places it there). Some POPS loaders rely on this alignment.
PSAR_ALIGNMENT = 0x50000
# One disc "block" in the ISO index = 16 raw 2352-byte sectors.
RAW_SECTOR = 0x930 # 2352
COOKED_SECTOR = 0x800 # 2048
BLOCK_SIZE = RAW_SECTOR * 16 # 0x9300
INDEX_ENTRY_SIZE = 32 # offset u32 + length u32 + 6 unused u32
SYNC_PATTERN = b"\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x00"
# SFO field formats (PSP PARAM.SFO).
FMT_U32 = 0x0004
FMT_STRING = 0x0204
SFO_DEFAULTS = {
"BOOTABLE": (FMT_U32, 1),
"CATEGORY": (FMT_STRING, "ME"),
"DISC_VERSION": (FMT_STRING, "1.00"),
"LICENSE": (FMT_STRING, "Library programs Copyright(C) Sony ComputerEntertainment Inc."),
"PARENTAL_LEVEL": (FMT_U32, 1),
"PSP_SYSTEM_VER": (FMT_STRING, "3.71"),
"REGION": (FMT_U32, 0x8000),
}
class ConversionError(Exception):
"""Raised when a disc cannot be converted."""
class Cancelled(Exception):
"""Raised internally when the caller requests cancellation."""
# --------------------------------------------------------------------------
# CUE sheet parsing
# --------------------------------------------------------------------------
@dataclass
class CueTrack:
number: int
mode: str # "MODE1/2048", "MODE1/2352", "MODE2/2352", "AUDIO"
file: str
file_type: str # "BINARY" | "WAVE" | ...
indexes: List[Tuple[int, int]] = field(default_factory=list) # (index, msf)
sector_size: int = 0
external_pregap: int = 0 # sectors from a PREGAP command (not in the file)
@property
def pregap_sectors(self) -> int:
"""Sectors between INDEX 00 and INDEX 01 (the in-file pregap).
This is informational only: when both INDEX 00 and INDEX 01 are
present in the same FILE, the pregap data is already inside the file
and must not be re-emitted. A ``PREGAP`` command, by contrast, is a
pregap that is *not* present in the file (see ``external_pregap``).
"""
idx = dict(self.indexes)
if 0 in idx and 1 in idx:
return max(0, idx[1] - idx[0])
return 0
@dataclass
class CueDisc:
tracks: List[CueTrack]
@property
def data_track(self) -> Optional[CueTrack]:
for t in self.tracks:
if t.mode != "AUDIO":
return t
return None
def _msf_to_sectors(msf: str) -> int:
m, s, f = (int(x) for x in msf.split(":"))
return (m * 60 + s) * 75 + f
def _sector_size_for_mode(mode: str) -> int:
if mode in ("MODE1/2048",):
return COOKED_SECTOR
if mode in ("MODE1/2352", "MODE2/2352", "AUDIO"):
return RAW_SECTOR
# CDG / MODE2/2336 etc. fall back to raw.
return RAW_SECTOR
def parse_cue(text: str) -> CueDisc:
"""Parse a CUE sheet into a :class:`CueDisc`.
Handles the common PS1 layouts: one or more ``FILE`` lines, ``TRACK``
lines (``MODE1/2048``, ``MODE1/2352``, ``MODE2/2352``, ``AUDIO``) and
``INDEX`` lines (``INDEX 00`` pregap + ``INDEX 01`` start).
"""
tracks: List[CueTrack] = []
current_file: Optional[str] = None
current_type = "BINARY"
current_track: Optional[CueTrack] = None
file_re = re.compile(r'^FILE\s+"([^"]+)"\s+(\w+)', re.I)
track_re = re.compile(r'^\s*TRACK\s+(\d{1,2})\s+(\S+)', re.I)
index_re = re.compile(r'^\s*INDEX\s+(\d{1,2})\s+(\d{1,2}:\d{2}:\d{2})', re.I)
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith(("REM", "CATALOG", "CDTEXTFILE", "PERFORMER", "TITLE", "ISRC", "FLAGS", "POSTGAP", "PREGAP")):
if line.startswith("PREGAP"):
# PREGAP <mm:ss:ff> is a pregap NOT stored in the track file;
# the writer must emit that many leading zero sectors.
m = re.search(r'PREGAP\s+(\d{1,2}:\d{2}:\d{2})', line, re.I)
if m and current_track is not None:
current_track.external_pregap = _msf_to_sectors(m.group(1))
continue
m = file_re.match(line)
if m:
current_file = m.group(1)
current_type = m.group(2).upper()
current_track = None
continue
m = track_re.match(line)
if m:
number = int(m.group(1))
mode = m.group(2).upper()
current_track = CueTrack(
number=number,
mode=mode,
file=current_file or "",
file_type=current_type,
sector_size=_sector_size_for_mode(mode),
)
tracks.append(current_track)
continue
m = index_re.match(line)
if m and current_track is not None:
current_track.indexes.append((int(m.group(1)), _msf_to_sectors(m.group(2))))
if not tracks:
raise ConversionError("CUE sheet contains no tracks")
for t in tracks:
if not t.file:
raise ConversionError(f"track {t.number} has no FILE")
return CueDisc(tracks)
# --------------------------------------------------------------------------
# Sector normalisation (cooked 2048 -> raw 2352 Mode 2 Form 1)
# --------------------------------------------------------------------------
def _to_bcd(value: int) -> int:
return ((value // 10) << 4) | (value % 10)
def _sectors_to_msf(sectors: int) -> Tuple[int, int, int]:
"""Convert a sector count to (minutes, seconds, frames) at 75 frames/s."""
total_seconds = sectors // 75
return total_seconds // 60, total_seconds % 60, sectors % 75
def _expand_cooked_sector(data: bytes, lba: int) -> bytes:
"""Expand a 2048-byte cooked sector to a 2352-byte raw Mode2 Form1 sector.
Rebuilds the 12-byte sync pattern, the 4-byte MSF header (BCD), the
8-byte subheader, and appends 280 bytes of EDC/ECC (zero-filled; the PSP
POPS loader does not verify EDC/ECC for converted games).
"""
assert len(data) == COOKED_SECTOR
frame = lba + 150 # 150-sector lead-in offset
msf = bytes([
_to_bcd((frame // 75 // 60) % 100),
_to_bcd((frame // 75) % 60),
_to_bcd(frame % 75),
0x02, # mode 2
])
# Subheader: file=2, channel=0, submode=0, coding=0; repeated once.
subheader = b"\x00\x00\x08\x00" * 2
return SYNC_PATTERN + msf + subheader + data + (b"\x00" * 280)
def iter_raw_sectors(
path: Path,
sector_size: int,
pregap_sectors: int = 0,
progress_cb=None,
cancel_event=None,
) -> Iterator[bytes]:
"""Yield 2352-byte raw sectors for one track file.
``sector_size`` is 2048 (cooked; expanded on the fly) or 2352 (raw).
``pregap_sectors`` leading zero sectors are emitted first.
"""
def _cancelled():
return cancel_event is not None and cancel_event.is_set()
if pregap_sectors:
for _ in range(pregap_sectors):
yield b"\x00" * RAW_SECTOR
size = path.stat().st_size
with open(path, "rb") as fh:
lba = 0
done = 0
while True:
if _cancelled():
raise Cancelled()
chunk = fh.read(sector_size)
if not chunk:
break
if len(chunk) != sector_size:
chunk = chunk.ljust(sector_size, b"\x00")
if sector_size == COOKED_SECTOR:
yield _expand_cooked_sector(chunk, lba)
else:
yield chunk
lba += 1
done += sector_size
if progress_cb:
progress_cb(done, size)
if progress_cb:
progress_cb(size, size)
# --------------------------------------------------------------------------
# PARAM.SFO writer
# --------------------------------------------------------------------------
def build_param_sfo(entries: List[Tuple[str, int, object]]) -> bytes:
"""Build a PSP PARAM.SFO from ``(key, fmt, value)`` entries.
``fmt`` is ``FMT_U32`` (uint32) or ``FMT_STRING`` (utf-8 string).
Layout matches the PSP format: header, index table, key table, data table.
"""
header_size = 20
index_size = len(entries) * 16
key_table_start = header_size + index_size
key_table = b"".join(key.encode("utf-8") + b"\x00" for key, _, _ in entries)
key_pad = (4 - (len(key_table) % 4)) % 4
data_table_start = key_table_start + len(key_table) + key_pad
data_table = b""
index_table = b""
key_offset = 0
data_offset = 0
for key, fmt, value in entries:
if fmt == FMT_U32:
raw = struct.pack("<I", int(value))
length = 4
max_len = 4
elif fmt == FMT_STRING:
s = str(value)
raw = s.encode("utf-8") + b"\x00"
length = len(s)
max_len = len(s) + 1
if len(raw) < max_len:
raw = raw.ljust(max_len, b"\x00")
else:
raise ConversionError(f"unsupported SFO format 0x{fmt:04x}")
index_table += struct.pack(
"<HHIII", key_offset, fmt, length, max_len, data_offset
)
data_table += raw
key_offset += len(key) + 1
data_offset += len(raw)
return (
b"\x00PSF"
+ struct.pack("<IIII", 0x00000101, key_table_start, data_table_start, len(entries))
+ index_table
+ key_table
+ (b"\x00" * key_pad)
+ data_table
)
def default_ps1_sfo(title: str, game_id: str) -> bytes:
"""A complete PARAM.SFO for a converted PS1 game."""
entries = []
for key, (fmt, value) in SFO_DEFAULTS.items():
entries.append((key, fmt, value))
entries.append(("DISC_ID", FMT_STRING, game_id))
entries.append(("TITLE", FMT_STRING, title))
return build_param_sfo(entries)
def extract_sfo(base_pbp: Path) -> bytes:
"""Return the PARAM.SFO section of a base PBP."""
header = read_pbp_header(base_pbp)
idx = SECTION_OFFSET["sfo"] // 4
return read_pbp_section(base_pbp, header, idx)
def extract_icon0(base_pbp: Path) -> bytes:
"""Return the ICON0.PNG section of a base PBP."""
header = read_pbp_header(base_pbp)
idx = SECTION_OFFSET["icon0"] // 4
return read_pbp_section(base_pbp, header, idx)
def _sfo_entry(base_sfo: bytes, key: str) -> Optional[bytes]:
"""Return the raw data of an SFO entry, or ``None`` if absent."""
if len(base_sfo) < 20 or base_sfo[:4] != b"\x00PSF":
return None
_, _, key_table_off, data_table_off, nitems = struct.unpack("<IIIII", base_sfo[:20])
for i in range(nitems):
entry_off = 20 + i * 16
koff, _, _, max_len, doff = struct.unpack("<HHIII", base_sfo[entry_off:entry_off + 16])
kend = base_sfo.find(b"\x00", key_table_off + koff)
k = bytes(base_sfo[key_table_off + koff:kend]).decode("ascii", "replace")
if k == key:
return base_sfo[data_table_off + doff: data_table_off + doff + max_len]
return None
def _warn_old_base(base_pbp: Path) -> None:
"""Warn when the base PBP carries a very old POPS loader.
PS1 classics from the 3.xx firmware era ship a POPS loader that often
fails to link on 6.60/6.61 CFW (the PSP reports error 80020148), so
newer base PBPs are strongly preferred.
"""
try:
raw = _sfo_entry(extract_sfo(base_pbp), "PSP_SYSTEM_VER")
if not raw:
return
version = raw.rstrip(b"\x00").decode("ascii", "replace")
major = version.split(".")[0]
if major.isdigit() and int(major) < 6:
import sys
print(
f"warning: base PBP is a {version}-era PS1 classic; its POPS loader "
"may fail to load on 6.60/6.61 CFW (error 80020148). "
"Prefer a newer base.pbp.",
file=sys.stderr,
)
except Exception:
pass
def patch_sfo(base_sfo: bytes, title: str, game_id: str) -> bytes:
"""Patch ``TITLE`` and ``DISC_ID`` in a base PARAM.SFO in place.
The reference writer reuses the base PBP's SFO verbatim (preserving the
original field types, sizes and values) and only rewrites the title and
game id. This keeps the output byte-compatible with official PSN classics
(e.g. integer fields stay ``0x0404`` rather than ``0x0004``).
"""
if len(base_sfo) < 20 or base_sfo[:4] != b"\x00PSF":
raise ConversionError("base PBP has no valid PARAM.SFO")
_, _, key_table_off, data_table_off, nitems = struct.unpack("<IIIII", base_sfo[:20])
sfo = bytearray(base_sfo)
patches = {"TITLE": title.encode("utf-8"), "DISC_ID": game_id.encode("ascii")}
for i in range(nitems):
entry_off = 20 + i * 16
koff, _, _, max_len, doff = struct.unpack("<HHIII", sfo[entry_off:entry_off + 16])
key_end = sfo.find(b"\x00", key_table_off + koff)
key = bytes(sfo[key_table_off + koff:key_end]).decode("ascii", "replace")
if key not in patches:
continue
value = patches[key]
# strncpy semantics: value + zero padding, capped at the entry's size.
raw = value[:max_len].ljust(max_len, b"\x00")
sfo[data_table_off + doff: data_table_off + doff + max_len] = raw
# SFO string length includes the NUL terminator (matches the base SFO
# and the reference SetSFOTitle implementation).
new_len = min(len(value) + 1, max_len)
struct.pack_into("<I", sfo, entry_off + 4, new_len)
return bytes(sfo)
# --------------------------------------------------------------------------
# Minimal PNG generator (no external image deps)
# --------------------------------------------------------------------------
def _png(width: int, height: int, rgb: Tuple[int, int, int]) -> bytes:
def chunk(typ: bytes, data: bytes) -> bytes:
payload = typ + data
return struct.pack(">I", len(data)) + payload + struct.pack(">I", zlib.crc32(payload) & 0xFFFFFFFF)
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8-bit RGB
raw = b"".join(b"\x00" + bytes(rgb) * width for _ in range(height))
return (
b"\x89PNG\r\n\x1a\n"
+ chunk(b"IHDR", ihdr)
+ chunk(b"IDAT", zlib.compress(raw, 9))
+ chunk(b"IEND", b"")
)
def default_icon0_png() -> bytes:
return _png(144, 80, (24, 24, 24))
def default_pic1_png() -> bytes:
return _png(480, 272, (10, 10, 16))
# --------------------------------------------------------------------------
# PBP reading (used both for base.pbp extraction and self-test validation)
# --------------------------------------------------------------------------
def read_pbp_header(path: Path) -> List[int]:
with open(path, "rb") as fh:
data = fh.read(PBP_HEADER_SIZE)
if len(data) < PBP_HEADER_SIZE or data[:4] != PBP_MAGIC:
raise ConversionError(f"{path} is not a PBP file (bad magic)")
return list(struct.unpack("<10I", data))
def read_pbp_section(path: Path, header: List[int], index: int) -> bytes:
start = header[index]
end = header[index + 1] if index + 1 < len(header) else None
with open(path, "rb") as fh:
fh.seek(start)
data = fh.read() if end is None else fh.read(end - start)
return data
def extract_data_psp(base_pbp: Path) -> bytes:
"""Extract the encrypted POPS loader (DATA.PSP) from a base PBP.
The reference writer reads the ``~PSP`` header's total-size field
(u32 at DATA.PSP + 0x2C, the 12th u32) and copies exactly that many
bytes. The section in the source PBP is padded with zeros up to the PSAR
offset; copying that padding would bloat the DATA.PSP section for no
reason, so we copy only ``prx_size`` bytes, matching popstation.
"""
header = read_pbp_header(base_pbp)
idx = SECTION_OFFSET["data_psp"] // 4 # header array index (2 + section index)
section = read_pbp_section(base_pbp, header, idx)
if len(section) < 0x30:
raise ConversionError(f"{base_pbp} has no DATA.PSP section")
if section[:4] != b"~PSP":
raise ConversionError(f"{base_pbp} DATA.PSP has no ~PSP header")
(prx_size,) = struct.unpack("<I", section[0x2C:0x30])
if prx_size == 0 or prx_size > len(section):
prx_size = len(section)
return section[:prx_size]
def extract_startdat(base_pbp: Path) -> bytes:
"""Extract the trailing STARTDAT block (boot.png + PGD) from a base PBP.
STARTDAT sits immediately after the disc data: at ``PSAR + p1`` where
``p1`` is the u32 at ``PSAR + 0x0C``. (The original popstation computes
``p1 + 0x50000`` as an absolute offset, which only works for bases whose
PSAR happens to be aligned to 0x50000; using ``PSAR + p1`` works for any
PS1 classic.)
"""
header = read_pbp_header(base_pbp)
psar_offset = header[SECTION_OFFSET["psar"] // 4] # header array index
with open(base_pbp, "rb") as fh:
fh.seek(psar_offset + 0x0C)
(x,) = struct.unpack("<I", fh.read(4))
marker_off = psar_offset + x
fh.seek(marker_off)
if fh.read(8) != b"STARTDAT":
raise ConversionError(f"cannot find STARTDAT in {base_pbp} (not a valid PS1 base PBP?)")
fh.seek(marker_off)
return fh.read()
raise ConversionError("unreachable")
# --------------------------------------------------------------------------
# PBP writer
# --------------------------------------------------------------------------
def _format_psar_game_id(game_id: str) -> str:
"""Format a PS1 serial for the PSAR header field.
The PBP's PARAM.SFO uses the plain 9-char id (``SLPS01762``), while the
PSAR header stores the region-delineated form (``_SLPS_01762``).
"""
m = re.match(r"^([A-Za-z]{4})[_-]?(\d{5})$", game_id)
if m:
return f"_{m.group(1).upper()}_{m.group(2)}"
return game_id
def _pad_to(out, target: int) -> None:
"""Write zero bytes until the stream reaches ``target`` (absolute offset)."""
delta = target - out.tell()
if delta < 0:
raise ConversionError(f"internal error: overshot offset by {-delta} bytes")
if delta:
out.write(b"\x00" * delta)
# Track type values used in the PS1 TOC (PSAR +0x800).
_TOC_DATA = 0x41
_TOC_AUDIO = 0x01
_TOC_A0 = 0xA0
_TOC_A1 = 0xA1
_TOC_A2 = 0xA2
_TOC_ENTRY = 10 # bytes per TOC entry
_TOC_LEAD_IN = 150 # 2-second lead-in, in sectors
def build_toc(disc: "DiscImage") -> bytes:
"""Generate the PS1 table-of-contents stored at PSAR +0x800.
Emits the standard CD-style TOC: an A0 entry (first track), an A1 entry
(last track), an A2 entry (lead-out), then one 10-byte entry per track.
Track addresses are BCD-encoded MSF; a 150-sector (2 s) lead-in is added
to the relative address field, matching the reference implementation.
"""
tracks = disc.tracks_info()
total = disc.total_sectors()
entries = [bytearray(_TOC_ENTRY) for _ in range(len(tracks) + 3)]
first_num, _, _ = tracks[0]
last_num, _, _ = tracks[-1]
# A-point entries use control byte 0x41 (data) regardless of track type,
# matching popstation and official PSN classics. A0's byte 8 is the raw
# Red Book disc-type bitfield 0x20 (CD-ROM XA), not a BCD value.
e = entries[0]
e[0] = _TOC_DATA
e[2] = _TOC_A0
e[7] = _to_bcd(first_num)
e[8] = 0x20
e = entries[1]
e[0] = _TOC_DATA
e[2] = _TOC_A1
e[7] = _to_bcd(last_num)
e = entries[2]
e[0] = _TOC_DATA
e[2] = _TOC_A2
mm, ss, ff = _sectors_to_msf(total)
e[7], e[8], e[9] = _to_bcd(mm), _to_bcd(ss), _to_bcd(ff)
for i, (num, is_data, start) in enumerate(tracks):
e = entries[3 + i]
e[0] = _TOC_DATA if is_data else _TOC_AUDIO
e[2] = _to_bcd(num)
mm, ss, ff = _sectors_to_msf(start)
e[3], e[4], e[5] = _to_bcd(mm), _to_bcd(ss), _to_bcd(ff)
mm, ss, ff = _sectors_to_msf(start + _TOC_LEAD_IN)
e[7], e[8], e[9] = _to_bcd(mm), _to_bcd(ss), _to_bcd(ff)
return b"".join(bytes(e) for e in entries)
def _build_psar_header(disc: "DiscImage", game_id: str, title: str) -> bytes:
"""Build the fixed PSAR header region 0x400..0x4000, byte-identical to the
popstation level-0 writer.
Layout (offsets relative to the PSAR start):
* 0x400 data1 (0xE20 bytes): game id at +0, TOC at +0x400, 0x10 at +0x7FE
* 0x1220 p2 offset (u32) = padded ISO size + 0x100000 + 0x2D31
* 0x1224 data2 (0x2DDC bytes): FF 07 at +4, title at +8, 0x03 at +0x88
"""
data1_size = 0xE20
data2_size = 0x2DDC
p2_delta = 0x2D31
block = bytearray(data1_size + 4 + data2_size) # 0x3C00 = 0x400..0x4000
gid = _format_psar_game_id(game_id).encode("ascii")[:11]
block[: len(gid)] = gid
toc = build_toc(disc)
block[0x400: 0x400 + len(toc)] = toc
block[0x7FE] = 0x10
p2 = disc.padded_size() + PSAR_ISO_OFFSET + p2_delta
block[data1_size: data1_size + 4] = struct.pack("<I", p2)
data2 = data1_size + 4
block[data2 + 4: data2 + 6] = b"\xFF\x07"
title_bytes = title.encode("ascii", "replace")[:0x80]
block[data2 + 8: data2 + 8 + len(title_bytes)] = title_bytes
block[data2 + 0x88] = 0x03
return bytes(block)
def _write_disc_block(
out: io.BufferedWriter,
block: bytes,
complevel: int,
data_offset: int,
indexes: bytearray,
) -> int:
"""Compress and write one 0x9300 disc block, appending its index entry.
Blocks are deflated with raw DEFLATE (the zlib header and Adler-32
trailer stripped). If a block does not shrink, the raw bytes are stored
instead. The index entry follows pop-fe: offset (u32), length (u16),
flags (u8), then the first 16 bytes of the block's SHA-1.
"""
assert len(block) == BLOCK_SIZE
compressed = zlib.compress(block, complevel)[2:-4] if complevel > 0 else block
entry = bytearray(INDEX_ENTRY_SIZE)
struct.pack_into("<I", entry, 0, data_offset)
if complevel == 0:
entry[6] = 0x01 # uncompressed block
entry[8:24] = hashlib.sha1(block).digest()[:16]
if len(compressed) >= BLOCK_SIZE:
struct.pack_into("<H", entry, 4, BLOCK_SIZE)
out.write(block)
data_offset += BLOCK_SIZE
else:
struct.pack_into("<H", entry, 4, len(compressed))
out.write(compressed)
data_offset += len(compressed)
indexes.extend(entry)
return data_offset
def _write_psar(
out: io.BufferedWriter,
disc: "DiscImage",
game_id: str,
title: str,
complevel: int = 1,
progress_cb=None,
cancel_event=None,
) -> None:
"""Write the DATA.PSAR section (single-disc).
Disc data is split into 0x9300-byte blocks and deflated (``complevel`` >
0). The PSAR header's p1/p2 fields and the index table are patched after
the compressed data is written, since the compressed size is not known
up front.
All offsets are relative to the start of the PSAR section.
"""
base = out.tell()
iso_size = disc.padded_size()
block_count = iso_size // BLOCK_SIZE
# 0x000: PSAR signature; 0x00C: p1 (disc data end, patched below).
out.write(PSAR_MAGIC)
out.write(struct.pack("<I", iso_size + PSAR_ISO_OFFSET))
_pad_to(out, base + PSAR_GAMEID_OFFSET)
# 0x400..0x4000: game id / TOC / title (p2 at 0x1220 is patched below).
out.write(_build_psar_header(disc, game_id, title))
_pad_to(out, base + PSAR_INDEX_OFFSET)
# Reserve the index table; filled in once the disc data is compressed.
for _ in range(block_count):
out.write(b"\x00" * INDEX_ENTRY_SIZE)
_pad_to(out, base + PSAR_ISO_OFFSET)
# Compress + write disc data, one 0x9300 block at a time.
indexes = bytearray()
data_offset = 0 # relative to PSAR_ISO_OFFSET
buf = bytearray()
for sector in disc.iter_raw(progress_cb=progress_cb, cancel_event=cancel_event):
buf.extend(sector)
if len(buf) < BLOCK_SIZE:
continue
data_offset = _write_disc_block(out, buf, complevel, data_offset, indexes)
buf = bytearray()
if buf:
data_offset = _write_disc_block(out, buf.ljust(BLOCK_SIZE, b"\x00"), complevel, data_offset, indexes)
# Align the end of the disc data to 0x10 before STARTDAT (which the
# caller appends), then patch p1/p2 and the index table.
_pad_to(out, (out.tell() + 0xF) & ~0xF)
end_offset = out.tell() - base
end = out.tell()
out.seek(base + 0x0C)
out.write(struct.pack("<I", end_offset))
out.seek(base + 0x1220)
out.write(struct.pack("<I", end_offset + 0x2D31))
out.seek(base + PSAR_INDEX_OFFSET)
out.write(bytes(indexes))
out.seek(end)
class DiscImage:
"""A normalised view of a PS1 disc as a stream of raw 2352-byte sectors."""
def __init__(self, cue: Optional[CueDisc], path: Path):
self.cue = cue
self.path = path
def total_sectors(self) -> int:
if self.cue is None:
size = self.path.stat().st_size
if size % RAW_SECTOR == 0:
return size // RAW_SECTOR
return (size + COOKED_SECTOR - 1) // COOKED_SECTOR
total = 0
base_dir = self.path.parent
for track in self.cue.tracks:
fpath = base_dir / track.file
if fpath.is_file():
total += track.external_pregap + (fpath.stat().st_size // track.sector_size)
return total
def padded_size(self) -> int:
size = self.total_sectors() * RAW_SECTOR
if size % BLOCK_SIZE:
size += BLOCK_SIZE - (size % BLOCK_SIZE)
return size
def tracks_info(self) -> List[Tuple[int, bool, int]]:
"""Return ``(track_number, is_data, absolute_start_sector)`` per track.
``absolute_start_sector`` is the disc-wide sector of each track's
INDEX 01 (i.e. accounting for every preceding track's pregap + data).
A cue-less single image is treated as one data track starting at 0.
"""
if self.cue is None:
return [(1, True, 0)]
base_dir = self.path.parent
out: List[Tuple[int, bool, int]] = []
cursor = 0
for track in self.cue.tracks:
idx01 = dict(track.indexes).get(1, 0)
out.append((track.number, track.mode != "AUDIO", cursor + track.external_pregap + idx01))
fpath = base_dir / track.file
data_sectors = fpath.stat().st_size // track.sector_size if fpath.is_file() else 0
cursor += track.external_pregap + data_sectors
return out
def iter_raw(self, progress_cb=None, cancel_event=None) -> Iterator[bytes]:
if self.cue is None:
yield from iter_raw_sectors(
self.path,
RAW_SECTOR if self.path.stat().st_size % RAW_SECTOR == 0 else COOKED_SECTOR,
progress_cb=progress_cb,
cancel_event=cancel_event,
)
return
base_dir = self.path.parent
for track in self.cue.tracks:
fpath = base_dir / track.file
if not fpath.is_file():
raise ConversionError(f"track file missing: {fpath}")
yield from iter_raw_sectors(
fpath,
track.sector_size,
pregap_sectors=track.external_pregap,
progress_cb=progress_cb,
cancel_event=cancel_event,
)
def _aligned_up(value: int, alignment: int) -> int:
return value + ((alignment - (value % alignment)) % alignment)
def convert(
source: Path,
title: str,
game_id: str,
base_pbp: Path,
out_path: Path,
progress_cb=None,
cancel_event=None,
complevel: int = 1,
) -> Path:
"""Convert a PS1 disc image to ``out_path`` (an ``EBOOT.PBP``).
``source`` may be a ``.cue`` (multi-track supported) or a single
``.iso``/``.img``/``.bin``. ``base_pbp`` must be a PSN PS1 classic EBOOT.
``complevel`` is the DEFLATE level for the disc data (1-9; 0 = raw, which
does not boot on real hardware and is only useful for diagnostics).
"""
if not base_pbp.is_file():
raise ConversionError(f"base.pbp not found: {base_pbp}")
if not source.is_file():
raise ConversionError(f"disc image not found: {source}")
game_id = game_id.strip()
if not game_id:
raise ConversionError("a game id (e.g. SLUS00594) is required")
if source.suffix.lower() == ".cue":
disc: DiscImage = DiscImage(parse_cue(source.read_text(encoding="utf-8", errors="replace")), source)
else:
disc = DiscImage(None, source)
# Resources required from the base PBP.
data_psp = extract_data_psp(base_pbp)
startdat = extract_startdat(base_pbp)
_warn_old_base(base_pbp)
# Reuse the base PBP's SFO, patched with our title + game id (the
# popstation approach), rather than rebuilding it from scratch.
sfo = patch_sfo(extract_sfo(base_pbp), title, game_id)
# Reuse the base PBP's ICON0 too. The XMB icon parser is picky about PNG
# variants, and the base icon is guaranteed to decode; the other image
# sections are left empty exactly like popstation when no assets are
# supplied alongside the conversion.
icon0 = extract_icon0(base_pbp)
# Sections in header order; empty sections keep the previous offset.
sections = {
"sfo": sfo,
"icon0": icon0,
"icon1": b"",
"pic0": b"",
"pic1": b"",
"snd0": b"",
"data_psp": data_psp,
# psar written separately, sized after everything else.
"psar": None,
}
header = [0] * PBP_HEADER_U32
header[0] = struct.unpack("<I", PBP_MAGIC)[0]
header[1] = 0x00010000
current = PBP_HEADER_SIZE
for i, key in enumerate(SECTION_ORDER):
header[2 + i] = current
if key != "psar":
current += len(sections[key])
# DATA.PSAR must start on a 0x10000 boundary. It is the final section, so
# its length is implicit (everything up to STARTDAT/EOF).
psar_offset = _aligned_up(header[2 + SECTION_ORDER.index("psar")], PSAR_ALIGNMENT)
header[2 + SECTION_ORDER.index("psar")] = psar_offset
out_path.parent.mkdir(parents=True, exist_ok=True)
tmp = out_path.with_suffix(out_path.suffix + ".tmp")
try:
with open(tmp, "wb") as fh:
fh.write(struct.pack("<10I", *header))
for key in SECTION_ORDER:
if key == "psar":
break
if sections[key]:
fh.write(sections[key])
# Pad up to the aligned PSAR offset.
while fh.tell() < psar_offset:
fh.write(b"\x00")
_write_psar(fh, disc, game_id, title, complevel=complevel,
progress_cb=progress_cb, cancel_event=cancel_event)
# STARTDAT goes at the very end.
fh.write(startdat)
os.replace(tmp, out_path)
except BaseException:
try:
tmp.unlink(missing_ok=True)
except OSError:
pass
raise
return out_path
def convert_to_folder(
source: Path,
title: str,
game_id: str,
base_pbp: Path,
out_dir: Path,
progress_cb=None,
cancel_event=None,
complevel: int = 1,
) -> Path:
"""Convert a PS1 disc into a game folder ready for /PSP/GAME.
Writes ``<out_dir>/<GAME_ID>/EBOOT.PBP`` — the same single-file output the
popstation tool produces. No KEYS.BIN or DOCUMENT.DAT is written:
popstation doesn't emit them and CFW (ARK-4 POPS) doesn't need them.
"""
folder_name = re.sub(r"[^A-Za-z0-9]", "", game_id).upper() or "PS1GAME"
folder = Path(out_dir) / folder_name
folder.mkdir(parents=True, exist_ok=True)
convert(source, title, game_id, base_pbp, folder / "EBOOT.PBP",
progress_cb=progress_cb, cancel_event=cancel_event, complevel=complevel)
return folder
# --------------------------------------------------------------------------
# Self-test / CLI
# --------------------------------------------------------------------------
def _self_test() -> None:
"""Validate the parts that do not need hardware: SFO round-trip, PNG
generation, PBP header round-trip, and cue parsing."""
import tempfile
# CUE parsing.
cue = parse_cue(
'FILE "Game (Track 1).bin" BINARY\n'
" TRACK 01 MODE2/2352\n"
" INDEX 01 00:00:00\n"
" TRACK 02 AUDIO\n"
" INDEX 00 00:02:00\n"
" INDEX 01 00:04:00\n"
'FILE "Game (Track 2).bin" BINARY\n'
" TRACK 03 AUDIO\n"
" INDEX 01 00:00:00\n"
)
assert len(cue.tracks) == 3, "expected 3 tracks"
assert cue.tracks[0].mode == "MODE2/2352"
assert cue.tracks[1].pregap_sectors == 150, "expected 2s (150 sector) pregap"
assert cue.data_track is cue.tracks[0]
# PREGAP command is a pregap NOT stored in the file.
cue_pg = parse_cue(
'FILE "t.bin" BINARY\n TRACK 01 AUDIO\n PREGAP 00:02:00\n INDEX 01 00:00:00\n'
)
assert cue_pg.tracks[0].external_pregap == 150, cue_pg.tracks[0].external_pregap
assert cue_pg.tracks[0].pregap_sectors == 0
# TOC generation: 1 data (10 sectors) + 1 audio track (20 sectors + 2s pregap).
with tempfile.TemporaryDirectory() as td:
d = Path(td)
(d / "t1.bin").write_bytes(b"\x00" * (10 * RAW_SECTOR))
(d / "t2.bin").write_bytes(b"\x00" * (20 * RAW_SECTOR))
cue2 = parse_cue(
'FILE "t1.bin" BINARY\n TRACK 01 MODE2/2352\n INDEX 01 00:00:00\n'
'FILE "t2.bin" BINARY\n TRACK 02 AUDIO\n INDEX 00 00:00:00\n INDEX 01 00:02:00\n'
)
disc2 = DiscImage(cue2, d / "t.cue")
# INDEX 00 pregap is inside the file, so the disc is just the files
# concatenated: 10 + 20 sectors (not 10 + 150 + 20).
assert disc2.total_sectors() == 30, disc2.total_sectors()
toc = build_toc(disc2)
assert len(toc) == (2 + 3) * 10, len(toc)
assert toc[0] == 0x41 and toc[2] == 0xA0 and toc[7] == 0x01 # A0: first track
assert toc[8] == 0x20 # A0: raw Red Book disc-type byte, not BCD
assert toc[10] == 0x41 and toc[12] == 0xA1 and toc[17] == 0x02 # A1: last track
assert toc[20] == 0x41 and toc[22] == 0xA2 # A2: lead-out (control 0x41)
assert toc[30] == 0x41 and toc[32] == 0x01 # track 1: data, number 1
assert toc[40] == 0x01 and toc[42] == 0x02 # track 2: audio, number 2
# track 2 relative address = start(160) + 150 lead-in = 310 = 00:04:10
assert toc[47] == 0x00 and toc[48] == 0x04 and toc[49] == 0x10
# SFO round-trip through the project's own parser.
from . import iso_reader # local import to keep the module importable standalone
sfo = default_ps1_sfo("Final Fantasy VII", "SCUS94163")
parsed = iso_reader.parse_sfo(sfo)
assert parsed.get("TITLE") == "Final Fantasy VII", parsed
assert parsed.get("DISC_ID") == "SCUS94163", parsed
assert parsed.get("CATEGORY") == "ME", parsed
assert parsed.get("BOOTABLE") == 1, parsed
# Base-SFO patching preserves everything but TITLE + DISC_ID.
patched = patch_sfo(default_ps1_sfo("Hot Shots Golf", "SCUS94476"), "Pepsiman", "SLPS01762")
parsed2 = iso_reader.parse_sfo(patched)
assert parsed2.get("TITLE") == "Pepsiman", parsed2
assert parsed2.get("DISC_ID") == "SLPS01762", parsed2
assert parsed2.get("BOOTABLE") == 1, parsed2
# PNG generation produces valid PNG signatures + IHDR dimensions.
icon = default_icon0_png()
assert icon[:8] == b"\x89PNG\r\n\x1a\n"
assert struct.unpack(">II", icon[16:24]) == (144, 80)
pic = default_pic1_png()
assert struct.unpack(">II", pic[16:24]) == (480, 272)
# PBP header round-trip: write a synthetic PBP and read it back.
with tempfile.TemporaryDirectory() as td:
p = Path(td) / "test.pbp"
header = [0] * 10
header[0] = struct.unpack("<I", PBP_MAGIC)[0]
header[1] = 0x00010000
header[2] = 0x28
header[3] = 0x28 + len(sfo)
with open(p, "wb") as fh:
fh.write(struct.pack("<10I", *header))
fh.write(sfo)
assert read_pbp_header(p)[0] == header[0]
assert read_pbp_section(p, header, 2) == sfo
# PSAR compression round-trip (level 1): the disc data must survive
# raw-DEFLATE and the index table must be readable with the pop-fe
# layout (offset u32, length u16, flags u8, SHA-1[16]).
with tempfile.TemporaryDirectory() as td:
d = Path(td)
src = d / "cd.bin"
src.write_bytes(bytes(range(256)) * (BLOCK_SIZE // 256 * 3))
disc = DiscImage(None, src)
buf = io.BytesIO()
_write_psar(buf, disc, "SLUS00594", "Round Trip", complevel=1)
raw = buf.getvalue()
assert raw[:12] == PSAR_MAGIC
rebuilt = bytearray()
i = 0
while True:
e = raw[PSAR_INDEX_OFFSET + i * 32: PSAR_INDEX_OFFSET + (i + 1) * 32]
off, = struct.unpack("<I", e[0:4])
ln, = struct.unpack("<H", e[4:6])
if ln == 0:
break
data = raw[PSAR_ISO_OFFSET + off: PSAR_ISO_OFFSET + off + ln]
rebuilt += zlib.decompress(data, wbits=-15) if ln < BLOCK_SIZE else data
i += 1
assert i == 3, i
assert len(rebuilt) == disc.padded_size(), len(rebuilt)
assert bytes(rebuilt[: src.stat().st_size]) == src.read_bytes()
# first index entry carries the SHA-1 of the first raw block
assert raw[PSAR_INDEX_OFFSET + 8: PSAR_INDEX_OFFSET + 24] == hashlib.sha1(bytes(rebuilt[:BLOCK_SIZE])).digest()[:16]
print("ps1_convert self-test OK")
def main(argv: Optional[List[str]] = None) -> int:
ap = argparse.ArgumentParser(description="Convert a PS1 disc image to a PSP EBOOT.PBP")
ap.add_argument("--source", help="path to .cue or a single .iso/.img/.bin")
ap.add_argument("--title", help="game title shown on the PSP")
ap.add_argument("--game-id", help="PS1 game id, e.g. SLUS00594")
ap.add_argument("--base-pbp", help="path to a PSN PS1 classic EBOOT.PBP (required)")
ap.add_argument("--out", help="output EBOOT.PBP path (single file)")
ap.add_argument("--out-dir", help="output folder: writes <dir>/<GAME_ID>/EBOOT.PBP")
ap.add_argument("--complevel", type=int, default=1, choices=range(0, 10),
help="DEFLATE compression level 0-9 for the disc data (default 1)")
ap.add_argument("--test", action="store_true", help="run self-test and exit")
args = ap.parse_args(argv)
if args.test:
_self_test()
return 0
if not all([args.source, args.title, args.game_id, args.base_pbp]) or (not args.out and not args.out_dir):
ap.error("--source --title --game-id --base-pbp and --out (or --out-dir) are all required (or use --test)")
try:
if args.out_dir:
folder = convert_to_folder(
Path(args.source), args.title, args.game_id,
Path(args.base_pbp), Path(args.out_dir), complevel=args.complevel,
)
print(f"wrote {folder}")
else:
convert(
Path(args.source), args.title, args.game_id,
Path(args.base_pbp), Path(args.out), complevel=args.complevel,
)
print(f"wrote {args.out}")
return 0
except ConversionError as exc:
print(f"error: {exc}")
return 1
if __name__ == "__main__":
raise SystemExit(main())