#!/usr/bin/env python3 """ CBS Cruncher command-line edition. Dependency-free Python port of: * CBS Cruncher v3 (original 1996 greedy encoder) * CBS Cruncher v4.3 (cost-aware experimental encoder) * CBS PCK decruncher and in-place RAM safety analysis The generated PCK files use the same byte format as the browser edition and the Z80 decrunchers. """ from __future__ import annotations import argparse import os import sys import time from array import array from concurrent.futures import ProcessPoolExecutor, as_completed from dataclasses import dataclass, field from functools import lru_cache from pathlib import Path from typing import Callable, Dict, List, Optional, Sequence, Tuple VERSION = "4.3" MAX_FILE_SIZE = 0xFF00 MAX_PAYLOAD_SIZE = MAX_FILE_SIZE - 4 FIRST_LONG_DISTANCE = 0x4000 CONTROL_STATES = 17 MATCH_SEARCH_DEPTH = 16384 SCALE_SETTINGS: Tuple[Tuple[int, int], ...] = ( (5, 4), (4, 8), (3, 16), (2, 32), (1, 64), ) ProgressCallback = Callable[[Dict[str, int]], None] class CbsError(Exception): """A readable CBS format or compression error.""" @dataclass(frozen=True) class Safety: safe: bool required_gap: int available_gap: int @dataclass(frozen=True) class Trial: threshold: int size: int safe: bool required_gap: int available_gap: int @dataclass(frozen=True) class ScaleTrial: distance_high_bits: int distance_log: int maximum_distance: int output_size: int token_count: int control_words: int verified: bool in_place_safe: bool @dataclass class CompressionResult: output: bytes input_size: int payload_size: int output_size: int token_count: int threshold: int verified: bool in_place_safe: bool required_gap: int available_gap: int trials: List[Trial] = field(default_factory=list) engine: str = "v3" literal_bytes: int = 0 match_bytes: int = 0 control_words: int = 0 search_depth: int = 0 distance_high_bits: int = 0 distance_log: int = 0 scale_trials: List[ScaleTrial] = field(default_factory=list) class BitWriter: def __init__(self) -> None: self.out = bytearray((0, 0)) self.pos = 0 self.word = 0 self.count = 0 def bit(self, value: int) -> None: if self.count == 16: self._next() self.word = ((self.word << 1) | (value & 1)) & 0xFFFF self.count += 1 def lsb(self, value: int, count: int) -> None: for bit_index in range(count): self.bit(value >> bit_index) def msb(self, value: int, count: int) -> None: for bit_index in range(count - 1, -1, -1): self.bit(value >> bit_index) def byte(self, value: int) -> None: self.out.append(value & 0xFF) def word_le(self, value: int) -> None: self.byte(value) self.byte(value >> 8) def finish(self) -> bytes: self.word = (self.word << (16 - self.count)) & 0xFFFF self._flush() return bytes(self.out) def _flush(self) -> None: self.out[self.pos] = self.word & 0xFF self.out[self.pos + 1] = (self.word >> 8) & 0xFF def _next(self) -> None: self._flush() self.pos = len(self.out) self.out.extend((0, 0)) self.word = 0 self.count = 0 class BitReader: def __init__(self, data: bytes) -> None: self.data = data self.pos = 0 self.word = 0 self.left = 0 def bit(self) -> int: if not self.left: self._need(2) self.word = self.data[self.pos] | (self.data[self.pos + 1] << 8) self.pos += 2 self.left = 16 value = self.word >> 15 self.word = (self.word << 1) & 0xFFFF self.left -= 1 return value def msb(self, count: int) -> int: value = 0 for _ in range(count): value = (value << 1) | self.bit() return value def byte(self) -> int: self._need(1) value = self.data[self.pos] self.pos += 1 return value def word_le(self) -> int: return self.byte() | (self.byte() << 8) @property def consumed(self) -> int: return self.pos def _need(self, count: int) -> None: if self.pos + count > len(self.data): raise CbsError("Unexpected end of compressed stream.") def _settings(payload_size: int) -> Tuple[int, int]: if payload_size >= 0x2000: return 5, 4 if payload_size >= 0x1000: return 4, 8 if payload_size >= 0x0800: return 3, 16 if payload_size >= 0x0400: return 2, 32 return 1, 64 def _maximum_distance(x: int) -> int: if x == 5: return 0xFFFF return 512 + ((1 << (x + 1)) - 1) * 256 + 255 def _build_links(source: bytes, length: int) -> array: last = array("i", [-1]) * 65536 previous = array("i", [-1]) * length for position in range(length): pair = (source[position] << 8) | source[position + 1] previous[position] = last[pair] last[pair] = position return previous def _long_length(writer: BitWriter, length: int, raw: bool) -> None: if length < 127: writer.byte((length << 1) | (0 if raw else 1)) else: writer.byte(254 if raw else 255) writer.word_le(length) def _write_distance(writer: BitWriter, distance: int, x: int) -> None: value = distance - 1 if value < 32: writer.lsb(1, 2) writer.msb(value, 5) return if value < 544: value -= 32 writer.bit(0) writer.msb(value >> 8, 1) writer.byte(value) return if x < 5 or value < 0x3FFF: value -= 511 writer.lsb(3, 2) writer.msb(value >> 8, x + 1) writer.byte(value) return writer.lsb(3, 2) writer.msb(62, 6) writer.word_le(distance) def _write_match(writer: BitWriter, length: int, distance: int, x: int) -> None: writer.bit(0) if length < 4: writer.bit(0) writer.bit(length) elif length < 8: writer.lsb(3, 2) writer.msb(length, 2) elif length < 24: writer.lsb(1, 3) writer.msb(length - 8, 4) else: writer.lsb(5, 3) _long_length(writer, length, False) _write_distance(writer, distance, x) def _assemble( original: bytes, payload_size: int, stream: bytes, token_count: int, distance_log: int, ) -> bytes: output = bytearray(len(stream) + 7) output[0:4] = original[payload_size:] output[4] = ((token_count & 0xFF) + 1) & 0xFF output[5] = (((token_count >> 8) & 0xFF) + 1) & 0xFF output[6] = distance_log output[7:] = stream return bytes(output) def _decode_cbs(packed: bytes, original_size: Optional[int] = None) -> Tuple[bytes, Safety]: if len(packed) < 9: raise CbsError("Not a valid CBS PCK stream.") token_count = (((packed[5] - 1) & 0xFF) << 8) | ((packed[4] - 1) & 0xFF) scale_to_x = {64: 1, 32: 2, 16: 3, 8: 4, 4: 5} x = scale_to_x.get(packed[6]) if x is None: raise CbsError("Unknown CBS distance scale.") reader = BitReader(packed[7:]) output = bytearray() required_gap = 0 def room(length: int) -> None: if len(output) + length > MAX_PAYLOAD_SIZE: raise CbsError( "Decompressed payload exceeds the original " f"{MAX_FILE_SIZE:,}-byte file limit." ) for _ in range(token_count): required_gap = max(required_gap, len(output) + 2 - (7 + reader.consumed)) if reader.bit(): room(1) output.append(reader.byte()) continue length = 0 raw = False if not reader.bit(): length = 2 + reader.bit() elif reader.bit(): length = 4 + reader.msb(2) elif not reader.bit(): length = 8 + reader.msb(4) else: marker = reader.byte() raw = not (marker & 1) length = reader.word_le() if marker in (254, 255) else marker >> 1 if raw: room(length) for _ in range(length): output.append(reader.byte()) continue if not reader.bit(): distance = (reader.msb(1) << 8) + reader.byte() + 33 elif not reader.bit(): distance = reader.msb(5) + 1 else: high = reader.msb(x + 1) distance = ( reader.word_le() if high == 62 else (high << 8) + reader.byte() + 512 ) if distance <= 0 or distance > len(output): raise CbsError("Invalid back-reference in CBS PCK stream.") room(length) required_gap = max( required_gap, len(output) + length - (7 + reader.consumed), ) for _ in range(length): output.append(output[-distance]) restored = bytes(output) + packed[0:4] available_gap = (original_size if original_size is not None else len(restored)) - len(packed) required_gap = max(0, required_gap) return restored, Safety( safe=available_gap >= required_gap, required_gap=required_gap, available_gap=available_gap, ) def decompress_cbs(packed: bytes) -> bytes: """Decompress a CBS PCK stream.""" return _decode_cbs(packed)[0] def analyze_in_place_safety(packed: bytes, original_size: int) -> Safety: """Analyze the original end-aligned RAM-source Z80 decrunch layout.""" return _decode_cbs(packed, original_size)[1] def _v3_match( source: bytes, length: int, position: int, previous: array, far_threshold: int, ) -> Tuple[int, int]: maximum = length - position if maximum < 2: return 0, 0 best_length = 0 best_distance = 0 candidate = previous[position] while candidate >= 0: distance = position - candidate matched = 2 while ( matched < maximum and source[candidate + matched] == source[position + matched] ): matched += 1 if distance >= FIRST_LONG_DISTANCE and matched < far_threshold: candidate = previous[candidate] continue if matched > best_length: best_length = matched best_distance = distance if matched == maximum: break candidate = previous[candidate] if best_length == 2 and best_distance >= 545: return 0, 0 return best_length, best_distance def _write_literals(writer: BitWriter, values: bytearray) -> int: if len(values) >= 12: writer.lsb(10, 4) _long_length(writer, len(values), True) for value in values: writer.byte(value) return 1 for value in values: writer.bit(1) writer.byte(value) return len(values) def _v3_pass( source: bytes, length: int, previous: array, far_threshold: int, x: int, progress: Optional[ProgressCallback], pass_number: int, pass_count: int, ) -> Tuple[bytes, int]: writer = BitWriter() raw = bytearray() token_count = 0 position = 0 while position < length: if progress is not None and (position & 2047) == 0: progress( { "engine": 3, "phase": 1, "pass": pass_number, "passes": pass_count, "threshold": far_threshold, "position": position, "total": length, } ) match_length, match_distance = _v3_match( source, length, position, previous, far_threshold, ) if match_length < 2: raw.append(source[position]) position += 1 continue token_count += _write_literals(writer, raw) raw.clear() _write_match(writer, match_length, match_distance, x) token_count += 1 position += match_length token_count += _write_literals(writer, raw) return writer.finish(), token_count def compress_v3( original: bytes, progress: Optional[ProgressCallback] = None, ) -> CompressionResult: """Compress with the original CBS v3 greedy algorithm.""" if len(original) < 5: raise CbsError("The original format needs at least 5 bytes.") if len(original) > MAX_FILE_SIZE: raise CbsError( f"The 1996 Amiga version accepts at most {MAX_FILE_SIZE:,} bytes." ) payload_size = len(original) - 4 previous = _build_links(original, payload_size) x, distance_log = _settings(payload_size) thresholds = list(range(4, 17)) if payload_size >= FIRST_LONG_DISTANCE else [4] trials: List[Trial] = [] best: Optional[Tuple[bytes, int, int, Safety]] = None for index, threshold in enumerate(thresholds): stream, token_count = _v3_pass( original, payload_size, previous, threshold, x, progress, index + 1, len(thresholds), ) packed = _assemble( original, payload_size, stream, token_count, distance_log, ) safety = analyze_in_place_safety(packed, len(original)) trials.append( Trial( threshold=threshold, size=len(packed), safe=safety.safe, required_gap=safety.required_gap, available_gap=safety.available_gap, ) ) if safety.safe and (best is None or len(stream) < len(best[0])): best = stream, token_count, threshold, safety if best is None: raise CbsError( "No 4-16 trial is safe for end-aligned in-place decrunching. " "The packed stream would be overwritten before it is fully read." ) stream, token_count, threshold, safety = best output = _assemble( original, payload_size, stream, token_count, distance_log, ) restored = decompress_cbs(output) verified = restored == original if not verified: raise CbsError("Internal v3 round-trip verification failed.") return CompressionResult( output=output, input_size=len(original), payload_size=payload_size, output_size=len(output), token_count=token_count, threshold=threshold, trials=trials, verified=verified, in_place_safe=safety.safe, required_gap=safety.required_gap, available_gap=safety.available_gap, engine="v3", distance_high_bits=x + 1, distance_log=distance_log, ) def _v4_match_options( source: bytes, length: int, position: int, previous: array, far_threshold: int, x: int, ) -> List[Tuple[int, int]]: maximum = length - position if maximum < 2: return [] best: List[Optional[Tuple[int, int]]] = [None, None, None, None] maximum_encodable_distance = _maximum_distance(x) visits = 0 candidate = previous[position] while candidate >= 0 and visits < MATCH_SEARCH_DEPTH: visits += 1 distance = position - candidate if distance > maximum_encodable_distance: break matched = 2 while ( matched < maximum and source[candidate + matched] == source[position + matched] ): matched += 1 if distance >= FIRST_LONG_DISTANCE and matched < far_threshold: candidate = previous[candidate] continue if matched == 2 and distance >= 545: candidate = previous[candidate] continue if distance <= 32: distance_class = 0 elif distance <= 544: distance_class = 1 elif distance <= 0x3FFF: distance_class = 2 else: distance_class = 3 current = best[distance_class] if current is None or matched > current[0]: best[distance_class] = matched, distance if matched == maximum and distance_class == 0: break candidate = previous[candidate] return [option for option in best if option is not None] def _length_shape(length: int) -> Tuple[int, int]: if length < 4: return 3, 0 if length < 8: return 5, 0 if length < 24: return 8, 0 return 4, 1 if length < 127 else 3 def _distance_shape(distance: int, x: int) -> Tuple[int, int]: if distance <= 32: return 7, 0 if distance <= 544: return 2, 1 if distance <= _maximum_distance(x) and ( x < 5 or distance < FIRST_LONG_DISTANCE ): return 2 + (x + 1), 1 if x == 5: return 8, 2 raise CbsError("Distance cannot be represented by this CBS scale.") @lru_cache(maxsize=None) def _candidate_lengths(maximum: int) -> Tuple[int, ...]: values = set() def add(length: int) -> None: if 2 <= length <= maximum: values.add(length) for length in range(2, min(maximum, 32) + 1): add(length) for boundary in ( 47, 63, 95, 126, 127, 128, 191, 255, 256, 383, 511, 512, 767, 1023, 1024, 1535, 2047, 2048, 4095, 4096, ): add(boundary) for tail in ( 0, 1, 2, 3, 4, 5, 6, 7, 8, 15, 16, 23, 24, 31, 32, 47, 48, 63, 64, 95, 96, 126, 127, 128, 191, 255, 256, 383, 511, 512, ): add(maximum - tail) add(maximum // 2) return tuple(sorted(values)) @lru_cache(maxsize=None) def _raw_lengths(remaining: int) -> Tuple[int, ...]: values = set() def add(length: int) -> None: if 12 <= length <= remaining: values.add(length) for length in range(12, min(remaining, 32) + 1): add(length) for boundary in ( 47, 63, 95, 126, 127, 128, 191, 255, 256, 383, 511, 512, 767, 1023, 1024, ): add(boundary) for tail in ( 0, 1, 2, 3, 4, 7, 8, 15, 16, 31, 32, 63, 64, 127, 128, 255, 256, 511, 512, ): add(remaining - tail) return tuple(sorted(values)) def _advance_control(phase: int, control_bits: int) -> Tuple[int, int]: available = 16 - phase if control_bits <= available: return phase + control_bits, 0 remaining = control_bits - available next_phase = remaining % 16 or 16 extra_bytes = ((remaining + 15) // 16) * 2 return next_phase, extra_bytes def _token_shape( kind: int, length: int, distance: int, x: int, ) -> Tuple[int, int]: if kind == 0: return 1, 1 if kind == 1: return 4, length + (1 if length < 127 else 3) length_control, length_data = _length_shape(length) distance_control, distance_data = _distance_shape(distance, x) return length_control + distance_control, length_data + distance_data def _v4_optimize( source: bytes, length: int, previous: array, far_threshold: int, x: int, progress: Optional[ProgressCallback], trial_number: int, trial_count: int, ) -> Tuple[array, array, bytearray, int]: state_count = (length + 1) * CONTROL_STATES cost = array("d", [0.0]) * state_count choice_length = array("H", [0]) * (length * CONTROL_STATES) choice_distance = array("H", [0]) * (length * CONTROL_STATES) choice_kind = bytearray(length * CONTROL_STATES) control_transitions = tuple( tuple(_advance_control(phase, bits) for bits in range(17)) for phase in range(CONTROL_STATES) ) for position in range(length - 1, -1, -1): raw_options = _raw_lengths(length - position) # Candidate order is significant for equal-cost tie breaks: # literal, ascending raw runs, then distance classes near to far. candidates: List[Tuple[int, int, int, int, int]] = [(0, 1, 0, 1, 1)] for raw_length in raw_options: candidates.append( ( 1, raw_length, 0, 4, raw_length + (1 if raw_length < 127 else 3), ) ) for match_length, match_distance in _v4_match_options( source, length, position, previous, far_threshold, x, ): distance_control, distance_data = _distance_shape(match_distance, x) for candidate_length in _candidate_lengths(match_length): length_control, length_data = _length_shape(candidate_length) candidates.append( ( 2, candidate_length, match_distance, length_control + distance_control, length_data + distance_data, ) ) for phase in range(CONTROL_STATES): best_cost = float("inf") best_length = 1 best_distance = 0 best_kind = 0 phase_transitions = control_transitions[phase] for ( kind, token_length, distance, control_bits, data_bytes, ) in candidates: next_phase, extra_bytes = phase_transitions[control_bits] future = (position + token_length) * CONTROL_STATES + next_phase candidate_cost = data_bytes + extra_bytes + cost[future] if candidate_cost < best_cost: best_cost = candidate_cost best_length = token_length best_distance = distance best_kind = kind index = position * CONTROL_STATES + phase cost[index] = best_cost choice_length[index] = best_length choice_distance[index] = best_distance choice_kind[index] = best_kind if progress is not None and (position & 1023) == 0: progress( { "engine": 43, "phase": 2, "trial": trial_number, "trials": trial_count, "distance_high_bits": x + 1, "position": length - position, "total": length, } ) return ( choice_length, choice_distance, choice_kind, int(2 + cost[0]), ) def _v4_encode( source: bytes, length: int, choices: Tuple[array, array, bytearray, int], x: int, ) -> Tuple[bytes, int, int, int, int]: choice_length, choice_distance, choice_kind, _ = choices writer = BitWriter() position = 0 tokens = 0 literal_bytes = 0 match_bytes = 0 control_phase = 0 control_bits_total = 0 while position < length: choice_index = position * CONTROL_STATES + control_phase token_length = choice_length[choice_index] kind = choice_kind[choice_index] token_distance = choice_distance[choice_index] if kind == 0: writer.bit(1) writer.byte(source[position]) literal_bytes += 1 elif kind == 1: writer.lsb(10, 4) _long_length(writer, token_length, True) for index in range(token_length): writer.byte(source[position + index]) literal_bytes += token_length else: _write_match(writer, token_length, token_distance, x) match_bytes += token_length control_bits, _ = _token_shape(kind, token_length, token_distance, x) control_bits_total += control_bits control_phase, _ = _advance_control(control_phase, control_bits) position += token_length tokens += 1 return ( writer.finish(), tokens, literal_bytes, match_bytes, control_bits_total, ) def _v43_scale_trial( original: bytes, payload_size: int, threshold: int, x: int, distance_log: int, progress: Optional[ProgressCallback] = None, trial_number: int = 1, trial_count: int = 1, ) -> Tuple[int, int, bytes, int, int, int, int, Safety, bool, ScaleTrial]: previous = _build_links(original, payload_size) choices = _v4_optimize( original, payload_size, previous, threshold, x, progress, trial_number, trial_count, ) stream, tokens, literals, matches, control_bits = _v4_encode( original, payload_size, choices, x, ) if len(stream) != choices[3]: raise CbsError("Internal v4.3 control-word cost mismatch.") output = _assemble( original, payload_size, stream, tokens, distance_log, ) safety = analyze_in_place_safety(output, len(original)) verified = decompress_cbs(output) == original scale_trial = ScaleTrial( distance_high_bits=x + 1, distance_log=distance_log, maximum_distance=_maximum_distance(x), output_size=len(output), token_count=tokens, control_words=(control_bits + 15) // 16, verified=verified, in_place_safe=safety.safe, ) return ( x, distance_log, output, tokens, literals, matches, control_bits, safety, verified, scale_trial, ) def compress_v43( original: bytes, threshold: int = 4, progress: Optional[ProgressCallback] = None, jobs: Optional[int] = None, ) -> CompressionResult: """Compress with the CBS v4.3 aligned-cost parser.""" if len(original) < 5: raise CbsError("CBS Cruncher needs at least 5 bytes.") if len(original) > MAX_FILE_SIZE: raise CbsError(f"Maximum input is {MAX_FILE_SIZE:,} bytes.") if threshold < 2 or threshold > MAX_PAYLOAD_SIZE: raise CbsError( f"The far-match threshold must be between 2 and {MAX_PAYLOAD_SIZE}." ) payload_size = len(original) - 4 if jobs is None: jobs = min(len(SCALE_SETTINGS), os.cpu_count() or 1) if jobs < 1: raise CbsError("The number of v4.3 worker jobs must be at least 1.") jobs = min(jobs, len(SCALE_SETTINGS)) # Process startup costs more than it saves on small inputs. use_parallel = jobs > 1 and payload_size >= 2048 completed_trials: List[ Tuple[int, int, bytes, int, int, int, int, Safety, bool, ScaleTrial] ] = [] if use_parallel: with ProcessPoolExecutor(max_workers=jobs) as executor: futures = { executor.submit( _v43_scale_trial, original, payload_size, threshold, x, distance_log, ): (trial_index, x) for trial_index, (x, distance_log) in enumerate(SCALE_SETTINGS) } finished = 0 for future in as_completed(futures): trial_index, x = futures[future] del trial_index completed_trials.append(future.result()) finished += 1 if progress is not None: progress( { "engine": 43, "phase": 3, "trial": finished, "trials": len(SCALE_SETTINGS), "distance_high_bits": x + 1, "position": finished, "total": len(SCALE_SETTINGS), } ) else: for trial_index, (x, distance_log) in enumerate(SCALE_SETTINGS): completed_trials.append( _v43_scale_trial( original, payload_size, threshold, x, distance_log, progress, trial_index + 1, len(SCALE_SETTINGS), ) ) order = {x: index for index, (x, _) in enumerate(SCALE_SETTINGS)} completed_trials.sort(key=lambda result: order[result[0]]) scale_trials: List[ScaleTrial] = [] best: Optional[ Tuple[int, int, bytes, int, int, int, int, Safety, bool] ] = None for ( x, distance_log, output, tokens, literals, matches, control_bits, safety, verified, scale_trial, ) in completed_trials: scale_trials.append(scale_trial) if verified and ( best is None or len(output) < len(best[2]) or ( len(output) == len(best[2]) and safety.safe and not best[7].safe ) ): best = ( x, distance_log, output, tokens, literals, matches, control_bits, safety, verified, ) if best is None: raise CbsError("No CBS distance-scale trial round-tripped correctly.") ( x, distance_log, output, tokens, literals, matches, control_bits, safety, verified, ) = best trial = Trial( threshold=threshold, size=len(output), safe=safety.safe, required_gap=safety.required_gap, available_gap=safety.available_gap, ) return CompressionResult( output=output, input_size=len(original), payload_size=payload_size, output_size=len(output), token_count=tokens, threshold=threshold, trials=[trial], verified=verified, in_place_safe=safety.safe, required_gap=safety.required_gap, available_gap=safety.available_gap, engine="v4.3", literal_bytes=literals, match_bytes=matches, control_words=(control_bits + 15) // 16, search_depth=MATCH_SEARCH_DEPTH, distance_high_bits=x + 1, distance_log=distance_log, scale_trials=sorted( scale_trials, key=lambda trial_value: trial_value.distance_high_bits, ), ) class CliProgress: def __init__(self, quiet: bool) -> None: self.quiet = quiet self.last_update = 0.0 def __call__(self, progress: Dict[str, int]) -> None: if self.quiet: return now = time.monotonic() if now - self.last_update < 0.20: return self.last_update = now total = max(1, progress.get("total", 1)) position = progress.get("position", 0) percent = min(100.0, position * 100.0 / total) if progress.get("engine") == 3: label = ( f"v3 threshold {progress.get('threshold')} " f"({progress.get('pass')}/{progress.get('passes')})" ) else: label = ( f"v4.3 scale {progress.get('distance_high_bits')} bits " f"({progress.get('trial')}/{progress.get('trials')})" ) print(f"\r{label}: {percent:5.1f}%", end="", flush=True) def finish(self) -> None: if not self.quiet and self.last_update: print("\r" + (" " * 72) + "\r", end="", flush=True) def _default_packed_path(input_path: Path) -> Path: return input_path.with_suffix(".pck") def _default_unpacked_path(input_path: Path) -> Path: extension = ".BIN" if input_path.suffix.isupper() else ".bin" return input_path.with_suffix(extension) def _write_output(path: Path, data: bytes, force: bool) -> None: if path.exists() and not force: raise CbsError( f"Output already exists: {path}. Use --force to overwrite it." ) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) def _print_result(result: CompressionResult, output_path: Path, elapsed: float) -> None: change = (1.0 - result.output_size / result.input_size) * 100.0 print(f"{result.engine}: {result.input_size:,} -> {result.output_size:,} bytes") print(f"Saved: {change:.1f}%") print(f"Tokens: {result.token_count:,}") print(f"Far-match threshold: {result.threshold}") print( "End-aligned RAM safety: " + ( f"safe ({result.available_gap}-byte gap, " f"{result.required_gap} required)" if result.in_place_safe else f"unsafe ({result.available_gap}-byte gap, " f"{result.required_gap} required)" ) ) if result.engine == "v4.3": print( f"Selected distance field: {result.distance_high_bits} high bits " f"(header scale {result.distance_log})" ) print(f"Verified: {'yes' if result.verified else 'no'}") print(f"Time: {elapsed:.2f} seconds") print(f"Wrote: {output_path}") def _compress_one( input_path: Path, output_path: Path, engine: str, threshold: int, force: bool, quiet: bool, jobs: int, ) -> CompressionResult: source = input_path.read_bytes() reporter = CliProgress(quiet) started = time.perf_counter() try: result = ( compress_v3(source, reporter) if engine == "v3" else compress_v43(source, threshold, reporter, jobs) ) finally: reporter.finish() elapsed = time.perf_counter() - started _write_output(output_path, result.output, force) if not quiet: _print_result(result, output_path, elapsed) return result def _compress_command(args: argparse.Namespace) -> int: input_path = Path(args.input) if not input_path.is_file(): raise CbsError(f"Input file was not found: {input_path}") if args.engine == "both": if args.output: base = Path(args.output) if base.suffix: base = base.with_suffix("") else: base = input_path.with_suffix("") v3_path = Path(str(base) + ".v3.pck") v43_path = Path(str(base) + ".v43.pck") if not args.force: existing = [path for path in (v3_path, v43_path) if path.exists()] if existing: raise CbsError( "Output already exists: " + ", ".join(str(path) for path in existing) + ". Use --force to overwrite." ) _compress_one( input_path, v3_path, "v3", args.threshold, args.force, args.quiet, args.jobs, ) if not args.quiet: print() _compress_one( input_path, v43_path, "v4.3", args.threshold, args.force, args.quiet, args.jobs, ) return 0 output_path = Path(args.output) if args.output else _default_packed_path(input_path) _compress_one( input_path, output_path, args.engine, args.threshold, args.force, args.quiet, args.jobs, ) return 0 def _decompress_command(args: argparse.Namespace) -> int: input_path = Path(args.input) if not input_path.is_file(): raise CbsError(f"Input file was not found: {input_path}") output_path = ( Path(args.output) if args.output else _default_unpacked_path(input_path) ) packed = input_path.read_bytes() started = time.perf_counter() restored = decompress_cbs(packed) elapsed = time.perf_counter() - started _write_output(output_path, restored, args.force) if not args.quiet: print(f"Decompressed: {len(packed):,} -> {len(restored):,} bytes") print(f"Time: {elapsed:.3f} seconds") print(f"Wrote: {output_path}") return 0 def _auto_command(args: argparse.Namespace) -> int: if Path(args.input).suffix.lower() == ".pck": return _decompress_command(args) return _compress_command(args) def _add_common_file_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("input", help="Input binary or PCK file") parser.add_argument("-o", "--output", help="Output file or base name") parser.add_argument( "-f", "--force", action="store_true", help="Overwrite an existing output file", ) parser.add_argument( "-q", "--quiet", action="store_true", help="Suppress progress and result details", ) def _add_compression_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument( "-e", "--engine", choices=("v3", "v4.3", "both"), default="v4.3", help="Compression engine (default: v4.3)", ) parser.add_argument( "-t", "--threshold", type=int, default=4, help="v4.3 minimum length for very-far matches (default: 4)", ) parser.add_argument( "-j", "--jobs", type=int, default=min(len(SCALE_SETTINGS), os.cpu_count() or 1), help=( "parallel v4.3 distance-scale workers " f"(default: {min(len(SCALE_SETTINGS), os.cpu_count() or 1)}; " "use 1 for sequential)" ), ) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="cbscrunch", description=( "Cross-platform CBS Cruncher v3/v4.3 compressor and PCK decruncher." ), ) parser.add_argument( "--version", action="version", version=f"CBS Cruncher Python {VERSION}", ) subparsers = parser.add_subparsers(dest="command") auto_parser = subparsers.add_parser( "auto", help="Compress a binary or automatically decrunch a .PCK file", ) _add_common_file_arguments(auto_parser) _add_compression_arguments(auto_parser) auto_parser.set_defaults(handler=_auto_command) compress_parser = subparsers.add_parser("compress", help="Compress a binary") _add_common_file_arguments(compress_parser) _add_compression_arguments(compress_parser) compress_parser.set_defaults(handler=_compress_command) decompress_parser = subparsers.add_parser( "decompress", aliases=("decrunch",), help="Decompress a CBS PCK file", ) _add_common_file_arguments(decompress_parser) decompress_parser.set_defaults(handler=_decompress_command) return parser def main(argv: Optional[Sequence[str]] = None) -> int: arguments = list(sys.argv[1:] if argv is None else argv) commands = {"auto", "compress", "decompress", "decrunch"} if arguments and arguments[0] not in commands and arguments[0] not in ( "-h", "--help", "--version", ): arguments.insert(0, "auto") parser = _build_parser() args = parser.parse_args(arguments) if not hasattr(args, "handler"): parser.print_help() return 0 try: return int(args.handler(args)) except (CbsError, OSError) as error: print(f"Error: {error}", file=sys.stderr) return 1 except KeyboardInterrupt: print("\nCancelled.", file=sys.stderr) return 130 if __name__ == "__main__": raise SystemExit(main())