Add compress, verify, upload and mount backend

This commit is contained in:
unicxrn committed 2026-10-10 16:20:32 +02:00
commit 55e4bc11ba
24 files changed
+8944

No files matched your search

+4
View File
@@ -0,0 +1,4 @@
node_modules/
dist/
src-tauri/target/
src-tauri/gen/
+1373
View File
File diff suppressed because it is too large. Load diff
+23
View File
@@ -0,0 +1,23 @@
{
"name": "ps5-packer",
"private": true,
"version": "0.1.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"preview": "vite preview",
"tauri": "tauri"
},
"devDependencies": {
"@sveltejs/vite-plugin-svelte": "^7.3.1",
"@tauri-apps/cli": "^2.12.1",
"svelte": "^5.57.2",
"vite": "^8.3.4"
},
"dependencies": {
"@tauri-apps/api": "^2.12.2",
"@tauri-apps/plugin-dialog": "^2.8.1",
"@tauri-apps/plugin-opener": "^2.7.0"
}
}
+5021
View File
File diff suppressed because it is too large. Load diff
+29
View File
@@ -0,0 +1,29 @@
[package]
name = "ps5-packer"
version = "0.1.0"
description = "Compress, verify and upload PS5 game dumps"
edition = "2021"
[lib]
name = "ps5_packer_lib"
crate-type = ["staticlib", "cdylib", "rlib"]
[build-dependencies]
tauri-build = { version = "2", features = [] }
[dependencies]
tauri = { version = "2", features = [] }
tauri-plugin-dialog = "2"
tauri-plugin-opener = "2"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
image = { version = "0.25", default-features = false, features = ["png", "jpeg"] }
base64 = "0.22"
libc = "0.2"
[dev-dependencies]
tempfile = "3"
[profile.release]
opt-level = 3
lto = "thin"
+3
View File
@@ -0,0 +1,3 @@
fn main() {
tauri_build::build()
}
+16
View File
@@ -0,0 +1,16 @@
{
"$schema": "../gen/schemas/desktop-schema.json",
"identifier": "default",
"description": "Main window",
"windows": ["main"],
"permissions": [
"core:default",
"core:window:allow-start-dragging",
"core:window:allow-minimize",
"core:window:allow-toggle-maximize",
"core:window:allow-close",
"dialog:default",
"opener:default",
"opener:allow-reveal-item-in-dir"
]
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 5.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 10 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.4 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.8 KiB

Binary file not shown.
Binary file not shown.

After

Width:  |  Height:  |  Size: 19 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 20 KiB

+155
View File
@@ -0,0 +1,155 @@
"""Deep-verify an exFAT-wrapped .ffpfsc against its source folder.
Decompresses the image through MkPFS's own reader, walks the inner exFAT and
compares every file byte-for-byte with the source tree. Reads only; never
writes to the image or the source.
Output: one JSON object per line on stdout.
{"type":"plan","files":N,"bytes":B}
{"type":"progress","files":n,"bytes":b}
{"type":"problem","kind":"missing|extra|size|content|read","path":"...","detail":"..."}
{"type":"done","ok":true|false,"files":N,"bytes":B,"problems":P}
"""
from __future__ import annotations
import json
import multiprocessing as mp
import os
import sys
import time
from pathlib import Path
CHUNK = 4 * 1024 * 1024
def emit(obj: dict) -> None:
sys.stdout.write(json.dumps(obj) + "\n")
sys.stdout.flush()
def open_reader(image: Path):
from mkpfs.exfat import ExfatReader
from mkpfs.pfs import open_inner_file_view
opened = open_inner_file_view(image)
if opened is None:
raise RuntimeError("image is not a single-file PFSC container (expected an exFAT-wrapped .ffpfsc)")
view, fh, _name = opened
return ExfatReader(view), fh
def scan_source(root: Path) -> dict[str, int]:
out: dict[str, int] = {}
for dirpath, _dirs, files in os.walk(root):
for name in files:
full = Path(dirpath) / name
if full.is_symlink():
continue
out[full.relative_to(root).as_posix()] = full.stat().st_size
return out
def worker(image: str, source: str, paths: list[str], q) -> None:
try:
reader, fh = open_reader(Path(image))
except Exception as exc: # noqa: BLE001
q.put(("fatal", str(exc)))
return
try:
entries = {e.rel_path.lstrip("/"): e for e in reader.iter_files()}
for rel in paths:
entry = entries[rel]
done = 0
try:
with open(Path(source) / rel, "rb") as src:
for chunk in reader.read_file(entry, CHUNK):
ref = src.read(len(chunk))
if ref != chunk:
off = done + next((i for i, (a, b) in enumerate(zip(ref, chunk)) if a != b), min(len(ref), len(chunk)))
q.put(("problem", "content", rel, f"first difference at byte {off}"))
break
done += len(chunk)
q.put(("bytes", len(chunk)))
except Exception as exc: # noqa: BLE001
q.put(("problem", "read", rel, str(exc)))
q.put(("file", entry.length - done))
finally:
fh.close()
q.put(("exit",))
def main() -> int:
if len(sys.argv) < 3:
print("usage: deep_verify.py IMAGE SOURCE_DIR [WORKERS]", file=sys.stderr)
return 2
image, source = Path(sys.argv[1]), Path(sys.argv[2])
workers = int(sys.argv[3]) if len(sys.argv) > 3 else max(1, min(8, (os.cpu_count() or 2) // 2))
try:
reader, fh = open_reader(image)
inner = {e.rel_path.lstrip("/"): e.length for e in reader.iter_files()}
fh.close()
except Exception as exc: # noqa: BLE001
emit({"type": "problem", "kind": "read", "path": str(image), "detail": str(exc)})
emit({"type": "done", "ok": False, "files": 0, "bytes": 0, "problems": 1})
return 1
src = scan_source(source)
problems = 0
for rel in sorted(src.keys() - inner.keys()):
emit({"type": "problem", "kind": "missing", "path": rel, "detail": "in dump but not in image"}); problems += 1
for rel in sorted(inner.keys() - src.keys()):
emit({"type": "problem", "kind": "extra", "path": rel, "detail": "in image but not in dump"}); problems += 1
common = sorted(src.keys() & inner.keys())
for rel in common:
if src[rel] != inner[rel]:
emit({"type": "problem", "kind": "size", "path": rel, "detail": f"dump {src[rel]} B, image {inner[rel]} B"}); problems += 1
to_check = [r for r in common if src[r] == inner[r]]
total_bytes = sum(inner[r] for r in to_check)
emit({"type": "plan", "files": len(to_check), "bytes": total_bytes, "workers": workers})
# Largest-first round robin keeps workers evenly loaded.
buckets: list[list[str]] = [[] for _ in range(workers)]
loads = [0] * workers
for rel in sorted(to_check, key=lambda r: inner[r], reverse=True):
i = loads.index(min(loads)); buckets[i].append(rel); loads[i] += inner[rel]
ctx = mp.get_context("spawn")
q = ctx.Queue(maxsize=4096)
procs = [ctx.Process(target=worker, args=(str(image), str(source), b, q)) for b in buckets if b]
for p in procs:
p.start()
files_done = bytes_done = 0
alive = len(procs)
last = 0.0
while alive:
msg = q.get()
kind = msg[0]
if kind == "bytes":
bytes_done += msg[1]
elif kind == "file":
files_done += 1
bytes_done += msg[1] # count the unread tail of a failed file so totals line up
elif kind == "problem":
problems += 1
emit({"type": "problem", "kind": msg[1], "path": msg[2], "detail": msg[3]})
elif kind == "fatal":
problems += 1
emit({"type": "problem", "kind": "read", "path": str(image), "detail": msg[1]})
elif kind == "exit":
alive -= 1
now = time.monotonic()
if now - last > 0.25:
emit({"type": "progress", "files": files_done, "bytes": bytes_done})
last = now
for p in procs:
p.join()
emit({"type": "progress", "files": files_done, "bytes": bytes_done})
emit({"type": "done", "ok": problems == 0, "files": files_done, "bytes": bytes_done, "problems": problems})
return 0 if problems == 0 else 1
if __name__ == "__main__":
sys.exit(main())
+245
View File
@@ -0,0 +1,245 @@
//! Reads a dump folder. Never writes to it.
use base64::Engine;
use serde::Serialize;
use std::path::{Path, PathBuf};
/// Files whose presence marks an APR Emu title (same list Game Compressor checks).
pub const AMPR_MARKERS: [&str; 4] = [
"fakelib/libSceAmpr.sprx",
"fakelib/libSceAmpr.prx",
"sce_module/libSceAmpr.sprx",
"sce_module/libSceAmpr.prx",
];
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DumpInfo {
pub path: String,
pub title_id: String,
pub title_name: String,
pub version: String,
pub file_count: u64,
pub total_bytes: u64,
pub apr: bool,
pub has_ampr_index: bool,
/// Hex colour pulled from the icon, used as the UI accent for this game.
pub accent: String,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DumpArt {
/// data: URLs, so the webview needs no file access.
pub icon: Option<String>,
pub hero: Option<String>,
}
pub fn inspect(path: &Path) -> Result<DumpInfo, String> {
let param_path = path.join("sce_sys/param.json");
let raw = std::fs::read_to_string(&param_path)
.map_err(|_| "This folder isn't a PS5 dump: sce_sys/param.json is missing.".to_string())?;
let param: serde_json::Value =
serde_json::from_str(&raw).map_err(|e| format!("sce_sys/param.json is not valid JSON: {e}"))?;
let title_id = param["titleId"].as_str().unwrap_or("").trim().to_string();
if title_id.len() != 9 || !title_id.chars().all(|c| c.is_ascii_alphanumeric()) {
return Err(format!("Unexpected title ID {title_id:?} in sce_sys/param.json."));
}
let loc = &param["localizedParameters"];
let lang = loc["defaultLanguage"].as_str().unwrap_or("en-US");
let title_name = loc[lang]["titleName"]
.as_str()
.or_else(|| loc["en-US"]["titleName"].as_str())
.unwrap_or(&title_id)
.to_string();
let version = param["contentVersion"].as_str().unwrap_or("").to_string();
if !path.join("eboot.bin").is_file() {
return Err("This folder has no eboot.bin, so it doesn't look like a complete game dump.".into());
}
let (file_count, total_bytes) = tally(path);
let apr = AMPR_MARKERS.iter().any(|m| path.join(m).is_file());
let has_ampr_index = path.join("ampr_emu.index").is_file();
let accent = icon_path(path)
.and_then(|p| image::open(p).ok())
.map(|img| accent_from(&img))
.unwrap_or_else(|| "#3d7bff".into());
Ok(DumpInfo {
path: path.to_string_lossy().into_owned(),
title_id,
title_name,
version,
file_count,
total_bytes,
apr,
has_ampr_index,
accent,
})
}
/// File count and byte total, skipping symlinks (MkPFS refuses them anyway).
pub fn tally(root: &Path) -> (u64, u64) {
let mut files = 0u64;
let mut bytes = 0u64;
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let Ok(rd) = std::fs::read_dir(&dir) else { continue };
for entry in rd.flatten() {
let Ok(ft) = entry.file_type() else { continue };
if ft.is_symlink() {
continue;
} else if ft.is_dir() {
stack.push(entry.path());
} else if ft.is_file() {
files += 1;
bytes += entry.metadata().map(|m| m.len()).unwrap_or(0);
}
}
}
(files, bytes)
}
fn icon_path(dump: &Path) -> Option<PathBuf> {
let p = dump.join("sce_sys/icon0.png");
p.is_file().then_some(p)
}
fn hero_path(dump: &Path) -> Option<PathBuf> {
["sce_sys/pic0.png", "sce_sys/pic1.png", "sce_sys/pic2.png"]
.iter()
.map(|p| dump.join(p))
.find(|p| p.is_file())
}
/// Downscaled artwork as data URLs, cached on disk per title so re-adding is instant.
pub fn art(dump: &Path, title_id: &str, cache_root: &Path) -> DumpArt {
let cache = cache_root.join("art").join(title_id);
let _ = std::fs::create_dir_all(&cache);
let icon = cached_jpeg(icon_path(dump), &cache.join("icon.jpg"), 256);
let hero = cached_jpeg(hero_path(dump), &cache.join("hero.jpg"), 1920);
DumpArt { icon, hero }
}
fn cached_jpeg(src: Option<PathBuf>, dst: &Path, width: u32) -> Option<String> {
let src = src?;
let fresh = match (std::fs::metadata(&src), std::fs::metadata(dst)) {
(Ok(s), Ok(d)) => d.modified().ok() >= s.modified().ok(),
_ => false,
};
if !fresh {
let img = image::open(&src).ok()?;
let img = if img.width() > width {
img.resize(width, u32::MAX, image::imageops::FilterType::Lanczos3)
} else {
img
};
let mut out = std::fs::File::create(dst).ok()?;
let enc = image::codecs::jpeg::JpegEncoder::new_with_quality(&mut out, 86);
img.to_rgb8().write_with_encoder(enc).ok()?;
}
let bytes = std::fs::read(dst).ok()?;
Some(format!(
"data:image/jpeg;base64,{}",
base64::engine::general_purpose::STANDARD.encode(bytes)
))
}
/// Picks the most saturated common colour from the icon, nudged to stay readable on near-black.
fn accent_from(img: &image::DynamicImage) -> String {
let small = img.resize_exact(48, 48, image::imageops::FilterType::Triangle).to_rgb8();
// 12 hue buckets weighted by saturation * value.
let mut buckets = [(0f64, 0f64, 0f64, 0f64); 12];
for p in small.pixels() {
let (r, g, b) = (p[0] as f64 / 255.0, p[1] as f64 / 255.0, p[2] as f64 / 255.0);
let max = r.max(g).max(b);
let min = r.min(g).min(b);
let sat = if max > 0.0 { (max - min) / max } else { 0.0 };
if sat < 0.25 || max < 0.2 {
continue;
}
let hue = hue_of(r, g, b, max, min);
let w = sat * max;
let i = ((hue / 30.0) as usize).min(11);
buckets[i].0 += r * w;
buckets[i].1 += g * w;
buckets[i].2 += b * w;
buckets[i].3 += w;
}
let best = buckets.iter().max_by(|a, b| a.3.total_cmp(&b.3)).copied().unwrap();
if best.3 == 0.0 {
return "#3d7bff".into();
}
let (mut r, mut g, mut b) = (best.0 / best.3, best.1 / best.3, best.2 / best.3);
// Lift dark accents so they read on the #0b0b0c background.
let lum = 0.2126 * r + 0.7152 * g + 0.0722 * b;
if lum < 0.32 {
let k = 0.32 / lum.max(0.05);
let k = k.min(2.6);
r = (r * k).min(1.0);
g = (g * k).min(1.0);
b = (b * k).min(1.0);
}
format!("#{:02x}{:02x}{:02x}", (r * 255.0) as u8, (g * 255.0) as u8, (b * 255.0) as u8)
}
fn hue_of(r: f64, g: f64, b: f64, max: f64, min: f64) -> f64 {
let d = max - min;
if d == 0.0 {
return 0.0;
}
let h = if max == r {
((g - b) / d).rem_euclid(6.0)
} else if max == g {
(b - r) / d + 2.0
} else {
(r - g) / d + 4.0
};
h * 60.0
}
#[cfg(test)]
mod tests {
use super::*;
fn make_dump(dir: &Path) {
std::fs::create_dir_all(dir.join("sce_sys")).unwrap();
std::fs::create_dir_all(dir.join("fakelib")).unwrap();
std::fs::write(
dir.join("sce_sys/param.json"),
r#"{"titleId":"PPSA00001","contentVersion":"01.002.000","localizedParameters":{"defaultLanguage":"en-US","en-US":{"titleName":"TEST GAME"}}}"#,
)
.unwrap();
std::fs::write(dir.join("eboot.bin"), vec![1u8; 1000]).unwrap();
std::fs::write(dir.join("fakelib/libSceAmpr.sprx"), b"x").unwrap();
}
#[test]
fn inspects_a_dump() {
let t = tempfile::tempdir().unwrap();
make_dump(t.path());
let info = inspect(t.path()).unwrap();
assert_eq!(info.title_id, "PPSA00001");
assert_eq!(info.title_name, "TEST GAME");
assert_eq!(info.version, "01.002.000");
assert_eq!(info.file_count, 3);
assert!(info.apr);
assert!(!info.has_ampr_index);
}
#[test]
fn rejects_non_dumps() {
let t = tempfile::tempdir().unwrap();
assert!(inspect(t.path()).is_err());
}
#[test]
fn accent_is_lifted_for_dark_icons() {
let img = image::DynamicImage::ImageRgb8(image::RgbImage::from_pixel(8, 8, image::Rgb([10, 20, 120])));
let hex = accent_from(&img);
let v = u32::from_str_radix(&hex[1..], 16).unwrap();
assert!(((v >> 16) & 0xff) + ((v >> 8) & 0xff) + (v & 0xff) > 150, "{hex} too dark");
}
}
+414
View File
@@ -0,0 +1,414 @@
//! The compression engine: a private Python venv with MkPFS, plus the deep verifier.
use serde::Serialize;
use std::io::Read;
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
/// Pinned so a new MkPFS release can't silently change the image format.
pub const MKPFS_VERSION: &str = "1.1.0";
const VERIFIER: &str = include_str!("../resources/deep_verify.py");
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EngineStatus {
pub python: Option<String>,
pub mkpfs: Option<String>,
pub ready: bool,
pub dir: String,
}
pub struct Engine {
root: PathBuf,
}
impl Engine {
pub fn new(data_dir: &Path) -> Self {
Self { root: data_dir.join("engine") }
}
fn venv(&self) -> PathBuf {
self.root.join("venv")
}
fn mkpfs_bin(&self) -> PathBuf {
self.venv().join("bin/mkpfs")
}
fn python_bin(&self) -> PathBuf {
self.venv().join("bin/python")
}
fn verifier(&self) -> PathBuf {
self.root.join("deep_verify.py")
}
pub fn status(&self) -> EngineStatus {
let python = Command::new("python3")
.arg("--version")
.output()
.ok()
.filter(|o| o.status.success())
.map(|o| String::from_utf8_lossy(&o.stdout).trim().replace("Python ", ""));
let mkpfs = if self.mkpfs_bin().is_file() {
Command::new(self.python_bin())
.args(["-c", "import importlib.metadata as m; print(m.version('mkpfs'))"])
.output()
.ok()
.filter(|o| o.status.success())
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
} else {
None
};
let ready = mkpfs.is_some();
EngineStatus { python, mkpfs, ready, dir: self.root.to_string_lossy().into_owned() }
}
/// Creates the venv and installs the pinned MkPFS. Streams pip's output to `log`.
pub fn install(&self, log: &mut dyn FnMut(String)) -> Result<(), String> {
std::fs::create_dir_all(&self.root).map_err(|e| e.to_string())?;
if !self.python_bin().is_file() {
log("Creating Python environment".into());
run_logged(Command::new("python3").arg("-m").arg("venv").arg(self.venv()), log)?;
}
log(format!("Installing MkPFS {MKPFS_VERSION}"));
run_logged(
Command::new(self.python_bin()).args([
"-m",
"pip",
"install",
"--disable-pip-version-check",
"--upgrade",
&format!("mkpfs=={MKPFS_VERSION}"),
]),
log,
)?;
self.write_verifier()?;
Ok(())
}
pub fn write_verifier(&self) -> Result<(), String> {
std::fs::create_dir_all(&self.root).map_err(|e| e.to_string())?;
if std::fs::read_to_string(self.verifier()).ok().as_deref() != Some(VERIFIER) {
std::fs::write(self.verifier(), VERIFIER).map_err(|e| e.to_string())?;
}
Ok(())
}
/// `mkpfs pack folder` into `out`. The inner exFAT is named after `out`'s stem,
/// so `out` must already be called `<TITLE>.ffpfsc`.
pub fn pack(
&self,
src: &Path,
out: &Path,
level: u8,
cpus: u32,
cancel: &Arc<AtomicBool>,
on: &mut dyn FnMut(PackEvent),
) -> Result<(), String> {
let tmp_dir = out.parent().unwrap().join("tmp");
std::fs::create_dir_all(&tmp_dir).map_err(|e| e.to_string())?;
let mut cmd = Command::new(self.mkpfs_bin());
cmd.args(["pack", "folder", "--no-adjust-output-file-extension", "--ampr-skip-regen-if-exists"])
.arg("--compression-level")
.arg(level.min(9).to_string())
.arg("--cpu-count")
.arg(cpus.to_string())
.arg("--temp-folder")
.arg(&tmp_dir)
.arg(src)
.arg(out);
let mut parser = PackParser::default();
let status = run_streaming(cmd, cancel, &mut |chunk| {
for ev in parser.feed(chunk) {
on(ev);
}
})?;
let _ = std::fs::remove_dir_all(&tmp_dir);
if cancel.load(Ordering::Relaxed) {
return Err("Cancelled".into());
}
if !status {
return Err(parser.last_error().unwrap_or_else(|| "MkPFS failed; see the activity log".into()));
}
Ok(())
}
pub fn verify(
&self,
image: &Path,
src: &Path,
workers: u32,
cancel: &Arc<AtomicBool>,
on: &mut dyn FnMut(VerifyEvent),
) -> Result<bool, String> {
self.write_verifier()?;
let workers = if workers == 0 {
(std::thread::available_parallelism().map(|n| n.get()).unwrap_or(2) / 2).clamp(1, 8) as u32
} else {
workers
};
let mut cmd = Command::new(self.python_bin());
cmd.arg(self.verifier()).arg(image).arg(src).arg(workers.to_string());
let mut buf = String::new();
let mut verdict: Option<bool> = None;
run_streaming(cmd, cancel, &mut |chunk| {
buf.push_str(chunk);
while let Some(nl) = buf.find('\n') {
let line: String = buf.drain(..=nl).collect();
if let Some(ev) = VerifyEvent::parse(line.trim()) {
if let VerifyEvent::Done { ok, .. } = ev {
verdict = Some(ok);
}
on(ev);
}
}
})?;
if cancel.load(Ordering::Relaxed) {
return Err("Cancelled".into());
}
verdict.ok_or_else(|| "The verifier stopped without a result".into())
}
}
fn run_logged(cmd: &mut Command, log: &mut dyn FnMut(String)) -> Result<(), String> {
let out = cmd.output().map_err(|e| e.to_string())?;
for line in String::from_utf8_lossy(&out.stdout).lines().chain(String::from_utf8_lossy(&out.stderr).lines()) {
if !line.trim().is_empty() {
log(line.to_string());
}
}
if out.status.success() {
Ok(())
} else {
Err(format!("{:?} exited with {}", cmd.get_program(), out.status))
}
}
/// Runs `cmd` in its own process group, feeding merged stdout+stderr to `on`.
/// Kills the whole group (MkPFS and verifier workers) when `cancel` flips.
fn run_streaming(mut cmd: Command, cancel: &Arc<AtomicBool>, on: &mut dyn FnMut(&str)) -> Result<bool, String> {
cmd.env("PYTHONUNBUFFERED", "1")
.env("PYTHONIOENCODING", "utf-8")
.env("COLUMNS", "120")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.process_group(0);
let mut child = cmd.spawn().map_err(|e| format!("Couldn't start {:?}: {e}", cmd.get_program()))?;
let pgid = child.id() as i32;
let (tx, rx) = std::sync::mpsc::channel::<Option<String>>();
let mut readers = Vec::new();
for mut pipe in [
Box::new(child.stdout.take().unwrap()) as Box<dyn Read + Send>,
Box::new(child.stderr.take().unwrap()) as Box<dyn Read + Send>,
] {
let tx = tx.clone();
readers.push(std::thread::spawn(move || {
let mut buf = [0u8; 16384];
let mut pending: Vec<u8> = Vec::new();
loop {
match pipe.read(&mut buf) {
Ok(0) | Err(_) => break,
Ok(n) => {
pending.extend_from_slice(&buf[..n]);
// Only hand over complete UTF-8.
let valid = match std::str::from_utf8(&pending) {
Ok(_) => pending.len(),
Err(e) => e.valid_up_to(),
};
let s = String::from_utf8_lossy(&pending[..valid]).into_owned();
pending.drain(..valid);
let _ = tx.send(Some(s));
}
}
}
let _ = tx.send(None);
}));
}
drop(tx);
let mut open = 2;
let mut killed = false;
while open > 0 {
match rx.recv_timeout(std::time::Duration::from_millis(200)) {
Ok(Some(s)) => on(&s),
Ok(None) => open -= 1,
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(_) => break,
}
if !killed && cancel.load(Ordering::Relaxed) {
unsafe {
libc::kill(-pgid, libc::SIGTERM);
}
killed = true;
}
}
for r in readers {
let _ = r.join();
}
let status = child.wait().map_err(|e| e.to_string())?;
if killed {
unsafe {
libc::kill(-pgid, libc::SIGKILL);
}
}
Ok(status.success())
}
// ---------------------------------------------------------------------------
// MkPFS output parsing
#[derive(Debug, Clone, PartialEq)]
pub enum PackEvent {
/// Percentage of the current MkPFS phase, its name, and MB/s when shown.
Progress { pct: f64, phase: String, mbps: Option<f64> },
/// Uncompressed size of the inner exFAT image, in bytes.
InnerSize(u64),
Line(String),
}
#[derive(Default)]
struct PackParser {
partial: String,
errors: Vec<String>,
}
impl PackParser {
fn feed(&mut self, chunk: &str) -> Vec<PackEvent> {
self.partial.push_str(chunk);
let mut out = Vec::new();
// MkPFS redraws its bar with '\r'; ordinary messages end in '\n'.
while let Some(i) = self.partial.find(['\r', '\n']) {
let piece: String = self.partial.drain(..=i).collect();
let piece = piece.trim_end_matches(['\r', '\n']).trim();
if piece.is_empty() {
continue;
}
if let Some(ev) = parse_bar(piece) {
out.push(ev);
continue;
}
if let Some(sz) = parse_inner_size(piece) {
out.push(PackEvent::InnerSize(sz));
}
if piece.contains("❌") || piece.to_ascii_lowercase().starts_with("error") {
self.errors.push(piece.to_string());
}
if !piece.chars().all(|c| c == '=' || c == '-') {
out.push(PackEvent::Line(piece.to_string()));
}
}
out
}
fn last_error(&self) -> Option<String> {
self.errors.last().cloned()
}
}
/// `[######-----] 21% compress @ 750.66 MB/s ETA 0s`
fn parse_bar(s: &str) -> Option<PackEvent> {
let rest = s.strip_prefix('[')?;
let close = rest.find(']')?;
let rest = rest[close + 1..].trim_start();
let pct_end = rest.find('%')?;
let pct: f64 = rest[..pct_end].trim().parse().ok()?;
let rest = rest[pct_end + 1..].trim();
let (phase, tail) = match rest.find(" @ ") {
Some(i) => (&rest[..i], Some(&rest[i + 3..])),
None => (rest, None),
};
let mbps = tail.and_then(|t| {
let num: String = t.chars().take_while(|c| c.is_ascii_digit() || *c == '.').collect();
let v: f64 = num.parse().ok()?;
let unit = t[num.len()..].trim_start();
Some(if unit.starts_with("GB/s") { v * 1000.0 } else if unit.starts_with("KB/s") { v / 1000.0 } else { v })
});
Some(PackEvent::Progress { pct, phase: phase.trim().to_string(), mbps })
}
/// `Inner image: PPSA00001.exfat (356.69 MB)` → bytes (decimal MB as MkPFS prints them).
fn parse_inner_size(s: &str) -> Option<u64> {
let rest = s.strip_prefix("Inner image:")?;
let open = rest.rfind('(')?;
let inner = rest[open + 1..].trim_end_matches(')');
let mut it = inner.split_whitespace();
let v: f64 = it.next()?.parse().ok()?;
let mult = match it.next()? {
"B" => 1.0,
"KB" => 1e3,
"MB" => 1e6,
"GB" => 1e9,
"TB" => 1e12,
_ => return None,
};
Some((v * mult) as u64)
}
// ---------------------------------------------------------------------------
// Verifier output
#[derive(Debug, Clone, PartialEq)]
pub enum VerifyEvent {
Plan { files: u64, bytes: u64 },
Progress { files: u64, bytes: u64 },
Problem { kind: String, path: String, detail: String },
Done { ok: bool, files: u64, problems: u64 },
}
impl VerifyEvent {
fn parse(line: &str) -> Option<Self> {
let v: serde_json::Value = serde_json::from_str(line).ok()?;
let n = |k: &str| v[k].as_u64().unwrap_or(0);
let s = |k: &str| v[k].as_str().unwrap_or("").to_string();
Some(match v["type"].as_str()? {
"plan" => Self::Plan { files: n("files"), bytes: n("bytes") },
"progress" => Self::Progress { files: n("files"), bytes: n("bytes") },
"problem" => Self::Problem { kind: s("kind"), path: s("path"), detail: s("detail") },
"done" => Self::Done { ok: v["ok"].as_bool().unwrap_or(false), files: n("files"), problems: n("problems") },
_ => return None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_progress_bars() {
assert_eq!(
parse_bar("[######--------------------------] 21% compress @ 750.66 MB/s ETA 0s"),
Some(PackEvent::Progress { pct: 21.0, phase: "compress".into(), mbps: Some(750.66) })
);
assert_eq!(
parse_bar("[--------------------------------] 0% compress"),
Some(PackEvent::Progress { pct: 0.0, phase: "compress".into(), mbps: None })
);
assert_eq!(parse_bar("Compressing inner exFAT (356.69 MB)..."), None);
}
#[test]
fn splits_carriage_return_redraws() {
let mut p = PackParser::default();
let evs = p.feed("Inner image: PPSA00001.exfat (356.69 MB)\n\r[#---] 5% compress @ 1.20 GB/s ETA 3s\r[##--] 10% comp");
assert_eq!(evs[0], PackEvent::InnerSize(356_690_000));
assert!(matches!(evs[1], PackEvent::Line(_)));
assert_eq!(evs[2], PackEvent::Progress { pct: 5.0, phase: "compress".into(), mbps: Some(1200.0) });
assert_eq!(evs.len(), 3, "partial bar must wait for the rest");
let evs = p.feed("ress @ 800.00 MB/s ETA 1s\r");
assert_eq!(evs, vec![PackEvent::Progress { pct: 10.0, phase: "compress".into(), mbps: Some(800.0) }]);
}
#[test]
fn parses_verifier_lines() {
assert_eq!(
VerifyEvent::parse(r#"{"type": "problem", "kind": "content", "path": "eboot.bin", "detail": "first difference at byte 12"}"#),
Some(VerifyEvent::Problem { kind: "content".into(), path: "eboot.bin".into(), detail: "first difference at byte 12".into() })
);
assert_eq!(
VerifyEvent::parse(r#"{"type": "done", "ok": true, "files": 409, "bytes": 1, "problems": 0}"#),
Some(VerifyEvent::Done { ok: true, files: 409, problems: 0 })
);
}
}
+235
View File
@@ -0,0 +1,235 @@
//! A small, blocking FTP client: just what talking to the PS5's FTP server needs.
//! Passive mode, binary transfers, resumable uploads.
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
const TIMEOUT: Duration = Duration::from_secs(30);
pub struct Ftp {
reader: BufReader<TcpStream>,
writer: TcpStream,
peer: SocketAddr,
}
#[derive(Debug)]
pub struct Reply {
pub code: u16,
pub text: String,
}
impl Ftp {
pub fn connect(host: &str, port: u16, connect_timeout: Duration) -> Result<Self, String> {
let addr = (host, port)
.to_socket_addrs()
.map_err(|e| format!("Can't resolve {host}: {e}"))?
.next()
.ok_or_else(|| format!("Can't resolve {host}"))?;
let stream = TcpStream::connect_timeout(&addr, connect_timeout)
.map_err(|e| format!("No FTP server at {host}:{port} ({e})"))?;
stream.set_read_timeout(Some(TIMEOUT)).ok();
stream.set_write_timeout(Some(TIMEOUT)).ok();
stream.set_nodelay(true).ok();
let writer = stream.try_clone().map_err(|e| e.to_string())?;
let mut ftp = Self { reader: BufReader::new(stream), writer, peer: addr };
let hello = ftp.read_reply()?;
if hello.code != 220 {
return Err(format!("Unexpected FTP greeting: {} {}", hello.code, hello.text));
}
let r = ftp.cmd("USER anonymous")?;
if r.code == 331 {
let r = ftp.cmd("PASS anonymous@")?;
if r.code != 230 && r.code != 202 {
return Err(format!("FTP login refused: {} {}", r.code, r.text));
}
} else if r.code != 230 {
return Err(format!("FTP login refused: {} {}", r.code, r.text));
}
ftp.expect("TYPE I", &[200])?;
Ok(ftp)
}
fn read_reply(&mut self) -> Result<Reply, String> {
let mut line = String::new();
let mut read_line = |line: &mut String| -> Result<(), String> {
line.clear();
let n = self.reader.read_line(line).map_err(|e| format!("FTP connection lost: {e}"))?;
if n == 0 {
return Err("FTP server closed the connection".into());
}
Ok(())
};
read_line(&mut line)?;
let code: u16 = line.get(..3).and_then(|c| c.parse().ok()).ok_or_else(|| format!("Bad FTP reply: {line:?}"))?;
let mut text = line[3..].trim_start_matches([' ', '-']).trim_end().to_string();
if line.as_bytes().get(3) == Some(&b'-') {
let end = format!("{code} ");
loop {
read_line(&mut line)?;
text.push('\n');
text.push_str(line.trim_end());
if line.starts_with(&end) {
break;
}
}
}
Ok(Reply { code, text })
}
pub fn cmd(&mut self, c: &str) -> Result<Reply, String> {
self.writer
.write_all(format!("{c}\r\n").as_bytes())
.map_err(|e| format!("FTP connection lost: {e}"))?;
self.read_reply()
}
fn expect(&mut self, c: &str, ok: &[u16]) -> Result<Reply, String> {
let r = self.cmd(c)?;
if ok.contains(&r.code) {
Ok(r)
} else {
let shown = if c.starts_with("PASS") { "PASS ***" } else { c };
Err(format!("{shown} failed: {} {}", r.code, r.text))
}
}
pub fn noop(&mut self) -> Result<(), String> {
self.expect("NOOP", &[200]).map(|_| ())
}
pub fn size(&mut self, path: &str) -> Result<Option<u64>, String> {
let r = self.cmd(&format!("SIZE {path}"))?;
Ok(if r.code == 213 { r.text.trim().parse().ok() } else { None })
}
pub fn rename(&mut self, from: &str, to: &str) -> Result<(), String> {
self.expect(&format!("RNFR {from}"), &[350])?;
self.expect(&format!("RNTO {to}"), &[250]).map(|_| ())
}
pub fn delete(&mut self, path: &str) -> Result<(), String> {
self.expect(&format!("DELE {path}"), &[250]).map(|_| ())
}
fn pasv(&mut self) -> Result<TcpStream, String> {
let r = self.expect("PASV", &[227])?;
let start = r.text.find('(').ok_or("Bad PASV reply")? + 1;
let end = r.text[start..].find(')').ok_or("Bad PASV reply")? + start;
let nums: Vec<u16> = r.text[start..end].split(',').filter_map(|n| n.trim().parse().ok()).collect();
if nums.len() != 6 {
return Err(format!("Bad PASV reply: {}", r.text));
}
let port = nums[4] * 256 + nums[5];
// Some console servers announce 0.0.0.0 or an internal address; the control peer is reliable.
let addr = SocketAddr::new(self.peer.ip(), port);
let s = TcpStream::connect_timeout(&addr, TIMEOUT).map_err(|e| format!("FTP data connection failed: {e}"))?;
s.set_read_timeout(Some(TIMEOUT)).ok();
s.set_write_timeout(Some(TIMEOUT)).ok();
Ok(s)
}
/// Downloads a (small) file. `Ok(None)` when it doesn't exist.
pub fn retr(&mut self, path: &str) -> Result<Option<Vec<u8>>, String> {
let mut data = self.pasv()?;
let r = self.cmd(&format!("RETR {path}"))?;
if r.code == 550 {
return Ok(None);
}
if r.code != 150 && r.code != 125 {
return Err(format!("RETR {path} failed: {} {}", r.code, r.text));
}
let mut buf = Vec::new();
data.read_to_end(&mut buf).map_err(|e| format!("Download of {path} failed: {e}"))?;
drop(data);
let done = self.read_reply()?;
if done.code != 226 && done.code != 250 {
return Err(format!("Download of {path} failed: {} {}", done.code, done.text));
}
Ok(Some(buf))
}
/// Names in a directory (via LIST, since NLST support varies). Hidden entries included.
pub fn list_names(&mut self, path: &str) -> Result<Vec<(String, bool)>, String> {
let mut data = self.pasv()?;
let r = self.cmd(&format!("LIST {path}"))?;
if r.code == 550 {
return Ok(Vec::new());
}
if r.code != 150 && r.code != 125 {
return Err(format!("LIST {path} failed: {} {}", r.code, r.text));
}
let mut raw = String::new();
data.read_to_string(&mut raw).map_err(|e| e.to_string())?;
drop(data);
self.read_reply()?;
Ok(raw
.lines()
.filter_map(|l| {
let parts: Vec<&str> = l.splitn(9, char::is_whitespace).filter(|p| !p.is_empty()).collect();
// Unix-style listing: perms links owner group size month day time name
let name = l.split_whitespace().skip(8).collect::<Vec<_>>().join(" ");
if parts.is_empty() || name.is_empty() || name == "." || name == ".." {
return None;
}
Some((name, l.starts_with('d')))
})
.collect())
}
/// Uploads `src` to `path`, appending from `offset` when the server already has a prefix.
pub fn store(
&mut self,
path: &str,
src: &mut dyn Read,
offset: u64,
cancel: &AtomicBool,
progress: &mut dyn FnMut(u64),
) -> Result<(), String> {
let mut data = self.pasv()?;
if offset > 0 {
self.expect(&format!("REST {offset}"), &[350])?;
}
let r = self.cmd(&format!("STOR {path}"))?;
if r.code != 150 && r.code != 125 {
return Err(format!("Upload refused: {} {}", r.code, r.text));
}
let mut buf = vec![0u8; 1 << 20];
let mut sent = offset;
loop {
if cancel.load(Ordering::Relaxed) {
let _ = data.shutdown(std::net::Shutdown::Both);
let _ = self.read_reply();
return Err("Cancelled".into());
}
let n = src.read(&mut buf).map_err(|e| format!("Reading the image failed: {e}"))?;
if n == 0 {
break;
}
data.write_all(&buf[..n]).map_err(|e| format!("Upload interrupted: {e}"))?;
sent += n as u64;
progress(sent);
}
data.flush().ok();
drop(data);
let done = self.read_reply()?;
if done.code != 226 && done.code != 250 {
return Err(format!("Upload failed: {} {}", done.code, done.text));
}
Ok(())
}
/// Writes a small file atomically: upload to a hidden temp name, then rename over.
pub fn put_atomic(&mut self, path: &str, bytes: &[u8]) -> Result<(), String> {
let (dir, name) = path.rsplit_once('/').ok_or("bad path")?;
let tmp = format!("{dir}/.{name}.packer-tmp");
let never = AtomicBool::new(false);
self.store(&tmp, &mut &bytes[..], 0, &never, &mut |_| {})?;
self.rename(&tmp, path)
}
pub fn quit(mut self) {
let _ = self.cmd("QUIT");
}
}
+141
View File
@@ -0,0 +1,141 @@
mod dump;
mod engine;
mod ftp;
mod pipeline;
mod ps5;
mod settings;
use std::path::PathBuf;
use std::sync::Arc;
use tauri::{Emitter as _, Manager, State};
pub type Emitter = Arc<dyn Fn(&str, serde_json::Value) + Send + Sync>;
struct App {
queue: pipeline::Queue,
engine: Arc<engine::Engine>,
}
async fn blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> Result<T, String> {
tauri::async_runtime::spawn_blocking(f).await.map_err(|e| e.to_string())
}
#[tauri::command]
fn get_settings() -> settings::Settings {
settings::load()
}
#[tauri::command]
fn save_settings(value: settings::Settings) -> Result<(), String> {
settings::save(&value)
}
#[tauri::command]
async fn engine_status(app: State<'_, App>) -> Result<engine::EngineStatus, String> {
let e = app.engine.clone();
blocking(move || e.status()).await
}
#[tauri::command]
async fn install_engine(app: State<'_, App>, handle: tauri::AppHandle) -> Result<engine::EngineStatus, String> {
let e = app.engine.clone();
blocking(move || {
e.install(&mut |line| {
let _ = handle.emit("engine-log", line);
})?;
Ok(e.status())
})
.await?
}
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct Inspected {
info: dump::DumpInfo,
art: dump::DumpArt,
}
#[tauri::command]
async fn inspect_dump(path: String) -> Result<Inspected, String> {
blocking(move || {
let p = PathBuf::from(&path);
let info = dump::inspect(&p)?;
let art = dump::art(&p, &info.title_id, &settings::data_dir());
Ok(Inspected { info, art })
})
.await?
}
#[tauri::command]
async fn add_job(app: State<'_, App>, path: String) -> Result<u64, String> {
let q = app.queue.clone();
blocking(move || {
let info = dump::inspect(&PathBuf::from(&path))?;
q.add(info)
})
.await?
}
#[tauri::command]
async fn job_art(path: String, title_id: String) -> Result<dump::DumpArt, String> {
blocking(move || dump::art(&PathBuf::from(path), &title_id, &settings::data_dir())).await
}
#[tauri::command]
fn list_jobs(app: State<'_, App>) -> Vec<pipeline::Job> {
app.queue.snapshot()
}
#[tauri::command]
fn cancel_job(app: State<'_, App>, id: u64) {
app.queue.cancel(id)
}
#[tauri::command]
fn remove_job(app: State<'_, App>, id: u64) -> Result<(), String> {
app.queue.remove(id)
}
#[tauri::command]
fn resume_job(app: State<'_, App>, id: u64) -> Result<(), String> {
app.queue.resume(id)
}
#[tauri::command]
async fn ps5_status() -> Result<ps5::Ps5Status, String> {
blocking(|| ps5::status(&settings::load())).await
}
#[cfg_attr(mobile, tauri::mobile_entry_point)]
pub fn run() {
tauri::Builder::default()
.plugin(tauri_plugin_dialog::init())
.plugin(tauri_plugin_opener::init())
.setup(|app| {
let handle = app.handle().clone();
let emitter: Emitter = Arc::new(move |event, payload| {
let _ = handle.emit(event, payload);
});
let engine = Arc::new(engine::Engine::new(&settings::data_dir()));
let _ = engine.write_verifier();
let queue = pipeline::Queue::new(emitter, engine.clone());
app.manage(App { queue, engine });
Ok(())
})
.invoke_handler(tauri::generate_handler![
get_settings,
save_settings,
engine_status,
install_engine,
inspect_dump,
add_job,
job_art,
list_jobs,
cancel_job,
remove_job,
resume_job,
ps5_status
])
.run(tauri::generate_context!())
.expect("error while running Packer");
}
+5
View File
@@ -0,0 +1,5 @@
#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
fn main() {
ps5_packer_lib::run()
}
+925
View File
@@ -0,0 +1,925 @@
//! The job queue: compress → verify → upload → mount, one game at a time.
use crate::dump::DumpInfo;
use crate::engine::{Engine, PackEvent, VerifyEvent};
use crate::settings::{self, Settings};
use crate::{ps5, Emitter};
use serde::Serialize;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
pub const STAGES: [&str; 4] = ["compress", "verify", "upload", "mount"];
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum JobStatus {
Queued,
Running,
/// Verified, waiting for the user to start the upload.
Ready,
Done,
Failed,
Cancelled,
}
#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum StageState {
Pending,
Live,
Done,
Warn,
Failed,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StageInfo {
pub key: &'static str,
pub state: StageState,
pub meta: String,
pub pct: f64,
pub secs: u64,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct LogLine {
pub t: u64,
pub text: String,
/// "", "ok", "warn", "err", "hl"
pub tone: &'static str,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Problem {
pub kind: String,
pub path: String,
pub detail: String,
}
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Job {
pub id: u64,
pub dump: DumpInfo,
pub status: JobStatus,
pub stage: usize,
pub stages: Vec<StageInfo>,
/// Progress of the current stage, 0..100.
pub pct: f64,
pub done_bytes: u64,
pub total_bytes: u64,
pub files_done: u64,
pub files_total: u64,
/// Bytes per second, smoothed.
pub rate: f64,
pub rate_history: Vec<f64>,
pub rate_label: &'static str,
pub eta_secs: Option<u64>,
pub stage_secs: u64,
pub image_path: Option<String>,
pub image_bytes: Option<u64>,
pub remote_path: Option<String>,
pub mismatches: u64,
pub problems: Vec<Problem>,
pub warnings: Vec<String>,
pub error: Option<String>,
pub log: Vec<LogLine>,
#[serde(skip)]
resume_from: usize,
}
impl Job {
fn new(id: u64, dump: DumpInfo) -> Self {
Self {
id,
dump,
status: JobStatus::Queued,
stage: 0,
stages: STAGES
.iter()
.map(|k| StageInfo { key: k, state: StageState::Pending, meta: String::new(), pct: 0.0, secs: 0 })
.collect(),
pct: 0.0,
done_bytes: 0,
total_bytes: 0,
files_done: 0,
files_total: 0,
rate: 0.0,
rate_history: Vec::new(),
rate_label: "",
eta_secs: None,
stage_secs: 0,
image_path: None,
image_bytes: None,
remote_path: None,
mismatches: 0,
problems: Vec::new(),
warnings: Vec::new(),
error: None,
log: Vec::new(),
resume_from: 0,
}
}
fn log(&mut self, tone: &'static str, text: impl Into<String>) {
self.log.push(LogLine { t: now(), text: text.into(), tone });
if self.log.len() > 500 {
self.log.drain(..100);
}
}
}
fn now() -> u64 {
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
}
struct State {
jobs: Vec<Job>,
next_id: u64,
cancel: Option<(u64, Arc<AtomicBool>)>,
}
#[derive(Clone)]
pub struct Queue {
inner: Arc<(Mutex<State>, Condvar)>,
emitter: Emitter,
engine: Arc<Engine>,
}
/// Smooths a byte counter into a rate and keeps a short history for the sparkline.
struct Meter {
last_t: Instant,
last_bytes: u64,
last_sample: Instant,
rate: f64,
}
impl Meter {
fn new(start_bytes: u64) -> Self {
let t = Instant::now();
Self { last_t: t, last_bytes: start_bytes, last_sample: t, rate: 0.0 }
}
/// Returns Some(rate) once per ~second for the history graph.
fn update(&mut self, bytes: u64) -> Option<f64> {
let dt = self.last_t.elapsed().as_secs_f64();
if dt >= 0.25 {
let inst = bytes.saturating_sub(self.last_bytes) as f64 / dt;
self.rate = if self.rate == 0.0 { inst } else { self.rate * 0.7 + inst * 0.3 };
self.last_t = Instant::now();
self.last_bytes = bytes;
}
if self.last_sample.elapsed() >= Duration::from_secs(1) {
self.last_sample = Instant::now();
return Some(self.rate);
}
None
}
}
impl Queue {
pub fn new(emitter: Emitter, engine: Arc<Engine>) -> Self {
let q = Self {
inner: Arc::new((Mutex::new(State { jobs: Vec::new(), next_id: 1, cancel: None }), Condvar::new())),
emitter,
engine,
};
let worker = q.clone();
std::thread::Builder::new().name("packer-worker".into()).spawn(move || worker.run()).unwrap();
q
}
pub fn snapshot(&self) -> Vec<Job> {
self.inner.0.lock().unwrap().jobs.clone()
}
fn emit(&self) {
let jobs = self.snapshot();
(self.emitter)("jobs", serde_json::to_value(jobs).unwrap());
}
fn with<R>(&self, id: u64, f: impl FnOnce(&mut Job) -> R) -> Option<R> {
let mut st = self.inner.0.lock().unwrap();
st.jobs.iter_mut().find(|j| j.id == id).map(f)
}
pub fn add(&self, dump: DumpInfo) -> Result<u64, String> {
let id = {
let mut st = self.inner.0.lock().unwrap();
if let Some(j) = st.jobs.iter().find(|j| {
j.dump.title_id == dump.title_id
&& matches!(j.status, JobStatus::Queued | JobStatus::Running | JobStatus::Ready)
}) {
return Err(format!("{} is already in the queue", j.dump.title_name));
}
let id = st.next_id;
st.next_id += 1;
let mut job = Job::new(id, dump);
job.log("hl", format!("Added {}", job.dump.path));
if job.dump.apr && !job.dump.has_ampr_index {
job.warnings.push(
"APR Emu game without ampr_emu.index: MkPFS will create that file inside the dump folder.".into(),
);
}
st.jobs.push(job);
id
};
self.inner.1.notify_all();
self.emit();
Ok(id)
}
pub fn remove(&self, id: u64) -> Result<(), String> {
{
let mut st = self.inner.0.lock().unwrap();
let Some(pos) = st.jobs.iter().position(|j| j.id == id) else { return Ok(()) };
if st.jobs[pos].status == JobStatus::Running {
return Err("Cancel it first".into());
}
st.jobs.remove(pos);
}
self.emit();
Ok(())
}
pub fn cancel(&self, id: u64) {
let flag = {
let mut st = self.inner.0.lock().unwrap();
if let Some(j) = st.jobs.iter_mut().find(|j| j.id == id) {
if matches!(j.status, JobStatus::Queued | JobStatus::Ready) {
j.status = JobStatus::Cancelled;
j.log("warn", "Cancelled");
}
}
st.cancel.as_ref().filter(|(cid, _)| *cid == id).map(|(_, f)| f.clone())
};
if let Some(f) = flag {
f.store(true, Ordering::Relaxed);
}
self.emit();
}
/// Re-queues a failed/cancelled job from the stage that didn't finish,
/// or starts the upload of a verified job.
pub fn resume(&self, id: u64) -> Result<(), String> {
self.with(id, |j| {
if !matches!(j.status, JobStatus::Failed | JobStatus::Cancelled | JobStatus::Ready) {
return Err("Nothing to resume".to_string());
}
let from = j.stages.iter().position(|s| s.state != StageState::Done).unwrap_or(STAGES.len());
// Without the local image there is nothing to verify or upload.
let from = if from > 0 && !j.image_path.as_ref().is_some_and(|p| Path::new(p).is_file()) { 0 } else { from };
j.resume_from = from;
j.status = JobStatus::Queued;
j.error = None;
for s in j.stages.iter_mut().skip(from) {
s.state = StageState::Pending;
s.pct = 0.0;
}
if j.status == JobStatus::Queued && from == 2 {
j.log("hl", "Upload requested");
}
Ok(())
})
.unwrap_or(Err("Unknown job".into()))?;
self.inner.1.notify_all();
self.emit();
Ok(())
}
fn run(self) {
loop {
let (id, cancel) = {
let (lock, cv) = &*self.inner;
let mut st = lock.lock().unwrap();
loop {
if let Some(j) = st.jobs.iter_mut().find(|j| j.status == JobStatus::Queued) {
j.status = JobStatus::Running;
let id = j.id;
let flag = Arc::new(AtomicBool::new(false));
st.cancel = Some((id, flag.clone()));
break (id, flag);
}
st = cv.wait(st).unwrap();
}
};
self.emit();
self.process(id, &cancel);
self.inner.0.lock().unwrap().cancel = None;
self.emit();
}
}
fn process(&self, id: u64, cancel: &Arc<AtomicBool>) {
let s = settings::load();
let from = self.with(id, |j| j.resume_from).unwrap_or(0);
for stage in from..STAGES.len() {
self.enter_stage(id, stage);
let started = Instant::now();
let result = match stage {
0 => self.compress(id, &s, cancel),
1 => self.verify(id, &s, cancel),
2 => self.upload(id, &s, cancel),
_ => self.mount(id, &s, cancel),
};
let secs = started.elapsed().as_secs();
match result {
Ok(()) => {
self.with(id, |j| {
let st = &mut j.stages[stage];
if st.state == StageState::Live {
st.state = StageState::Done;
}
st.pct = 100.0;
st.secs = secs;
});
if stage == 1 && !s.auto_upload {
self.with(id, |j| {
j.status = JobStatus::Ready;
j.resume_from = 2;
j.log("ok", "Verified. Waiting for you to start the upload.");
});
return;
}
}
Err(e) => {
let cancelled = cancel.load(Ordering::Relaxed);
self.with(id, |j| {
j.stages[stage].state = if cancelled { StageState::Pending } else { StageState::Failed };
j.stages[stage].secs = secs;
j.status = if cancelled { JobStatus::Cancelled } else { JobStatus::Failed };
j.rate = 0.0;
j.eta_secs = None;
if cancelled {
j.log("warn", "Cancelled");
} else {
j.log("err", e.clone());
j.error = Some(e);
}
});
return;
}
}
}
self.with(id, |j| {
j.status = JobStatus::Done;
j.rate = 0.0;
j.eta_secs = None;
j.stage = STAGES.len() - 1;
j.log("ok", "Finished");
});
}
fn enter_stage(&self, id: u64, stage: usize) {
self.with(id, |j| {
j.stage = stage;
j.stages[stage].state = StageState::Live;
j.stages[stage].pct = 0.0;
j.pct = 0.0;
j.done_bytes = 0;
j.total_bytes = 0;
j.files_done = 0;
j.files_total = 0;
j.rate = 0.0;
j.rate_history.clear();
j.eta_secs = None;
j.stage_secs = 0;
j.log("hl", format!("{} started", title_case(STAGES[stage])));
});
self.emit();
}
/// Shared progress bookkeeping; emits at most ~8 times a second.
fn progress(&self, id: u64, last_emit: &mut Instant, started: Instant, f: impl FnOnce(&mut Job)) {
self.with(id, |j| {
f(j);
j.stage_secs = started.elapsed().as_secs();
j.stages[j.stage].pct = j.pct;
if j.rate > 0.0 && j.total_bytes > j.done_bytes {
j.eta_secs = Some(((j.total_bytes - j.done_bytes) as f64 / j.rate) as u64);
}
});
if last_emit.elapsed() >= Duration::from_millis(120) {
*last_emit = Instant::now();
self.emit();
}
}
// -- stages -------------------------------------------------------------
fn compress(&self, id: u64, s: &Settings, cancel: &Arc<AtomicBool>) -> Result<(), String> {
let dump = self.with(id, |j| j.dump.clone()).unwrap();
let out_dir = PathBuf::from(&s.output_dir);
std::fs::create_dir_all(&out_dir).map_err(|e| format!("Can't create {}: {e}", out_dir.display()))?;
let final_path = out_dir.join(format!("{}.ffpfsc", dump.title_id));
if final_path.is_file() {
let size = std::fs::metadata(&final_path).map(|m| m.len()).unwrap_or(0);
self.with(id, |j| {
j.image_path = Some(final_path.to_string_lossy().into_owned());
j.image_bytes = Some(size);
j.stages[0].meta = format!("Reusing existing image · {}", gb(size));
j.log("warn", format!("{} already exists; verifying it instead of compressing again", final_path.display()));
});
return Ok(());
}
let free = free_space(&out_dir).unwrap_or(u64::MAX);
if free < dump.total_bytes / 3 {
return Err(format!(
"Only {} free in {}. The image could need up to {}.",
gb(free),
out_dir.display(),
gb(dump.total_bytes)
));
}
if free < dump.total_bytes {
self.with(id, |j| {
j.warnings.push(format!(
"{} free in the output folder; that's enough only if the game compresses well.",
gb(free)
))
});
}
let work = out_dir.join(format!(".work-{}", dump.title_id));
let _ = std::fs::remove_dir_all(&work);
std::fs::create_dir_all(&work).map_err(|e| e.to_string())?;
let work_img = work.join(format!("{}.ffpfsc", dump.title_id));
let started = Instant::now();
let mut last = Instant::now();
let mut meter = Meter::new(0);
let mut inner = dump.total_bytes;
self.with(id, |j| {
j.total_bytes = inner;
j.rate_label = "compress";
j.log("", format!("zlib level {} · writing to {}", s.compression_level, out_dir.display()));
});
let res = self.engine.pack(Path::new(&dump.path), &work_img, s.compression_level, s.cpu_count, cancel, &mut |ev| {
match ev {
PackEvent::InnerSize(n) => inner = n,
PackEvent::Progress { pct, .. } => {
let done = (inner as f64 * pct / 100.0) as u64;
let sample = meter.update(done);
let rate = meter.rate;
self.progress(id, &mut last, started, |j| {
j.pct = pct;
j.total_bytes = inner;
j.done_bytes = done;
j.rate = rate;
if let Some(r) = sample {
push_hist(&mut j.rate_history, r);
}
j.stages[0].meta = format!("{:.0}% of {}", pct, gb(inner));
});
}
PackEvent::Line(l) => {
if is_interesting(&l) {
self.with(id, |j| j.log("", l));
}
}
}
});
if let Err(e) = res {
let _ = std::fs::remove_dir_all(&work);
return Err(e);
}
std::fs::rename(&work_img, &final_path).map_err(|e| format!("Couldn't move the image into place: {e}"))?;
let _ = std::fs::remove_dir_all(&work);
let size = std::fs::metadata(&final_path).map(|m| m.len()).unwrap_or(0);
let secs = started.elapsed().as_secs();
self.with(id, |j| {
j.image_path = Some(final_path.to_string_lossy().into_owned());
j.image_bytes = Some(size);
let saved = j.dump.total_bytes.saturating_sub(size);
j.stages[0].meta = format!("{} → {} · {}", gb(j.dump.total_bytes), gb(size), dur(secs));
j.log(
"ok",
format!(
"Compressed to {} ({}% smaller) in {}",
gb(size),
(saved as f64 * 100.0 / j.dump.total_bytes.max(1) as f64).round(),
dur(secs)
),
);
});
Ok(())
}
fn verify(&self, id: u64, s: &Settings, cancel: &Arc<AtomicBool>) -> Result<(), String> {
let (image, src) = self.with(id, |j| (j.image_path.clone(), j.dump.path.clone())).unwrap();
let image = image.ok_or("No image to verify")?;
let started = Instant::now();
let mut last = Instant::now();
let mut meter = Meter::new(0);
self.with(id, |j| {
j.mismatches = 0;
j.problems.clear();
j.rate_label = "verify";
j.log("", "Comparing every file in the image with the dump, byte for byte");
});
let ok = self.engine.verify(Path::new(&image), Path::new(&src), s.verify_workers, cancel, &mut |ev| match ev {
VerifyEvent::Plan { files, bytes } => {
self.with(id, |j| {
j.files_total = files;
j.total_bytes = bytes;
});
}
VerifyEvent::Progress { files, bytes } => {
let sample = meter.update(bytes);
let rate = meter.rate;
self.progress(id, &mut last, started, |j| {
j.files_done = files;
j.done_bytes = bytes;
j.pct = if j.total_bytes > 0 { bytes as f64 * 100.0 / j.total_bytes as f64 } else { 0.0 };
j.rate = rate;
if let Some(r) = sample {
push_hist(&mut j.rate_history, r);
}
j.stages[1].meta = format!("{} of {} files · {} mismatches", fmt_n(files), fmt_n(j.files_total), j.mismatches);
});
}
VerifyEvent::Problem { kind, path, detail } => {
self.with(id, |j| {
j.mismatches += 1;
j.log("err", format!("{kind}: {path} ({detail})"));
if j.problems.len() < 200 {
j.problems.push(Problem { kind, path, detail });
}
});
}
VerifyEvent::Done { .. } => {}
})?;
let secs = started.elapsed().as_secs();
if !ok {
let n = self.with(id, |j| j.mismatches).unwrap_or(0);
return Err(format!("Verification failed: {n} problem(s). The image does not match the dump, so it won't be uploaded."));
}
self.with(id, |j| {
j.pct = 100.0;
j.stages[1].meta = format!("{} files · all match · {}", fmt_n(j.files_total), dur(secs));
j.log("ok", format!("Verify passed: {} files, 0 mismatches", fmt_n(j.files_total)));
});
Ok(())
}
fn upload(&self, id: u64, s: &Settings, cancel: &Arc<AtomicBool>) -> Result<(), String> {
let (image, title) = self.with(id, |j| (j.image_path.clone(), j.dump.title_id.clone())).unwrap();
let image = PathBuf::from(image.ok_or("No image to upload")?);
let local_size = std::fs::metadata(&image).map_err(|e| format!("Image missing: {e}"))?.len();
let dir = s.remote_dir.trim_end_matches('/').to_string();
let final_remote = format!("{dir}/{title}.ffpfsc");
let tmp_remote = format!("{dir}/.{title}.ffpfsc.packer-part");
let mut ftp = ps5::connect(s)?;
let others = ps5::competitors(&mut ftp, &dir, &title).unwrap_or_default();
if !others.is_empty() {
let msg = format!(
"{dir} also has {}. ShadowMountPlus may mount that instead of the new image; remove it once the image works.",
others.join(", ")
);
self.with(id, |j| {
j.warnings.push(msg.clone());
j.log("warn", msg);
});
}
if let Some(existing) = ftp.size(&final_remote)? {
if existing == local_size {
self.with(id, |j| {
j.remote_path = Some(final_remote.clone());
j.stages[2].meta = "Already on the PS5".into();
j.log("ok", format!("{final_remote} is already on the PS5 with the same size; skipping upload"));
});
return Ok(());
}
return Err(format!(
"The PS5 already has a different {title}.ffpfsc ({}). Delete or rename it first; Packer never overwrites.",
gb(existing)
));
}
let started = Instant::now();
let mut last = Instant::now();
self.with(id, |j| {
j.total_bytes = local_size;
j.rate_label = "upload";
j.log("", format!("{} → {}:{}", image.display(), s.ps5_host, final_remote));
});
let mut attempt = 0;
loop {
let offset = match ftp.size(&tmp_remote)? {
Some(n) if n <= local_size => n,
Some(_) => {
ftp.delete(&tmp_remote)?;
0
}
None => 0,
};
if offset >= local_size {
break;
}
if offset > 0 {
self.with(id, |j| j.log("", format!("Resuming at {}", gb(offset))));
}
let mut f = std::fs::File::open(&image).map_err(|e| e.to_string())?;
std::io::Seek::seek(&mut f, std::io::SeekFrom::Start(offset)).map_err(|e| e.to_string())?;
let mut meter = Meter::new(offset);
let res = ftp.store(&tmp_remote, &mut f, offset, cancel, &mut |sent| {
let sample = meter.update(sent);
let rate = meter.rate;
self.progress(id, &mut last, started, |j| {
j.done_bytes = sent;
j.pct = sent as f64 * 100.0 / local_size.max(1) as f64;
j.rate = rate;
if let Some(r) = sample {
push_hist(&mut j.rate_history, r);
}
j.stages[2].meta = format!("{} of {}", gb(sent), gb(local_size));
});
});
match res {
Ok(()) => break,
Err(e) if cancel.load(Ordering::Relaxed) => return Err(e),
Err(e) => {
attempt += 1;
if attempt > 5 {
return Err(format!("{e} (gave up after 5 retries; Retry resumes where it stopped)"));
}
self.with(id, |j| j.log("warn", format!("{e}; reconnecting ({attempt}/5)")));
std::thread::sleep(Duration::from_secs(2 * attempt));
ftp = ps5::connect(s)?;
}
}
}
let remote_size = ftp.size(&tmp_remote)?.unwrap_or(0);
if remote_size != local_size {
return Err(format!(
"Size check failed after upload: PS5 has {remote_size} bytes, image is {local_size}. Retry will resume."
));
}
ftp.rename(&tmp_remote, &final_remote)?;
if ftp.size(&final_remote)? != Some(local_size) {
return Err("The image was renamed but its size on the PS5 doesn't match".into());
}
ftp.quit();
let secs = started.elapsed().as_secs();
self.with(id, |j| {
j.remote_path = Some(final_remote.clone());
j.stages[2].meta = format!("{} · {}", gb(local_size), dur(secs));
j.log("ok", format!("Uploaded {} in {} ({}/s)", gb(local_size), dur(secs), gb(local_size / secs.max(1))));
});
Ok(())
}
fn mount(&self, id: u64, s: &Settings, cancel: &Arc<AtomicBool>) -> Result<(), String> {
let (title, remote, image) =
self.with(id, |j| (j.dump.title_id.clone(), j.remote_path.clone(), j.image_path.clone())).unwrap();
let remote = remote.ok_or("Not uploaded yet")?;
self.with(id, |j| j.rate_label = "");
if s.write_smp_hints {
let mut ftp = ps5::connect(s)?;
match ps5::write_hints(&mut ftp, &title, &settings::data_dir().join("backups")) {
Ok(r) => self.with(id, |j| {
if r.changed.is_empty() {
j.log("", "ShadowMountPlus settings already correct");
}
for c in r.changed {
j.log("", format!("ShadowMountPlus {c}"));
}
for b in r.backups {
j.log("", format!("Backup of the previous file: {b}"));
}
}),
Err(e) => self.with(id, |j| {
j.warnings.push(format!("Couldn't write ShadowMountPlus settings: {e}"));
j.log("warn", format!("Couldn't write ShadowMountPlus settings: {e}"));
}),
};
ftp.quit();
}
let started = Instant::now();
let mut last = Instant::now();
let mounted = ps5::wait_for_mount(s, &title, &remote, Duration::from_secs(120), cancel, &mut |secs| {
self.progress(id, &mut last, started, |j| {
j.pct = (secs as f64 / 120.0 * 100.0).min(99.0);
j.stages[3].meta = format!("Waiting for ShadowMountPlus · {secs}s");
});
})?;
self.with(id, |j| {
if mounted {
j.stages[3].meta = "Mounted by ShadowMountPlus".into();
j.log("ok", "ShadowMountPlus mounted the image. It's ready to play.");
} else {
j.stages[3].state = StageState::Warn;
j.stages[3].meta = "Not confirmed within 2 minutes".into();
j.warnings.push(
"ShadowMountPlus hasn't mounted the image yet. It scans every 15 s; check the PS5, or restart ShadowMountPlus."
.into(),
);
j.log("warn", "Mount not confirmed within 2 minutes");
}
});
if mounted && s.delete_local_after_upload {
if let Some(p) = image {
if std::fs::remove_file(&p).is_ok() {
self.with(id, |j| j.log("", format!("Deleted local image {p}")));
}
}
}
Ok(())
}
}
fn push_hist(h: &mut Vec<f64>, v: f64) {
h.push(v);
if h.len() > 90 {
h.remove(0);
}
}
fn is_interesting(l: &str) -> bool {
let l = l.trim();
!l.is_empty()
&& !l.starts_with("MkPFS ")
&& (l.starts_with("Wrapping")
|| l.starts_with("Inner image")
|| l.starts_with("Compressing")
|| l.contains("ampr")
|| l.contains("AMPR")
|| l.starts_with("Final image size")
|| l.starts_with("Actual gain")
|| l.starts_with("Throughput")
|| l.contains("❌")
|| l.contains("⚠")
|| l.to_ascii_lowercase().contains("error")
|| l.to_ascii_lowercase().contains("warning"))
}
fn free_space(p: &Path) -> Option<u64> {
let c = std::ffi::CString::new(p.to_string_lossy().as_bytes()).ok()?;
let mut st: libc::statvfs = unsafe { std::mem::zeroed() };
(unsafe { libc::statvfs(c.as_ptr(), &mut st) } == 0).then(|| st.f_bavail as u64 * st.f_frsize as u64)
}
pub fn gb(b: u64) -> String {
let g = b as f64 / 1e9;
if g >= 100.0 {
format!("{g:.0} GB")
} else if g >= 1.0 {
format!("{g:.1} GB")
} else {
format!("{:.0} MB", b as f64 / 1e6)
}
}
fn dur(secs: u64) -> String {
if secs >= 3600 {
format!("{}h {:02}m", secs / 3600, secs % 3600 / 60)
} else if secs >= 60 {
format!("{}m {:02}s", secs / 60, secs % 60)
} else {
format!("{secs}s")
}
}
fn fmt_n(n: u64) -> String {
let s = n.to_string();
let mut out = String::new();
for (i, c) in s.chars().enumerate() {
if i > 0 && (s.len() - i) % 3 == 0 {
out.push(',');
}
out.push(c);
}
out
}
fn title_case(s: &str) -> String {
let mut c = s.chars();
c.next().map(|f| f.to_uppercase().collect::<String>() + c.as_str()).unwrap_or_default()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn formats() {
assert_eq!(fmt_n(166733), "166,733");
assert_eq!(fmt_n(12), "12");
assert_eq!(gb(148_900_000_000), "149 GB");
assert_eq!(gb(71_733_805_056), "71.7 GB");
assert_eq!(dur(372), "6m 12s");
}
}
/// End-to-end run against a local fake PS5 (FTP + fake ShadowMountPlus). Never touches a real
/// console. Run with: PACKER_E2E_SCRATCH=<dir with mkvenv, fake_ps5.py, fakedump> cargo test e2e -- --ignored
#[cfg(test)]
mod e2e {
use super::*;
fn wait_until(q: &Queue, id: u64, secs: u64) -> Job {
let t = Instant::now();
loop {
let j = q.snapshot().into_iter().find(|j| j.id == id).unwrap();
if matches!(j.status, JobStatus::Done | JobStatus::Failed | JobStatus::Cancelled | JobStatus::Ready) {
return j;
}
assert!(t.elapsed().as_secs() < secs, "timed out in stage {}", j.stage);
std::thread::sleep(Duration::from_millis(200));
}
}
#[test]
#[ignore]
fn e2e_local_fake_ps5() {
let scratch = PathBuf::from(std::env::var("PACKER_E2E_SCRATCH").expect("PACKER_E2E_SCRATCH"));
let tmp = tempfile::tempdir().unwrap();
std::env::set_var("XDG_CONFIG_HOME", tmp.path().join("config"));
std::env::set_var("XDG_DATA_HOME", tmp.path().join("data"));
let engine_dir = tmp.path().join("data/ps5-packer/engine");
std::fs::create_dir_all(&engine_dir).unwrap();
std::os::unix::fs::symlink(scratch.join("mkvenv"), engine_dir.join("venv")).unwrap();
// Fresh copy of the fake dump so MkPFS's ampr_emu.index write can't leak between runs.
let dump = tmp.path().join("PPSA00001-app");
assert!(std::process::Command::new("cp")
.args(["-r", scratch.join("fakedump/PPSA00001-app").to_str().unwrap(), dump.to_str().unwrap()])
.status()
.unwrap()
.success());
let _ = std::fs::remove_file(dump.join("ampr_emu.index"));
let ps5_root = tmp.path().join("ps5");
let port = 22121;
let mut server = std::process::Command::new(scratch.join("mkvenv/bin/python"))
.arg(scratch.join("fake_ps5.py"))
.arg(&ps5_root)
.arg(port.to_string())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
std::thread::sleep(Duration::from_millis(1500));
let s = Settings {
ps5_host: "127.0.0.1".into(),
ftp_port: port,
web_port: 1,
output_dir: tmp.path().join("images").to_string_lossy().into_owned(),
..Settings::default()
};
settings::save(&s).unwrap();
let engine = Arc::new(Engine::new(&settings::data_dir()));
let q = Queue::new(Arc::new(|_, _| {}), engine);
let id = q.add(crate::dump::inspect(&dump).unwrap()).unwrap();
let j = wait_until(&q, id, 300);
for l in &j.log {
eprintln!("[{}] {}", l.tone, l.text);
}
assert_eq!(j.status, JobStatus::Done, "error: {:?}", j.error);
assert!(j.stages.iter().all(|s| s.state == StageState::Done), "{:?}", j.stages);
let local = std::fs::metadata(j.image_path.as_ref().unwrap()).unwrap().len();
let remote = ps5_root.join("data/homebrew/PPSA00001.ffpfsc");
assert_eq!(std::fs::metadata(&remote).unwrap().len(), local);
assert!(!ps5_root.join("data/homebrew/.PPSA00001.ffpfsc.packer-part").exists());
let cfg = std::fs::read_to_string(ps5_root.join("data/shadowmount/config.ini")).unwrap();
assert!(cfg.contains("image_ro=OTHER.exfat") && cfg.contains("image_ro=PPSA00001.exfat"), "{cfg}");
let at = std::fs::read_to_string(ps5_root.join("data/shadowmount/autotune.ini")).unwrap();
assert_eq!(at, "image_sector=PPSA00001.ffpfsc:65536\n");
assert!(dump.join("ampr_emu.index").is_file(), "MkPFS should have generated the AMPR index");
// Same title again: image reused locally, upload skipped as identical.
q.remove(id).unwrap();
let id2 = q.add(crate::dump::inspect(&dump).unwrap()).unwrap();
let j2 = wait_until(&q, id2, 300);
assert_eq!(j2.status, JobStatus::Done, "error: {:?}", j2.error);
assert!(j2.log.iter().any(|l| l.text.contains("already on the PS5")));
// A tampered image must fail verification and never reach the upload.
q.remove(id2).unwrap();
{
use std::io::{Seek, SeekFrom, Write};
let mut f = std::fs::OpenOptions::new().write(true).open(j.image_path.as_ref().unwrap()).unwrap();
f.seek(SeekFrom::Start(local / 2)).unwrap();
f.write_all(&[0xde, 0xad, 0xbe, 0xef, 0x00, 0xff]).unwrap();
}
std::fs::remove_file(&remote).unwrap();
let id3 = q.add(crate::dump::inspect(&dump).unwrap()).unwrap();
let j3 = wait_until(&q, id3, 300);
assert_eq!(j3.status, JobStatus::Failed);
assert_eq!(j3.stage, 1, "must stop at verify");
assert!(j3.mismatches > 0);
assert!(!remote.exists(), "a failed image must not be uploaded");
server.kill().ok();
}
}
+236
View File
@@ -0,0 +1,236 @@
//! Everything that talks to the console: connection checks, ShadowMountPlus hints, mount checks.
use crate::ftp::Ftp;
use crate::settings::Settings;
use serde::Serialize;
use std::io::{Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::Path;
use std::time::{Duration, Instant};
pub const SMP_DIR: &str = "/data/shadowmount";
/// Game Compressor writes the same value for PFSC containers.
pub const PFSC_SECTOR: u32 = 65536;
#[derive(Debug, Clone, Serialize, Default)]
#[serde(rename_all = "camelCase")]
pub struct Ps5Status {
pub online: bool,
pub latency_ms: Option<u64>,
pub error: Option<String>,
pub free_bytes: Option<u64>,
pub total_bytes: Option<u64>,
pub shadowmount: bool,
}
pub fn connect(s: &Settings) -> Result<Ftp, String> {
Ftp::connect(&s.ps5_host, s.ftp_port, Duration::from_secs(4))
}
/// Read-only check: FTP login + NOOP, ShadowMountPlus config present, and free space from
/// the web file manager if it's running.
pub fn status(s: &Settings) -> Ps5Status {
let t0 = Instant::now();
let mut ftp = match connect(s) {
Ok(f) => f,
Err(e) => return Ps5Status { error: Some(e), ..Default::default() },
};
if let Err(e) = ftp.noop() {
return Ps5Status { error: Some(e), ..Default::default() };
}
let latency_ms = Some(t0.elapsed().as_millis() as u64);
let shadowmount = ftp.size(&format!("{SMP_DIR}/config.ini")).ok().flatten().is_some();
ftp.quit();
let (free_bytes, total_bytes) = space(s).unwrap_or((None, None));
Ps5Status { online: true, latency_ms, error: None, free_bytes, total_bytes, shadowmount }
}
/// GET /api/space?path=/data on the PS5 web file manager. Optional; a missing server is fine.
fn space(s: &Settings) -> Option<(Option<u64>, Option<u64>)> {
let body = http_get(&s.ps5_host, s.web_port, "/api/space?path=/data")?;
let v: serde_json::Value = serde_json::from_str(&body).ok()?;
let sp = v["spaces"].as_array()?.iter().find(|x| x["path"] == "/data")?;
Some((sp["free"].as_u64(), sp["total"].as_u64()))
}
fn http_get(host: &str, port: u16, path: &str) -> Option<String> {
let addr = (host, port).to_socket_addrs().ok()?.next()?;
let mut s = TcpStream::connect_timeout(&addr, Duration::from_secs(2)).ok()?;
s.set_read_timeout(Some(Duration::from_secs(4))).ok();
write!(s, "GET {path} HTTP/1.0\r\nHost: {host}\r\nAccept-Encoding: identity\r\nConnection: close\r\n\r\n").ok()?;
let mut raw = Vec::new();
s.read_to_end(&mut raw).ok()?;
let raw = String::from_utf8_lossy(&raw);
let (head, body) = raw.split_once("\r\n\r\n")?;
head.starts_with("HTTP/1.").then(|| ())?;
head.split_whitespace().nth(1).filter(|c| *c == "200")?;
Some(body.to_string())
}
/// Other sources for the same title already in the upload folder (e.g. a loose `-app` folder),
/// which ShadowMountPlus could mount instead of the new image.
pub fn competitors(ftp: &mut Ftp, remote_dir: &str, title_id: &str) -> Result<Vec<String>, String> {
Ok(ftp
.list_names(remote_dir)?
.into_iter()
.map(|(n, _)| n)
.filter(|n| {
n.len() >= 9
&& n[..9].eq_ignore_ascii_case(title_id)
&& !n.starts_with('.')
&& n != &format!("{title_id}.ffpfsc")
&& !n.ends_with(".vhash")
})
.collect())
}
// ---------------------------------------------------------------------------
// ShadowMountPlus hints, matching what Game Compressor writes for a compressed image:
// autotune.ini: image_sector=<TITLE>.ffpfsc:65536
// config.ini: image_ro=<TITLE>.exfat
// Both files are flat key=value; names are basenames compared case-insensitively.
/// Removes `keys` lines whose image name is in `names`, then appends `add` (if any).
/// Returns `None` when nothing would change.
pub fn upsert_hint(content: &str, keys: &[&str], names: &[String], add: Option<&str>) -> Option<String> {
let mut out: Vec<String> = Vec::new();
let mut have_add = false;
for line in content.lines() {
let trimmed = line.trim();
if Some(trimmed) == add && !have_add {
have_add = true;
out.push(line.to_string());
continue;
}
if let Some((k, v)) = trimmed.split_once('=') {
let k = k.trim();
if keys.contains(&k) {
let v = v.trim();
// image_sector values are NAME:SECTOR (split on the last colon).
let name = if k == "image_sector" { v.rsplit_once(':').map(|x| x.0).unwrap_or(v) } else { v };
let name = name.rsplit('/').next().unwrap_or(name).trim();
if names.iter().any(|n| n.eq_ignore_ascii_case(name)) {
continue;
}
}
}
out.push(line.to_string());
}
if let (Some(a), false) = (add, have_add) {
out.push(a.to_string());
}
let mut next = out.join("\n");
if !next.is_empty() {
next.push('\n');
}
let same = next.trim_end() == content.trim_end();
(!same).then_some(next)
}
pub struct HintResult {
pub changed: Vec<String>,
pub backups: Vec<String>,
}
pub fn write_hints(ftp: &mut Ftp, title_id: &str, backup_dir: &Path) -> Result<HintResult, String> {
let outer = format!("{title_id}.ffpfsc");
let nested = format!("{title_id}.exfat");
let names = vec![outer.clone(), nested.clone(), format!("{title_id}.pfs")];
let plan: [(&str, Vec<&str>, String); 2] = [
("autotune.ini", vec!["image_sector"], format!("image_sector={outer}:{PFSC_SECTOR}")),
("config.ini", vec!["image_ro", "image_rw"], format!("image_ro={nested}")),
];
let mut res = HintResult { changed: vec![], backups: vec![] };
std::fs::create_dir_all(backup_dir).map_err(|e| e.to_string())?;
let stamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
for (file, keys, line) in plan {
let path = format!("{SMP_DIR}/{file}");
let current = ftp.retr(&path)?;
if current.is_none() && file == "config.ini" {
return Err("ShadowMountPlus isn't set up on this PS5 (no /data/shadowmount/config.ini)".into());
}
let text = String::from_utf8_lossy(current.as_deref().unwrap_or_default()).into_owned();
let Some(next) = upsert_hint(&text, &keys, &names, Some(&line)) else { continue };
if let Some(bytes) = &current {
let b = backup_dir.join(format!("{stamp}-{file}"));
std::fs::write(&b, bytes).map_err(|e| e.to_string())?;
res.backups.push(b.to_string_lossy().into_owned());
}
if ftp.put_atomic(&path, next.as_bytes()).is_err() {
// Some servers refuse to rename over an existing file.
ftp.delete(&path).ok();
let (dir, name) = path.rsplit_once('/').unwrap();
ftp.rename(&format!("{dir}/.{name}.packer-tmp"), &path)?;
}
res.changed.push(format!("{file}: {line}"));
}
Ok(res)
}
/// Polls `/user/app/<TITLE>/mount_img.lnk` until ShadowMountPlus records the image.
pub fn wait_for_mount(
s: &Settings,
title_id: &str,
image_remote: &str,
timeout: Duration,
cancel: &std::sync::atomic::AtomicBool,
tick: &mut dyn FnMut(u64),
) -> Result<bool, String> {
let start = Instant::now();
let link = format!("/user/app/{title_id}/mount_img.lnk");
while start.elapsed() < timeout {
if cancel.load(std::sync::atomic::Ordering::Relaxed) {
return Err("Cancelled".into());
}
tick(start.elapsed().as_secs());
if let Ok(mut ftp) = connect(s) {
if let Ok(Some(bytes)) = ftp.retr(&link) {
let text = String::from_utf8_lossy(&bytes);
if text.lines().any(|l| l.trim() == image_remote) {
ftp.quit();
return Ok(true);
}
}
ftp.quit();
}
std::thread::sleep(Duration::from_secs(3));
}
Ok(false)
}
#[cfg(test)]
mod tests {
use super::*;
fn names() -> Vec<String> {
vec!["PPSA01234.ffpfsc".into(), "PPSA01234.exfat".into(), "PPSA01234.pfs".into()]
}
#[test]
fn adds_hint_and_keeps_other_lines() {
let before = "# comment\nscan_interval=15\nimage_ro=OTHER.exfat\n";
let after = upsert_hint(before, &["image_ro", "image_rw"], &names(), Some("image_ro=PPSA01234.exfat")).unwrap();
assert_eq!(after, "# comment\nscan_interval=15\nimage_ro=OTHER.exfat\nimage_ro=PPSA01234.exfat\n");
}
#[test]
fn replaces_stale_hints_for_the_same_title() {
let before = "image_sector=ppsa01234.FFPFSC:4096\nimage_sector=PPSA01234.exfat:8192\nimage_sector=X.ffpfsc:65536\n";
let after =
upsert_hint(before, &["image_sector"], &names(), Some("image_sector=PPSA01234.ffpfsc:65536")).unwrap();
assert_eq!(after, "image_sector=X.ffpfsc:65536\nimage_sector=PPSA01234.ffpfsc:65536\n");
}
#[test]
fn unchanged_when_already_correct() {
let before = "a=1\nimage_ro=PPSA01234.exfat\n";
assert!(upsert_hint(before, &["image_ro", "image_rw"], &names(), Some("image_ro=PPSA01234.exfat")).is_none());
}
#[test]
fn drops_rw_rule_that_would_conflict() {
let before = "image_rw=PPSA01234.exfat\n";
let after = upsert_hint(before, &["image_ro", "image_rw"], &names(), Some("image_ro=PPSA01234.exfat")).unwrap();
assert_eq!(after, "image_ro=PPSA01234.exfat\n");
}
}
+84
View File
@@ -0,0 +1,84 @@
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, rename_all = "camelCase")]
pub struct Settings {
pub ps5_host: String,
pub ftp_port: u16,
pub web_port: u16,
/// Where finished images go on the PS5. ShadowMountPlus scans this by default.
pub remote_dir: String,
/// Where images are written on this PC.
pub output_dir: String,
pub compression_level: u8,
/// 0 = let MkPFS decide.
pub cpu_count: u32,
/// 0 = half the cores, capped at 8.
pub verify_workers: u32,
/// Continue to upload straight after a passed verify.
pub auto_upload: bool,
/// Write the image_sector / image_ro hints Game Compressor writes.
pub write_smp_hints: bool,
/// Delete the local image once it's confirmed on the PS5.
pub delete_local_after_upload: bool,
}
impl Default for Settings {
fn default() -> Self {
Self {
ps5_host: "192.168.0.50".into(),
ftp_port: 2121,
web_port: 8888,
remote_dir: "/data/homebrew".into(),
output_dir: default_output_dir().to_string_lossy().into_owned(),
compression_level: 7,
cpu_count: 0,
verify_workers: 0,
auto_upload: true,
write_smp_hints: true,
delete_local_after_upload: false,
}
}
}
fn home() -> PathBuf {
std::env::var_os("HOME").map(PathBuf::from).unwrap_or_else(|| PathBuf::from("/tmp"))
}
fn default_output_dir() -> PathBuf {
home().join("PS5 Images")
}
pub fn config_dir() -> PathBuf {
std::env::var_os("XDG_CONFIG_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home().join(".config"))
.join("ps5-packer")
}
pub fn data_dir() -> PathBuf {
std::env::var_os("XDG_DATA_HOME")
.map(PathBuf::from)
.unwrap_or_else(|| home().join(".local/share"))
.join("ps5-packer")
}
fn settings_path() -> PathBuf {
config_dir().join("settings.json")
}
pub fn load() -> Settings {
std::fs::read_to_string(settings_path())
.ok()
.and_then(|s| serde_json::from_str(&s).ok())
.unwrap_or_default()
}
pub fn save(s: &Settings) -> Result<(), String> {
let path = settings_path();
std::fs::create_dir_all(path.parent().unwrap()).map_err(|e| e.to_string())?;
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, serde_json::to_vec_pretty(s).unwrap()).map_err(|e| e.to_string())?;
std::fs::rename(&tmp, &path).map_err(|e| e.to_string())
}
+35
View File
@@ -0,0 +1,35 @@
{
"$schema": "https://schema.tauri.app/config/2",
"productName": "Packer",
"version": "0.1.0",
"identifier": "dev.unicxrn.ps5packer",
"build": {
"beforeDevCommand": "npm run dev",
"beforeBuildCommand": "npm run build",
"devUrl": "http://localhost:1430",
"frontendDist": "../dist"
},
"app": {
"windows": [
{
"label": "main",
"title": "Packer",
"width": 1320,
"height": 860,
"minWidth": 760,
"minHeight": 560,
"resizable": true,
"decorations": false,
"center": true,
"dragDropEnabled": true
}
],
"security": { "csp": null }
},
"bundle": {
"active": true,
"targets": ["deb", "appimage"],
"icon": ["icons/32x32.png", "icons/128x128.png", "icons/128x128@2x.png", "icons/icon.png"],
"resources": []
}
}