Prepare original HUD assets and port initial APT execution and display lists

This commit is contained in:
chazmm committed 2026-09-08 16:29:18 +01:00
1 parent 3c535a392b
commit 1c768a3e6c
21 files changed
+3438 -3

No files matched your search

+125
View File
@@ -0,0 +1,125 @@
//! Bytecode-only validation. No Bevy, game, input, rendering or simulation.
#[path = "../../skate-game/src/apt_display.rs"]
mod apt_display;
#[path = "../../skate-game/src/apt_vm.rs"]
mod apt_vm;
use apt_vm::*;
struct Audit;
// Build the authored frame-zero hierarchy, so constructor property writes
// have actual targets. This is not a timeline playback or renderer test.
fn initial_tree(
vm: &mut Vm,
json: &serde_json::Value,
character: i32,
nesting: usize,
) -> Result<usize, String> {
if nesting > 32 {
return Err("APT hierarchy nesting limit".into());
}
let record = json["characters"]
.as_array()
.ok_or("No characters")?
.iter()
.find(|c| c["id"].as_i64() == Some(character as i64))
.ok_or("Unknown APT character")?;
let object = vm.object(ObjectKind::Native(format!("character:{character}")));
vm.set(object, "_visible", Value::Bool(true))?;
vm.set(object, "_x", Value::Number(0.0))?;
vm.set(object, "_y", Value::Number(0.0))?;
let mut list = apt_display::DisplayList::default();
if let Some(controls) = record["frames"][0]["controls"].as_array() {
for control in controls {
list.apply(&serde_json::from_value(control.clone()).map_err(|e| e.to_string())?)?;
}
}
for placement in list.depths.values() {
let child = initial_tree(vm, json, placement.character, nesting + 1)?;
vm.set(child, "_parent", Value::Object(object))?;
vm.set(child, "_x", Value::Number(placement.matrix[4] as f64))?;
vm.set(child, "_y", Value::Number(placement.matrix[5] as f64))?;
if !placement.name.is_empty() {
vm.set(object, &placement.name, Value::Object(child))?;
}
}
Ok(object)
}
impl Host for Audit {
fn call(
&mut self,
_vm: &mut Vm,
object: usize,
method: &str,
args: Vec<Value>,
) -> Result<Value, String> {
println!("native {object}.{method}({args:?})");
Ok(Value::Undefined)
}
}
fn main() -> Result<(), String> {
let path = std::env::args_os()
.nth(1)
.ok_or("Expected private trickdisplay.json")?;
let json: serde_json::Value =
serde_json::from_slice(&std::fs::read(path).map_err(|e| e.to_string())?)
.map_err(|e| e.to_string())?;
let mut vm = Vm::new();
let mut timeline_count = 0;
for character in json["characters"].as_array().ok_or("No characters")? {
let Some(frames) = character["frames"].as_array() else {
continue;
};
let mut list = apt_display::DisplayList::default();
for frame in frames {
for control in frame["controls"].as_array().ok_or("No controls")? {
list.apply(&serde_json::from_value(control.clone()).map_err(|e| e.to_string())?)?;
}
}
timeline_count += 1;
}
println!("Validated placement controls in {timeline_count} original timelines");
let clip = vm.object(ObjectKind::Native("MovieClip".into()));
let proto = vm.object(ObjectKind::Plain);
vm.set(clip, "prototype", Value::Object(proto))?;
vm.set(vm.global, "MovieClip", Value::Object(clip))?;
for c in json["characters"].as_array().ok_or("No characters")? {
if let Some(frames) = c["frames"].as_array() {
for frame in frames {
for control in frame["controls"].as_array().ok_or("No controls")? {
if control["type_name"] == "do_init_action" {
let key = control["actions_offset"].as_u64().unwrap().to_string();
let code: Vec<Instruction> =
serde_json::from_value(json["actions"][&key].clone())
.map_err(|e| e.to_string())?;
vm.run(&code, &mut Audit)?;
}
}
}
}
}
let Value::Object(class) = vm.get(vm.global, "trickdisplay2") else {
return Err("APT class initialization failed".into());
};
println!("APT class initialized; {} objects", vm.objects.len());
let screen = initial_tree(&mut vm, &json, 0, 0)?;
vm.begin_update();
let instance = vm.construct(class, vec![Value::Object(screen)], &mut Audit)?;
for name in [
"mcTrickAnim0",
"mcTrickAnim1",
"mcTrickAnim2",
"mcTrickAnim3",
] {
let Value::Object(child) = vm.get(screen, name) else {
return Err(format!("Missing authored child {name}"));
};
if vm.get(child, "_visible") != Value::Bool(false) {
return Err(format!("Constructor failed to hide {name}"));
}
}
println!(
"APT constructor completed; instance {instance}; {} objects",
vm.objects.len()
);
Ok(())
}
+106
View File
@@ -0,0 +1,106 @@
//! Retained APT depth list. Placement flags update individual properties;
//! omitted properties on a move retain their previous values.
use serde::Deserialize;
use std::collections::BTreeMap;
#[derive(Clone, Debug, Deserialize)]
pub struct Control {
pub type_name: String,
#[serde(default)]
pub flags: u32,
#[serde(default)]
pub depth: i32,
pub character_id: Option<i32>,
pub matrix: Option<[f32; 6]>,
pub color_transform: Option<[u8; 8]>,
pub ratio: Option<f32>,
pub name: Option<String>,
pub clip_depth: Option<i32>,
pub blend_mode: Option<i32>,
#[serde(default)]
pub filter_pointer: u32,
#[serde(default)]
pub actions_offset: u32,
}
#[derive(Clone, Debug, PartialEq)]
pub struct Placement {
pub character: i32,
pub matrix: [f32; 6],
pub color: [u8; 8],
pub ratio: f32,
pub name: String,
pub clip_depth: i32,
pub blend_mode: i32,
}
#[derive(Clone, Debug, Default)]
pub struct DisplayList {
pub depths: BTreeMap<i32, Placement>,
}
impl DisplayList {
pub fn apply(&mut self, control: &Control) -> Result<(), String> {
match control.type_name.as_str() {
"remove_object2" | "remove_object3" => {
self.depths.remove(&control.depth);
}
"place_object2" | "place_object3" => {
if control.filter_pointer != 0 {
return Err("APT placement filter is not implemented".into());
}
if control.flags & 0x80 != 0 && control.actions_offset != 0 {
return Err("APT placement clip actions are not implemented".into());
}
let mut placement = if control.flags & 1 != 0 {
self.depths.get(&control.depth).cloned().ok_or_else(|| {
format!("APT move references empty depth {}", control.depth)
})?
} else {
Placement {
character: control
.character_id
.filter(|_| control.flags & 2 != 0)
.ok_or("APT new placement lacks character")?,
matrix: [1.0, 0.0, 0.0, 1.0, 0.0, 0.0],
color: [255, 255, 255, 255, 0, 0, 0, 0],
ratio: 0.0,
name: String::new(),
clip_depth: -1,
blend_mode: -1,
}
};
if control.flags & 2 != 0 {
placement.character = control
.character_id
.ok_or("APT character flag lacks value")?;
}
if control.flags & 4 != 0 {
placement.matrix = control.matrix.ok_or("APT matrix flag lacks value")?;
}
if control.flags & 8 != 0 {
placement.color = control
.color_transform
.ok_or("APT color flag lacks value")?;
}
if control.flags & 0x10 != 0 {
placement.ratio = control.ratio.ok_or("APT ratio flag lacks value")?;
}
if control.flags & 0x20 != 0 {
placement.name = control.name.clone().ok_or("APT name flag lacks value")?;
}
if control.flags & 0x40 != 0 {
placement.clip_depth = control
.clip_depth
.ok_or("APT clip-depth flag lacks value")?;
}
if control.type_name == "place_object3" {
placement.blend_mode = control.blend_mode.unwrap_or(-1);
}
self.depths.insert(control.depth, placement);
}
"do_action" | "do_init_action" | "frame_label" | "background_color" => {}
kind => return Err(format!("Unsupported APT control {kind}")),
}
Ok(())
}
}
+584
View File
@@ -0,0 +1,584 @@
//! APT's compact ActionScript instruction stream. All execution is bounded;
//! native callbacks are supplied by the HUD owner, never by extracted scripts.
use serde::Deserialize;
use std::collections::BTreeMap;
#[derive(Clone, Debug, Default, PartialEq)]
pub enum Value {
#[default]
Undefined,
Number(f64),
Bool(bool),
Text(String),
Object(usize),
}
impl Value {
pub fn number(&self) -> f64 {
match self {
Self::Number(n) => *n,
Self::Bool(v) => {
if *v {
1.0
} else {
0.0
}
}
Self::Text(s) => s.parse().unwrap_or(f64::NAN),
_ => f64::NAN,
}
}
pub fn truth(&self) -> bool {
match self {
Self::Undefined => false,
Self::Number(n) => *n != 0.0 && !n.is_nan(),
Self::Bool(v) => *v,
Self::Text(s) => !s.is_empty(),
Self::Object(_) => true,
}
}
pub fn text(&self) -> String {
match self {
Self::Undefined => "undefined".into(),
Self::Number(n) => n.to_string(),
Self::Bool(v) => v.to_string(),
Self::Text(s) => s.clone(),
Self::Object(_) => "[object Object]".into(),
}
}
}
#[derive(Clone, Debug, Deserialize)]
pub struct Constant {
pub kind: u32,
pub value: serde_json::Value,
}
impl Constant {
fn value(&self) -> Result<Value, String> {
Ok(match self.kind {
1 => Value::Text(
self.value
.as_str()
.ok_or("Invalid APT string constant")?
.into(),
),
5 => Value::Bool(self.value.as_u64().ok_or("Invalid APT bool constant")? != 0),
6 | 7 => Value::Number(self.value.as_f64().ok_or("Invalid APT number constant")?),
3 => Value::Undefined,
_ => return Err(format!("Unresolved APT constant type {}", self.kind)),
})
}
}
#[derive(Clone, Debug, Deserialize)]
pub struct Parameter {
pub register: usize,
pub name: String,
}
#[derive(Clone, Debug, Deserialize)]
pub struct Instruction {
pub offset: u32,
pub opcode: u8,
#[serde(default)]
pub next: u32,
#[serde(default)]
pub operand: serde_json::Value,
pub target: Option<u32>,
#[serde(default)]
pub values: Vec<Constant>,
#[serde(default)]
pub body: Vec<Instruction>,
#[serde(default)]
pub flags: u32,
#[serde(default)]
pub parameters: Vec<Parameter>,
#[serde(default)]
pub name: String,
}
#[derive(Clone, Debug)]
pub enum ObjectKind {
Plain,
Function(usize),
Native(String),
}
#[derive(Clone, Debug)]
pub struct Object {
pub kind: ObjectKind,
pub fields: BTreeMap<String, Value>,
pub prototype: Option<usize>,
}
#[derive(Clone, Debug)]
struct Function {
code: Instruction,
constants: Vec<Value>,
}
struct Scope {
this: usize,
locals: BTreeMap<String, Value>,
}
pub trait Host {
fn call(
&mut self,
vm: &mut Vm,
object: usize,
method: &str,
arguments: Vec<Value>,
) -> Result<Value, String>;
}
#[derive(Default)]
pub struct Vm {
pub objects: Vec<Object>,
functions: Vec<Function>,
pub global: usize,
remaining: usize,
depth: usize,
}
impl Vm {
fn variable(&self, scope: &Scope, key: &str) -> Value {
if let Some(value) = scope.locals.get(key) {
return value.clone();
}
let value = self.get(scope.this, key);
if value != Value::Undefined {
return value;
}
self.get(self.global, key)
}
pub fn new() -> Self {
let mut vm = Self::default();
vm.global = vm.object(ObjectKind::Plain);
vm
}
pub fn object(&mut self, kind: ObjectKind) -> usize {
let id = self.objects.len();
self.objects.push(Object {
kind,
fields: BTreeMap::new(),
prototype: None,
});
id
}
pub fn set(
&mut self,
object: usize,
key: impl Into<String>,
value: Value,
) -> Result<(), String> {
self.objects
.get_mut(object)
.ok_or("Invalid APT object handle")?
.fields
.insert(key.into(), value);
Ok(())
}
pub fn get(&self, object: usize, key: &str) -> Value {
let mut current = Some(object);
for _ in 0..64 {
let Some(o) = current.and_then(|i| self.objects.get(i)) else {
break;
};
if let Some(v) = o.fields.get(key) {
return v.clone();
}
current = o.prototype;
}
Value::Undefined
}
pub fn begin_update(&mut self) {
self.remaining = 100_000;
}
pub fn construct(
&mut self,
class: usize,
args: Vec<Value>,
host: &mut impl Host,
) -> Result<usize, String> {
let id = self.object(ObjectKind::Plain);
if let Value::Object(proto) = self.get(class, "prototype") {
self.objects[id].prototype = Some(proto);
}
if let ObjectKind::Function(f) = self.objects[class].kind.clone() {
self.invoke(f, id, args, host)?;
}
Ok(id)
}
pub fn run(&mut self, code: &[Instruction], host: &mut impl Host) -> Result<Value, String> {
self.begin_update();
self.run_on(self.global, code, host)
}
pub fn run_on(
&mut self,
object: usize,
code: &[Instruction],
host: &mut impl Host,
) -> Result<Value, String> {
self.execute(
code,
&mut vec![Value::Undefined; 256],
&mut Vec::new(),
&mut Scope {
this: object,
locals: BTreeMap::new(),
},
host,
)
}
pub fn call_method(
&mut self,
object: usize,
method: &str,
args: Vec<Value>,
host: &mut impl Host,
) -> Result<Value, String> {
if let Value::Object(f) = self.get(object, method) {
if let Some(Object {
kind: ObjectKind::Function(index),
..
}) = self.objects.get(f)
{
return self.invoke(*index, object, args, host);
}
}
host.call(self, object, method, args)
}
fn invoke(
&mut self,
index: usize,
this: usize,
args: Vec<Value>,
host: &mut impl Host,
) -> Result<Value, String> {
if self.depth >= 32 {
return Err("APT call-depth limit".into());
}
let f = self
.functions
.get(index)
.ok_or("Invalid APT function")?
.clone();
let mut regs = vec![Value::Undefined; 256];
let mut reg = 1;
let mut scope = Scope {
this,
locals: BTreeMap::new(),
};
if f.code.flags & 2 == 0 {
scope.locals.insert("this".into(), Value::Object(this));
}
let root = match self.get(this, "_root") {
Value::Undefined => Value::Object(self.global),
value => value,
};
let parent = self.get(this, "_parent");
let arguments = if f.code.flags & 4 != 0 || f.code.flags & 8 == 0 {
let object = self.object(ObjectKind::Plain);
for (index, value) in args.iter().enumerate() {
self.set(object, index.to_string(), value.clone())?;
}
self.set(object, "length", Value::Number(args.len() as f64))?;
Value::Object(object)
} else {
Value::Undefined
};
if f.code.flags & 8 == 0 {
scope.locals.insert("arguments".into(), arguments.clone());
}
// DefineFunction2 preload flags (82E47688). Native integer registers
// are scoped to a call, unlike persistent display-object properties.
for (flag, value) in [
(1, Value::Object(this)),
(4, arguments),
(16, Value::Undefined),
(64, root),
(128, parent),
(256, Value::Object(self.global)),
] {
if f.code.flags & flag != 0 {
regs[reg] = value;
reg += 1;
}
}
for (i, p) in f.code.parameters.iter().enumerate() {
if p.register >= regs.len() {
return Err("APT parameter register outside bank".into());
}
if p.register != 0 {
regs[p.register] = args.get(i).cloned().unwrap_or_default();
} else {
scope
.locals
.insert(p.name.clone(), args.get(i).cloned().unwrap_or_default());
}
}
self.depth += 1;
let result = self.execute(
&f.code.body,
&mut regs,
&mut f.constants.clone(),
&mut scope,
host,
);
self.depth -= 1;
result
}
fn execute(
&mut self,
code: &[Instruction],
regs: &mut Vec<Value>,
constants: &mut Vec<Value>,
scope: &mut Scope,
host: &mut impl Host,
) -> Result<Value, String> {
let mut stack = Vec::<Value>::new();
let mut pc = 0;
fn pop(s: &mut Vec<Value>) -> Result<Value, String> {
s.pop().ok_or("APT stack underflow".into())
}
while let Some(i) = code.get(pc) {
if self.remaining == 0 || self.objects.len() > 4096 {
return Err("APT execution/object budget exceeded".into());
}
self.remaining -= 1;
pc += 1;
let op = i.opcode;
let operand = i.operand.as_u64().unwrap_or(0) as usize;
let constant = |k: usize| {
constants
.get(k)
.cloned()
.ok_or_else(|| format!("APT constant index {k} at {:x}", i.offset))
};
match op {
0 => break,
0x06 | 0x07 => {
host.call(
self,
scope.this,
if op == 6 { "play" } else { "stop" },
vec![],
)?;
}
0x17 => {
pop(&mut stack)?;
}
0x12 => {
let a = pop(&mut stack)?;
stack.push(Value::Bool(!a.truth()));
}
0x4c => stack.push(
stack
.last()
.cloned()
.ok_or("APT duplicate on empty stack")?,
),
0x59 => stack.push(Value::Number(0.0)),
0x5a => stack.push(Value::Number(1.0)),
0x71 => stack.push(Value::Object(self.global)),
0x73 => stack.push(Value::Bool(true)),
0x74 => stack.push(Value::Bool(false)),
0x75 | 0x76 => stack.push(Value::Undefined),
0xb5 => stack.push(Value::Number((operand as u8 as i8) as f64)),
0xb6 => stack.push(Value::Number((operand as u16 as i16) as f64)),
0xb7 => stack.push(Value::Number((operand as u32 as i32) as f64)),
0xb4 => stack.push(Value::Number(f32::from_bits(operand as u32) as f64)),
0xb9 => stack.push(
regs.get(operand)
.cloned()
.ok_or("APT register outside bank")?,
),
0x87 => {
*regs.get_mut(operand).ok_or("APT register outside bank")? = stack
.last()
.cloned()
.ok_or("APT register store on empty stack")?;
}
0x88 => {
*constants = i
.values
.iter()
.map(Constant::value)
.collect::<Result<_, _>>()?;
}
0x96 => {
for c in &i.values {
stack.push(c.value()?);
}
}
0xa2 | 0xa3 => stack.push(constant(operand)?),
0xae => stack.push(self.variable(scope, &constant(operand)?.text())),
0x4e | 0xaf => {
let name = if op == 0xaf {
constant(operand)?
} else {
pop(&mut stack)?
}
.text();
let obj = pop(&mut stack)?;
stack.push(if let Value::Object(id) = obj {
self.get(id, &name)
} else {
Value::Undefined
});
}
0x4f => {
let v = pop(&mut stack)?;
let k = pop(&mut stack)?.text();
let o = pop(&mut stack)?;
if let Value::Object(id) = o {
self.set(id, k, v)?;
}
}
0x1c => {
let k = pop(&mut stack)?.text();
stack.push(self.variable(scope, &k));
}
0x1d | 0x3c => {
let v = pop(&mut stack)?;
let k = pop(&mut stack)?.text();
if op == 0x3c || scope.locals.contains_key(&k) {
scope.locals.insert(k, v);
} else {
self.set(scope.this, k, v)?;
}
}
0x3a => {
let k = pop(&mut stack)?.text();
let o = pop(&mut stack)?;
let existed = if let Value::Object(id) = o {
self.objects[id].fields.remove(&k).is_some()
} else {
false
};
stack.push(Value::Bool(existed));
}
0x47 | 0x0b | 0x0c | 0x0d | 0x48 | 0x49 | 0x66 | 0x67 => {
let b = pop(&mut stack)?;
let a = pop(&mut stack)?;
stack.push(match op {
0x47 if matches!(a, Value::Text(_)) || matches!(b, Value::Text(_)) => {
Value::Text(a.text() + &b.text())
}
0x47 => Value::Number(a.number() + b.number()),
0x0b => Value::Number(a.number() - b.number()),
0x0c => Value::Number(a.number() * b.number()),
0x0d => Value::Number(a.number() / b.number()),
0x48 => Value::Bool(a.number() < b.number()),
0x67 => Value::Bool(a.number() > b.number()),
0x66 => Value::Bool(a == b),
_ => Value::Bool(a == b || a.number() == b.number()),
});
}
0x50 | 0x51 => {
let a = pop(&mut stack)?.number();
stack.push(Value::Number(a + if op == 0x50 { 1.0 } else { -1.0 }));
}
0x99 | 0x9d | 0xb8 => {
let jump = op == 0x99 || {
let v = pop(&mut stack)?.truth();
if op == 0xb8 { !v } else { v }
};
if jump {
let target = i.target.ok_or("APT branch lacks target")?;
if target == code.last().map_or(0, |x| x.next) {
break;
}
pc = code
.iter()
.position(|x| x.offset == target)
.ok_or_else(|| format!("APT invalid jump target {target:x}"))?;
}
}
0x8e | 0x9b => {
let index = self.functions.len();
self.functions.push(Function {
code: i.clone(),
constants: constants.clone(),
});
let object = self.object(ObjectKind::Function(index));
let proto = self.object(ObjectKind::Plain);
self.set(object, "prototype", Value::Object(proto))?;
if i.name.is_empty() {
stack.push(Value::Object(object));
} else {
self.set(self.global, &i.name, Value::Object(object))?;
}
}
0x69 => {
let super_class = pop(&mut stack)?;
let sub_class = pop(&mut stack)?;
if let (Value::Object(a), Value::Object(b)) = (sub_class, super_class) {
if let (Value::Object(ap), Value::Object(bp)) =
(self.get(a, "prototype"), self.get(b, "prototype"))
{
self.objects[ap].prototype = Some(bp);
}
}
}
0x40 => {
let name = pop(&mut stack)?.text();
let n = pop(&mut stack)?.number() as usize;
if n > 256 {
return Err("APT argument limit".into());
}
let args = (0..n)
.map(|_| pop(&mut stack))
.collect::<Result<Vec<_>, _>>()?;
let id = self.object(ObjectKind::Plain);
if name == "Array" {
for (j, v) in args.into_iter().enumerate() {
self.set(id, j.to_string(), v)?;
}
self.set(id, "length", Value::Number(n as f64))?;
} else if let Value::Object(class) = self.get(self.global, &name) {
if let Value::Object(proto) = self.get(class, "prototype") {
self.objects[id].prototype = Some(proto);
}
if let ObjectKind::Function(f) = self.objects[class].kind.clone() {
self.invoke(f, id, args, host)?;
}
} else {
return Err(format!("APT constructor absent: {name}"));
}
stack.push(Value::Object(id));
}
0x52 | 0xb0 | 0xb1 | 0xb2 | 0xb3 | 0x3d | 0x5d => {
let method = if matches!(op, 0xb0 | 0xb1 | 0xb2 | 0xb3) {
constant(operand)?
} else {
pop(&mut stack)?
};
let object = if matches!(op, 0x3d | 0xb0 | 0xb1) {
Value::Object(self.global)
} else {
pop(&mut stack)?
};
let count = pop(&mut stack)?.number() as usize;
if count > 256 {
return Err("APT argument limit".into());
}
let args = (0..count)
.map(|_| pop(&mut stack))
.collect::<Result<Vec<_>, _>>()?;
let value = if let Value::Object(id) = object {
self.call_method(id, &method.text(), args, host)?
} else {
Value::Undefined
};
if matches!(op, 0xb1 | 0xb3) {
return Ok(value);
}
if !matches!(op, 0xb0 | 0xb2 | 0x5d) {
stack.push(value);
}
}
0x3e => return Ok(stack.pop().unwrap_or_default()),
_ => return Err(format!("Unsupported APT opcode {op:02x} at {:x}", i.offset)),
}
if stack.len() > 1024 {
return Err("APT stack limit".into());
}
}
Ok(Value::Undefined)
}
}
+2
View File
@@ -1,4 +1,6 @@
mod animation;
mod apt_vm;
mod apt_display;
mod animation_pose;
mod app;
mod assets;
+13 -3
View File
@@ -29,8 +29,13 @@ Implemented source components:
geometry, font metadata, textures and compact action stream into a private
cache. Source asset hashes are retained in the compiled manifest.
* `apt_vm`: partial bounded interpreter for the original compact instructions.
The data-only audit initializes the original class and runs its constructor
against a dummy host. This does not validate a functioning display.
The data-only audit initializes the original class, builds the authored
initial child hierarchy and checks the constructor's four trick-name
visibility writes. Native method calls still use a dummy host, so this does
not validate a functioning display.
* `apt_display`: retained depth-list placement flags and transforms. The
data-only audit validates all controls across 38 original movie timelines.
Unsupported clip actions and filters fail explicitly.
## Extractor provenance
@@ -61,7 +66,7 @@ announcement delays to ticks. A physics-frame count is not a safe substitute.
The HUD still needs a complete movie object hierarchy, native bindings,
timeline execution and rendering with multiplicative and additive colors.
Function scope/preloads and supported action coverage also remain incomplete.
Superclass preloads and supported action coverage also remain incomplete.
The audit's dummy host intentionally does not establish native binding parity.
The movie references `FuturaOuterGlow`, absent under that exact name from the
@@ -74,3 +79,8 @@ Pause, map-change generation reset, resource reuse and the final task-specific
release build/launcher remain to be integrated and checked. No game/recomp,
controller harness, gameplay automation, `--check-assets`, or screenshot
capture has been run for this work.
Validation so far: eight scoring unit tests pass (including unsigned clock
rollover, early completion, conversion, timer thresholds, one-time banking and
cancelled rewards); the owned-data loader audit resolves 300 definitions; the
APT data-only audit passes placement validation and constructor checks.
+72
View File
@@ -0,0 +1,72 @@
"""Incrementally prepare original HUD assets from the user's extracted game.
No executable/game launch. Output contains copyrighted assets and stays private.
The vendored preview.18 UI extractor supplies the original decoding pipeline.
"""
from pathlib import Path
import argparse
import hashlib
import json
import sys
from vendor.skate3_ui.project import extract_project
from vendor.skate3_ui.scene_graph import AssetCache, SceneFlattener
from vendor.skate3_ui.actions import Actions
def prepare(game: Path, output: Path):
extract_project(game, output, prefixes=("data/fe/source/screens/hud2/",
"data/fe/source/controls/"), update=True)
cache = AssetCache(output)
name = "data/fe/source/screens/hud2/trickdisplay2"
bundle = cache.load_bundle(name)
apt_path = output / "raw" / (name + ".apt")
const_path = apt_path.with_suffix('.const')
actions = Actions(apt_path.read_bytes(), const_path.read_bytes())
blocks = {}
for c in bundle['characters'].values():
for f in c.get('frames', []):
for control in f['controls']:
if control['type_name'] in ('do_action', 'do_init_action'):
offset = control.get('actions_offset', 0)
if offset:
blocks[str(offset)] = actions.stream(offset)
shapes = {}
fonts = {}
for c in bundle['characters'].values():
if c['type_name'] == 'shape':
scene = SceneFlattener(cache, lambda *_: {}).flatten(name, c['id'])
if scene['unresolved']:
raise ValueError(f"Unresolved original HUD shape: {scene['unresolved']}")
shapes[str(c['id'])] = scene['primitives']
elif c['type_name'] == 'font':
family = c['font']['name']
asset = cache.font_asset(family)
# Preserve this explicit source limitation. Preview.18 inferred a
# metric alias in HighlightText; that is not proof of retail HUD
# glyph/raster substitution. Never silently substitute a font.
fonts[family] = asset
result = {
'format': 'skate3-scoring-hud', 'version': 1,
'source': {'bundle': name,
'apt_sha256': hashlib.sha256(apt_path.read_bytes()).hexdigest(),
'const_sha256': hashlib.sha256(const_path.read_bytes()).hexdigest()},
'characters': list(bundle['characters'].values()),
'shapes': shapes, 'fonts': fonts, 'actions': blocks,
'unresolved_fonts': [family for family, asset in fonts.items() if asset is None],
}
target = output / 'runtime' / 'trickdisplay.json'
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(json.dumps(result, separators=(',', ':')) + '\n', encoding='utf-8')
print(f"Prepared {len(shapes)} original HUD shapes and {len(blocks)} action blocks: {target}")
if result['unresolved_fonts']:
print('Unresolved authored font families: ' + ', '.join(result['unresolved_fonts']))
return result
if __name__ == '__main__':
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('--game', type=Path, required=True)
p.add_argument('--output', type=Path, required=True)
args = p.parse_args()
prepare(args.game.resolve(), args.output.resolve())
+26
View File
@@ -0,0 +1,26 @@
MIT License
Copyright (c) 2026 Skate 3 Custom Engine Layer contributors
Permission is hereby granted, free of charge, to any person obtaining a copy
of the original Custom Engine Layer software and associated documentation
files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to permit
persons to whom the Software is furnished to do so, subject to the following
conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
This license applies only to original project-owned additions. It does not
relicense Skate3Recomp, rexglue, third-party dependencies, Electronic Arts
game data, or assets embedded by map authors.
+3
View File
@@ -0,0 +1,3 @@
"""Read-only Skate 3 front-end asset inspection helpers."""
__version__ = "0.1.0"
+96
View File
@@ -0,0 +1,96 @@
"""Read TU3 APT bytecode without running it.
Operand alignment/relocation: native 82E67868, switch table 82E67914.
Function body boundaries: 82E73010 (DefineFunction2), 82E72EC0.
CONSTANTPOOL/PUSHDATA use CONST indices, not SWF's inline value encoding.
"""
from .binary import Reader, FormatError, align
BYTE = {0xA2, 0xAE, 0xAF, 0xB0, 0xB1, 0xB2, 0xB3, 0xB5, 0xB9}
SHORT = {0xA3, 0xB6}
WORD = {0x81, 0x87, 0x99, 0x9D, 0x9F, 0xB8}
STRING = {0x8B, 0x8C, 0xA1, 0xA4, 0xA5, 0xA6, 0xA7}
class Actions:
def __init__(self, apt: bytes, const: bytes):
self.apt = Reader(apt, 'APT actions')
c = Reader(const, 'APT constants')
count, start = c.u32be(24), c.u32be(28)
c.require(start, count * 8)
self.constants = []
for i in range(count):
at = start + i * 8
kind, value = c.u32be(at), c.u32be(at + 4)
if kind == 1:
c.require(value, 1)
value = c.cstring(value)
elif kind == 6:
value = c.f32be(at + 4)
elif kind == 7:
value = c.i32be(at + 4)
self.constants.append({'kind': kind, 'value': value})
def stream(self, offset: int, end: int | None = None, nesting: int = 0):
if nesting > 64:
raise FormatError('APT action nesting exceeds 64')
r = self.apt
end = len(r.data) if end is None else end
r.require(offset, end - offset)
result = []
while offset < end:
at, op = offset, r.u8(offset)
offset += 1
row = {'offset': at, 'opcode': op}
if op == 0:
result.append(row)
break
if op in BYTE:
row['operand'] = r.u8(offset); offset += 1
elif op in SHORT:
row['operand'] = r.u16be(offset); offset += 2
elif op in {0x77, 0xB4, 0xB7}:
row['operand'] = r.i32be(offset); offset += 4
elif op in WORD | STRING | {0x94}:
offset = align(offset, 4)
value = r.u32be(offset)
row['operand'] = r.cstring(value) if op in STRING else value
if op in {0x99, 0x9D, 0x94, 0xB8}:
row['target'] = offset + 4 + r.i32be(offset)
offset += 4
elif op in {0x88, 0x96}:
offset = align(offset, 4)
count, pointer = r.u32be(offset), r.u32be(offset + 4)
r.require(pointer, count * 4)
indices = [r.u32be(pointer + i * 4) for i in range(count)]
if any(i >= len(self.constants) for i in indices):
raise FormatError(f'APT constant index outside CONST at {at:#x}')
row['values'] = [self.constants[i] for i in indices]
offset += 8
elif op in {0x8E, 0x9B}:
offset = align(offset, 4)
header_size = 28 if op == 0x8E else 24
r.require(offset, header_size)
row['name'] = r.cstring(r.u32be(offset))
count = r.u32be(offset + 4)
pointer = r.u32be(offset + (12 if op == 0x8E else 8))
length = r.u32be(offset + (16 if op == 0x8E else 12))
stride = 8 if op == 0x8E else 4
r.require(pointer, count * stride)
row['parameters'] = [
{'register': r.u32be(pointer + i * 8) if op == 0x8E else 0,
'name': r.cstring(r.u32be(pointer + i * stride + (4 if op == 0x8E else 0)))}
for i in range(count)]
row['flags'] = r.u32be(offset + 8) if op == 0x8E else 0
offset += header_size
row['body'] = self.stream(offset, offset + length, nesting + 1)
offset += length
elif op in {0x83, 0x8F}:
# Keep unsupported compound instructions explicit, never scan
# their payload as though it were a stream of no-operand ops.
raise FormatError(f'APT compound opcode {op:#x} at {at:#x} needs decoding')
row['next'] = offset
if offset > end:
raise FormatError(f'APT instruction at {at:#x} crosses its function body')
result.append(row)
return result
+330
View File
@@ -0,0 +1,330 @@
from __future__ import annotations
from pathlib import Path
from .binary import FormatError, Reader
CHARACTER_TYPES = {
1: "shape",
2: "text",
3: "font",
4: "button",
5: "sprite",
6: "sound",
7: "bitmap",
8: "morph",
9: "animation",
10: "static_text",
11: "none",
12: "video",
13: "command",
14: "command_list",
15: "level",
16: "custom_control",
17: "image",
18: "plugin",
19: "bitmap_slice",
20: "avatar",
}
CONTROL_TYPES = {
1: "do_action",
2: "frame_label",
3: "place_object2",
4: "remove_object2",
5: "background_color",
6: "start_sound",
7: "start_sound_stream",
8: "do_init_action",
9: "place_object3",
10: "place_object4",
11: "remove_object3",
}
class AptInspector:
def __init__(self, apt_path: Path, const_path: Path):
self.apt_path = Path(apt_path)
self.const_path = Path(const_path)
self.apt = Reader(self.apt_path.read_bytes(), str(self.apt_path))
self.const = Reader(self.const_path.read_bytes(), str(self.const_path))
self.const.require(0, 32)
self.apt.require(0, 16)
def character(self, offset: int, character_id: int) -> dict | None:
if not offset:
return None
self.apt.require(offset, 16)
type_id = self.apt.u32be(offset)
body = offset + 16
result = {
"id": character_id,
"offset": offset,
"type": type_id,
"type_name": CHARACTER_TYPES.get(type_id, f"unknown_{type_id}"),
}
if type_id in (1, 2, 10):
self.apt.require(body, 16)
result["bounds"] = [self.apt.f32be(body + i * 4) for i in range(4)]
if type_id == 2:
self.apt.require(body, 52)
color = self.apt.u32be(body + 24)
result["text"] = {
"font_id": self.apt.u32be(body + 16),
"alignment": self.apt.u32be(body + 20),
"color_argb": f"#{color:08x}",
"font_height": self.apt.f32be(body + 28),
"read_only": bool(self.apt.u32be(body + 32)),
"multiline": bool(self.apt.u32be(body + 36)),
"word_wrap": bool(self.apt.u32be(body + 40)),
"initial_text": self.apt.cstring(self.apt.u32be(body + 44)),
"variable": self.apt.cstring(self.apt.u32be(body + 48)),
}
elif type_id == 3:
self.apt.require(body, 12)
glyph_count = self.apt.u32be(body + 4)
glyphs_offset = self.apt.u32be(body + 8)
if glyph_count > 100_000:
raise FormatError("unreasonable APT font glyph count")
glyphs = []
if glyph_count:
self.apt.require(glyphs_offset, glyph_count * 4)
glyphs = [
self.apt.u32be(glyphs_offset + index * 4)
for index in range(glyph_count)
]
result["font"] = {
"name": self.apt.cstring(self.apt.u32be(body)),
"glyph_character_ids": glyphs,
}
elif type_id == 10:
self.apt.require(body, 48)
result["matrix"] = [
self.apt.f32be(body + 16 + i * 4) for i in range(6)
]
record_count = self.apt.u32be(body + 40)
records_offset = self.apt.u32be(body + 44)
if record_count > 100_000:
raise FormatError("unreasonable APT static-text record count")
self.apt.require(records_offset, record_count * 56)
records = []
for record_index in range(record_count):
record_offset = records_offset + record_index * 56
glyph_count = self.apt.u32be(record_offset + 48)
glyphs_offset = self.apt.u32be(record_offset + 52)
if glyph_count > 1_000_000:
raise FormatError("unreasonable APT static-text glyph count")
self.apt.require(glyphs_offset, glyph_count * 4)
glyphs = []
for glyph_index in range(glyph_count):
glyph_offset = glyphs_offset + glyph_index * 4
glyphs.append(
{
"index": self.apt.u16be(glyph_offset),
"advance": int.from_bytes(
self.apt.bytes(glyph_offset + 2, 2),
"big",
signed=True,
),
}
)
records.append(
{
"offset": record_offset,
"font_id": self.apt.u32be(record_offset),
"color_transform": [
self.apt.f32be(record_offset + 4 + i * 4)
for i in range(8)
],
"x_offset": self.apt.f32be(record_offset + 36),
"y_offset": self.apt.f32be(record_offset + 40),
"scale": self.apt.f32be(record_offset + 44),
"glyphs": glyphs,
}
)
result["static_text"] = {"records": records}
elif type_id in (5, 9):
self.apt.require(body, 12)
movie = {
"frame_count": self.apt.u32be(body),
"frames_offset": self.apt.u32be(body + 4),
"labels_offset": self.apt.u32be(body + 8),
}
if movie["frame_count"] > 100_000:
raise FormatError("unreasonable APT frame count")
if type_id == 9:
self.apt.require(body, 52)
movie.update(
{
"character_count": self.apt.u32be(body + 12),
"characters_offset": self.apt.u32be(body + 16),
"width": self.apt.u32be(body + 20),
"height": self.apt.u32be(body + 24),
"milliseconds_per_frame": self.apt.u32be(body + 28),
"import_count": self.apt.u32be(body + 32),
"imports_offset": self.apt.u32be(body + 36),
"export_count": self.apt.u32be(body + 40),
"exports_offset": self.apt.u32be(body + 44),
"current_constant_index": self.apt.u32be(body + 48),
}
)
result["movie"] = movie
elif type_id == 7:
self.apt.require(body, 4)
result["bitmap"] = {"texture_id": self.apt.u32be(body)}
return result
def control(self, offset: int) -> dict:
self.apt.require(offset, 4)
type_id = self.apt.u32be(offset)
result = {
"offset": offset,
"type": type_id,
"type_name": CONTROL_TYPES.get(type_id, f"unknown_{type_id}"),
}
if type_id in (3, 9):
size = 72 if type_id == 9 else 64
self.apt.require(offset, size)
name_pointer = self.apt.u32be(offset + 52)
result.update(
{
"flags": self.apt.u32be(offset + 4),
"depth": self.apt.i32be(offset + 8),
"character_id": self.apt.i32be(offset + 12),
"matrix": [
self.apt.f32be(offset + 16 + i * 4) for i in range(6)
],
"color_transform": list(self.apt.bytes(offset + 40, 8)),
"ratio": self.apt.f32be(offset + 48),
"name": self.apt.cstring(name_pointer),
"clip_depth": self.apt.i32be(offset + 56),
"actions_offset": self.apt.u32be(offset + 60),
}
)
if type_id == 9:
result["blend_mode"] = self.apt.i32be(offset + 64)
result["filter_pointer"] = self.apt.u32be(offset + 68)
elif type_id == 2:
self.apt.require(offset, 8)
result["label"] = self.apt.cstring(self.apt.u32be(offset + 4))
elif type_id == 5:
self.apt.require(offset, 8)
result["color"] = f"#{self.apt.u32be(offset + 4):08x}"
elif type_id == 1:
self.apt.require(offset, 8)
result["actions_offset"] = self.apt.u32be(offset + 4)
action_offset = result["actions_offset"]
if action_offset:
self.apt.require(action_offset, 1)
result["first_action_opcode"] = self.apt.u8(action_offset)
elif type_id == 8:
self.apt.require(offset, 12)
result["sprite_id"] = self.apt.u32be(offset + 4)
result["actions_offset"] = self.apt.u32be(offset + 8)
action_offset = result["actions_offset"]
if action_offset:
self.apt.require(action_offset, 1)
result["first_action_opcode"] = self.apt.u8(action_offset)
elif type_id in (4, 11):
self.apt.require(offset, 8)
result["depth"] = self.apt.i32be(offset + 4)
elif type_id == 6:
self.apt.require(offset, 8)
result["sound_id"] = self.apt.u32be(offset + 4)
return result
def frames(self, movie: dict) -> list[dict]:
count = movie["frame_count"]
frames_offset = movie["frames_offset"]
self.apt.require(frames_offset, count * 8)
frames = []
for frame_index in range(count):
frame = frames_offset + frame_index * 8
control_count = self.apt.u32be(frame)
controls_offset = self.apt.u32be(frame + 4)
if control_count > 100_000:
raise FormatError(f"frame {frame_index}: unreasonable control count")
self.apt.require(controls_offset, control_count * 4)
controls = []
for control_index in range(control_count):
pointer = self.apt.u32be(controls_offset + control_index * 4)
if pointer:
controls.append(self.control(pointer))
frames.append({"index": frame_index, "controls": controls})
return frames
def inspect(self) -> dict:
header = {
"magic_hex": self.const.bytes(0, 20).hex(),
"main_character_offset": self.const.u32be(20),
"constant_count": self.const.u32be(24),
"constants_offset": self.const.u32be(28),
}
root = self.character(header["main_character_offset"], 0)
imports = []
exports = []
characters = []
if root and root["type"] in (5, 9):
movie = root["movie"]
import_count = movie.get("import_count", 0)
imports_offset = movie.get("imports_offset", 0)
if import_count > 100_000:
raise FormatError("unreasonable APT import count")
if import_count:
self.apt.require(imports_offset, import_count * 16)
for index in range(import_count):
record = imports_offset + index * 16
imports.append(
{
"index": index,
"offset": record,
"file": self.apt.cstring(self.apt.u32be(record)),
"name": self.apt.cstring(self.apt.u32be(record + 4)),
"character_id": self.apt.u32be(record + 8),
}
)
export_count = movie.get("export_count", 0)
exports_offset = movie.get("exports_offset", 0)
if export_count > 100_000:
raise FormatError("unreasonable APT export count")
if export_count:
self.apt.require(exports_offset, export_count * 8)
for index in range(export_count):
record = exports_offset + index * 8
exports.append(
{
"index": index,
"offset": record,
"name": self.apt.cstring(self.apt.u32be(record)),
"character_id": self.apt.u32be(record + 4),
}
)
count = movie.get("character_count", 0)
table = movie.get("characters_offset", 0)
if count > 100_000:
raise FormatError("unreasonable APT character count")
if count:
self.apt.require(table, count * 4)
for character_id in range(count):
parsed = self.character(
self.apt.u32be(table + character_id * 4), character_id
)
if parsed:
if "movie" in parsed:
parsed["frames"] = self.frames(parsed["movie"])
characters.append(parsed)
root["frames"] = self.frames(movie)
return {
"format": "skate3-apt",
"apt_source": str(self.apt_path),
"const_source": str(self.const_path),
"const_header": header,
"root": root,
"characters": characters,
"imports": imports,
"exports": exports,
}
def inspect_apt(apt_path: Path, const_path: Path) -> dict:
return AptInspector(apt_path, const_path).inspect()
+154
View File
@@ -0,0 +1,154 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from pathlib import Path, PurePosixPath
from .binary import FormatError, Reader, align
from .refpack import decompress
@dataclass(frozen=True)
class BigEntry:
index: int
path: str
offset: int
stored_size: int
unpacked_size: int
compression: int
class BigArchive:
"""Reader for the EB BIG v3 archives used by Skate 3 front-end data."""
def __init__(self, path: Path):
self.path = Path(path)
self.data = self.path.read_bytes()
self.reader = Reader(self.data, str(self.path))
self.entries = self._parse()
def _parse(self) -> list[BigEntry]:
r = self.reader
r.require(0, 48)
if r.u16be(0) != 0x4542:
raise FormatError("not an EB BIG archive")
version = r.u16be(2)
if version != 3:
raise FormatError(f"unsupported EB BIG version {version}")
count = r.u32be(4)
if count > 1_000_000:
raise FormatError(f"unreasonable BIG entry count {count}")
flags = r.u16be(8)
shift = r.u8(10)
if shift > 31:
raise FormatError(f"invalid BIG alignment shift {shift}")
index_size = r.u32be(12)
names_size = r.u32be(16)
name_record_size = r.u8(20)
directory_record_size = r.u8(21)
if name_record_size < 3 or directory_record_size < 2:
raise FormatError("invalid BIG name record sizing")
if index_size > len(self.data) or names_size > len(self.data) - index_size:
raise FormatError("BIG index/name blocks exceed file")
entry_size = 20 if flags & 1 else 16
entries_start = 48
entries_bytes = entry_size * count
compression_start = entries_start + align(entries_bytes, 16)
names_start = index_size
directories_start = names_start + align(name_record_size * count, 16)
r.require(entries_start, entries_bytes)
r.require(compression_start, count)
r.require(names_start, name_record_size * count)
directories: list[str] = []
directory_bytes_end = index_size + names_size
if directory_bytes_end > len(self.data):
raise FormatError("BIG directory block exceeds file")
cursor = directories_start
while cursor + directory_record_size <= directory_bytes_end:
directories.append(
r.bytes(cursor, directory_record_size)
.split(b"\0", 1)[0]
.decode("utf-8", errors="replace")
)
cursor += directory_record_size
result: list[BigEntry] = []
for index in range(count):
entry = entries_start + index * entry_size
file_offset = r.u32be(entry) << shift
declared_stored = r.u32be(entry + 4)
unpacked_size = r.u32be(entry + 8) or declared_stored
# EB v3 stores zero in the packed-size field for raw entries.
stored_size = declared_stored or unpacked_size
compression = r.u8(compression_start + index)
name_record = names_start + index * name_record_size
directory_index = r.u16be(name_record)
filename = (
r.bytes(name_record + 2, name_record_size - 2)
.split(b"\0", 1)[0]
.decode("utf-8", errors="replace")
)
directory = (
directories[directory_index]
if directory_index < len(directories)
else "."
)
archive_path = filename if directory in ("", ".") else f"{directory}/{filename}"
r.require(file_offset, stored_size)
result.append(
BigEntry(
index,
archive_path.replace("\\", "/"),
file_offset,
stored_size,
unpacked_size,
compression,
)
)
return result
def manifest(self) -> list[dict]:
return [asdict(entry) for entry in self.entries]
def read(self, entry: BigEntry) -> bytes:
packed = self.reader.bytes(entry.offset, entry.stored_size)
if entry.compression == 0:
if len(packed) != entry.unpacked_size:
raise FormatError(
f"{entry.path}: uncompressed size metadata does not match"
)
return packed
if entry.compression == 1:
return decompress(packed, entry.unpacked_size)
raise FormatError(
f"{entry.path}: compression {entry.compression} is not yet supported "
"(2=chunked RefPack, 3=chunked zlib, 4=Xbox LZX)"
)
@staticmethod
def _safe_relative(path: str) -> Path:
pure = PurePosixPath(path.replace("\\", "/"))
if pure.is_absolute() or not pure.parts or any(
part in ("", ".", "..") for part in pure.parts
):
raise FormatError(f"unsafe archive path {path!r}")
if ":" in pure.parts[0]:
raise FormatError(f"unsafe archive path {path!r}")
return Path(*pure.parts)
def extract(self, output: Path, force: bool = False) -> list[Path]:
output = Path(output).resolve()
output.mkdir(parents=True, exist_ok=True)
written = []
for entry in self.entries:
relative = self._safe_relative(entry.path)
target = (output / relative).resolve()
if output != target and output not in target.parents:
raise FormatError(f"archive path escapes output: {entry.path!r}")
target.parent.mkdir(parents=True, exist_ok=True)
if target.exists() and not force:
raise FileExistsError(f"refusing to overwrite {target}")
target.write_bytes(self.read(entry))
written.append(target)
return written
+61
View File
@@ -0,0 +1,61 @@
from __future__ import annotations
import struct
class FormatError(ValueError):
"""Raised when an input is malformed or outside supported bounds."""
class Reader:
def __init__(self, data: bytes, label: str = "input"):
self.data = data
self.label = label
def require(self, offset: int, size: int) -> None:
if offset < 0 or size < 0 or offset > len(self.data) - size:
raise FormatError(
f"{self.label}: range {offset:#x}+{size:#x} is outside "
f"{len(self.data):#x} bytes"
)
def bytes(self, offset: int, size: int) -> bytes:
self.require(offset, size)
return self.data[offset : offset + size]
def u8(self, offset: int) -> int:
self.require(offset, 1)
return self.data[offset]
def u16be(self, offset: int) -> int:
self.require(offset, 2)
return struct.unpack_from(">H", self.data, offset)[0]
def u32be(self, offset: int) -> int:
self.require(offset, 4)
return struct.unpack_from(">I", self.data, offset)[0]
def i32be(self, offset: int) -> int:
self.require(offset, 4)
return struct.unpack_from(">i", self.data, offset)[0]
def u64be(self, offset: int) -> int:
self.require(offset, 8)
return struct.unpack_from(">Q", self.data, offset)[0]
def f32be(self, offset: int) -> float:
self.require(offset, 4)
return struct.unpack_from(">f", self.data, offset)[0]
def cstring(self, offset: int, limit: int | None = None) -> str:
if offset <= 0 or offset >= len(self.data):
return ""
end_limit = len(self.data) if limit is None else min(len(self.data), offset + limit)
end = self.data.find(b"\0", offset, end_limit)
if end < 0:
end = end_limit
return self.data[offset:end].decode("utf-8", errors="replace")
def align(value: int, alignment: int) -> int:
return (value + alignment - 1) & ~(alignment - 1)
+196
View File
@@ -0,0 +1,196 @@
from __future__ import annotations
import re
from pathlib import Path
from .binary import FormatError
_FIELD = re.compile(r"(?m)^([A-Za-z][A-Za-z0-9]*):[ \t]*(.*)$")
def _fields(text: str) -> dict[str, str]:
matches = list(_FIELD.finditer(text))
result = {}
for index, match in enumerate(matches):
end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
value = text[match.start(2) : end].replace("\r", "").replace("\n", " ")
result[match.group(1)] = value.strip()
return result
def _number(value: str) -> int | float:
return float(value) if any(char in value for char in ".eE") else int(value)
def parse_bitmap_font(path: Path) -> dict:
path = Path(path)
fields = _fields(path.read_text(encoding="cp1252"))
required = {"Version", "FullName", "Family", "GlyphMetricsMap", "CharMapSet"}
missing = sorted(required - fields.keys())
if missing:
raise FormatError(f"{path}: missing bitmap-font fields: {', '.join(missing)}")
if fields["Version"] != "3":
raise FormatError(f"{path}: unsupported bitmap-font version")
metric_chunks = [chunk.strip() for chunk in fields["GlyphMetricsMap"].split(",")]
first = metric_chunks[0].split()
if len(first) != 10:
raise FormatError(f"{path}: malformed first glyph metric")
declared_glyphs = int(first.pop(0))
metric_chunks[0] = " ".join(first)
glyphs = []
for chunk in metric_chunks:
values = [int(value) for value in chunk.split()]
if len(values) != 9:
raise FormatError(f"{path}: malformed glyph metric {chunk!r}")
glyph = dict(
zip(
(
"glyph_index",
"texture_index",
"x",
"y",
"width",
"height",
"x_offset",
"y_offset",
"x_advance",
),
values,
)
)
# EA::Text::BmpGlyphMetrics stores X/Y as the glyph's pen origin.
# EATextBmpFont.cpp derives the physical atlas rectangle by adding the
# horizontal bearing and subtracting the vertical bearing.
atlas_left = glyph["x"] + glyph["x_offset"]
atlas_top = glyph["y"] - glyph["y_offset"]
glyph["atlas_bounds"] = [
atlas_left,
atlas_top,
atlas_left + glyph["width"],
atlas_top + glyph["height"],
]
glyphs.append(glyph)
if len(glyphs) != declared_glyphs:
raise FormatError(
f"{path}: declared {declared_glyphs} glyphs, found {len(glyphs)}"
)
char_chunks = [chunk.strip() for chunk in fields["CharMapSet"].split(",")]
first = char_chunks[0].split()
if len(first) != 3:
raise FormatError(f"{path}: malformed first character mapping")
declared_chars = int(first.pop(0))
char_chunks[0] = " ".join(first)
characters = []
for chunk in char_chunks:
values = [int(value) for value in chunk.split()]
if len(values) != 2:
raise FormatError(f"{path}: malformed character mapping {chunk!r}")
characters.append({"codepoint": values[0], "glyph_index": values[1]})
if len(characters) != declared_chars:
raise FormatError(
f"{path}: declared {declared_chars} characters, found {len(characters)}"
)
scalar_keys = (
"Weight",
"Stretch",
"Size",
"HAdvanceXMax",
"VAdvanceYMax",
"Ascent",
"Descent",
"Leading",
"Baseline",
"LineHeight",
"XHeight",
"CapsHeight",
"UnderlinePosition",
"UnderlineThickness",
"LinethroughPosition",
"LinethroughThickness",
)
metrics = {key: _number(fields[key]) for key in scalar_keys if key in fields}
textures = []
for key, value in fields.items():
if not key.startswith("Texture"):
continue
parts = value.split()
if len(parts) < 4:
raise FormatError(f"{path}: malformed texture declaration {value!r}")
textures.append(
{
"index": int(key[7:]),
"bytes": int(parts[0]),
"width": int(parts[1]),
"height": int(parts[2]),
"name": " ".join(parts[3:]),
}
)
return {
"format": "skate3-bitmap-font",
"source": str(path),
"full_name": fields["FullName"],
"family": fields["Family"],
"style": fields.get("Style", ""),
"smooth": fields.get("Smooth") == "Yes",
"fixed_pitch": fields.get("FixedPitch") == "Yes",
"metrics": metrics,
"glyphs": glyphs,
"characters": characters,
"textures": sorted(textures, key=lambda item: item["index"]),
}
def measure_bitmap_text(font: dict, text: str, height: float) -> dict:
mappings = {
item["codepoint"]: item["glyph_index"] for item in font["characters"]
}
glyphs = {item["glyph_index"]: item for item in font["glyphs"]}
source_ascent = float(font["metrics"]["Ascent"])
if source_ascent <= 0:
raise FormatError("bitmap font has a non-positive ascent")
# APT dynamic-text fontHeight is the Flash ascent height. The bitmap
# glyph bearings and advances are expressed in the bmpFont's source-pixel
# coordinate system, whose matching vertical metric is Ascent rather than
# the nominal point Size.
scale = height / source_ascent
pen = 0.0
minimum = 0.0
maximum = 0.0
used = []
for character in text:
codepoint = ord(character)
glyph_index = mappings.get(codepoint)
if glyph_index is None or glyph_index not in glyphs:
raise FormatError(
f"bitmap font {font['family']!r} lacks codepoint {codepoint}"
)
glyph = glyphs[glyph_index]
left = pen + glyph["x_offset"] * scale
right = left + glyph["width"] * scale
minimum = min(minimum, left)
maximum = max(maximum, right)
used.append(
{
"codepoint": codepoint,
"glyph_index": glyph_index,
"pen_x": pen,
"left": left,
"right": right,
}
)
pen += glyph["x_advance"] * scale
# Flash TextField autoSize retains the two-pixel gutter on each side.
width = max(pen, maximum) - min(0.0, minimum) + 4.0
return {
"width": width,
"advance": pen,
"ink_bounds_x": [minimum, maximum],
"height": height,
"scale": scale,
"glyphs": used,
}
+80
View File
@@ -0,0 +1,80 @@
from __future__ import annotations
from pathlib import Path
from .binary import FormatError, Reader
RENDER_TYPES = {
0: "line",
1: "solid",
2: "texture_clamped",
3: "texture_wrapped",
}
def parse_geo(path: Path) -> dict:
data = Path(path).read_bytes()
r = Reader(data, str(path))
# Retail bundles use an eight-byte header as the canonical empty GEO.
if len(data) == 8 and r.u32be(4) == 0:
return {"format": "skate3-geo", "source": str(path), "shapes": []}
r.require(0, 12)
count = r.u32be(4)
if count > 4096:
raise FormatError(f"unreasonable GEO shape count {count}")
r.require(8, count * 4)
shapes = []
for index in range(count):
shape_offset = r.u32be(8 + index * 4)
if not shape_offset:
continue
r.require(shape_offset, 8)
shape_id = r.u32be(shape_offset)
unit_count = r.u32be(shape_offset + 4)
if unit_count > 65536:
raise FormatError(f"shape {shape_id}: unreasonable unit count")
r.require(shape_offset + 8, unit_count * 4)
units = []
all_points: list[list[float]] = []
for unit_index in range(unit_count):
unit_offset = r.u32be(shape_offset + 8 + unit_index * 4)
r.require(unit_offset, 0x34)
render_type = r.u32be(unit_offset)
color = [r.f32be(unit_offset + 4 + channel * 4) for channel in range(4)]
texture_id = r.u32be(unit_offset + 0x14)
uv_matrix = [
r.f32be(unit_offset + 0x18 + component * 4)
for component in range(6)
]
primitive_count = r.u32be(unit_offset + 0x30)
if primitive_count > 1_000_000:
raise FormatError(f"shape {shape_id}: unreasonable primitive count")
r.require(unit_offset + 0x34, primitive_count * 0x18)
triangles = []
for primitive in range(primitive_count):
base = unit_offset + 0x34 + primitive * 0x18
triangle = [
[r.f32be(base + vertex * 8), r.f32be(base + vertex * 8 + 4)]
for vertex in range(3)
]
triangles.append(triangle)
all_points.extend(triangle)
units.append(
{
"render_type": render_type,
"render_type_name": RENDER_TYPES.get(
render_type, f"unknown_{render_type}"
),
"color": color,
"texture_id": texture_id,
"uv_matrix": uv_matrix,
"triangles": triangles,
}
)
bounds = None
if all_points:
xs = [point[0] for point in all_points]
ys = [point[1] for point in all_points]
bounds = [min(xs), min(ys), max(xs), max(ys)]
shapes.append({"id": shape_id, "bounds": bounds, "units": units})
return {"format": "skate3-geo", "source": str(path), "shapes": shapes}
+76
View File
@@ -0,0 +1,76 @@
from __future__ import annotations
import struct
from dataclasses import dataclass
from pathlib import Path
from .binary import FormatError
@dataclass(frozen=True)
class LanguageTable:
source: str
declared_count: int
pool_offset: int
strings: tuple[str, ...]
def parse_language_table(path: Path) -> LanguageTable:
path = Path(path)
data = path.read_bytes()
if len(data) < 28:
raise FormatError(f"{path}: language table is shorter than its header")
declared_count = struct.unpack_from("<I", data, 8)[0]
header_size = struct.unpack_from("<I", data, 12)[0]
pool_offset = struct.unpack_from("<I", data, 16)[0]
if header_size != 28:
raise FormatError(f"{path}: unsupported language header size {header_size}")
# The pool descriptor occupies eight bytes at pool_offset. The actual
# null-separated byte strings follow it.
strings_offset = pool_offset + 8
if strings_offset > len(data):
raise FormatError(f"{path}: string pool lies outside the file")
raw_strings = data[strings_offset:].split(b"\0")
# The UI uses byte-oriented font character maps, including custom glyphs
# in positions that are undefined in Windows-1252. Latin-1 gives every
# source byte a lossless, one-to-one Unicode codepoint for the manifest.
strings = tuple(value.decode("latin-1") for value in raw_strings)
expected = declared_count + 2
if len(strings) != expected:
raise FormatError(
f"{path}: expected {expected} pool strings, found {len(strings)}"
)
return LanguageTable(str(path), declared_count, pool_offset, strings)
def pair_language_tables(labels_path: Path, values_path: Path) -> dict:
labels = parse_language_table(labels_path)
values = parse_language_table(values_path)
if len(labels.strings) != len(values.strings):
raise FormatError(
"language label/value pools have different numbers of strings"
)
entries = []
seen: dict[str, int] = {}
duplicate_labels = []
for index, (label, value) in enumerate(zip(labels.strings, values.strings)):
if label:
if label in seen:
duplicate_labels.append(
{"label": label, "first_index": seen[label], "index": index}
)
else:
seen[label] = index
entries.append({"index": index, "label": label, "value": value})
return {
"format": "skate3-language-pair",
"encoding": "iso-8859-1-byte-preserving",
"labels_source": labels.source,
"values_source": values.source,
"declared_count": labels.declared_count,
"pool_entries": len(entries),
"entries": entries,
"duplicate_labels": duplicate_labels,
}
+375
View File
@@ -0,0 +1,375 @@
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from .apt import inspect_apt
from .bitmap_font import parse_bitmap_font
from .big import BigArchive, BigEntry
from .binary import FormatError
from .geo import parse_geo
from .language import pair_language_tables
from .rw4 import extract_textures
CORE_ARCHIVES = ("fedata.big", "fetexture.big", "miscboot.big")
DYNAMIC_ARCHIVE = "fedynamic.big"
DEFAULT_PREFIXES = (
"data/fe/source/screens/",
"data/fe/source/controls/",
)
UI_EXTENSIONS = {
".apt",
".bmpfont",
".const",
".geo",
".rx2",
".rps3",
".jpg",
".png",
}
SUPPORT_PATHS = {
"data/fe/languages/english/language_english_global_skate3ng.bin",
"data/fe/languages/labels/language_labels_global_skate3ng.bin",
}
SUPPORT_PREFIXES = ("data/fe/fonts/",)
CACHE_VERSION = 3
@dataclass(frozen=True)
class SelectedEntry:
archive_name: str
archive: BigArchive
entry: BigEntry
def _sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _write_json(path: Path, value: object, force: bool) -> None:
if path.exists() and not force:
raise FileExistsError(f"refusing to overwrite {path}; pass --force")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
def find_big_directory(game_root: Path) -> Path:
root = Path(game_root).resolve()
candidates = (root / "data" / "big", root / "big", root)
for candidate in candidates:
if (candidate / "fedata.big").is_file():
return candidate
raise FileNotFoundError(
f"could not find data/big/fedata.big beneath the supplied game root: {root}"
)
def is_selected_ui_path(path: str, prefixes: tuple[str, ...]) -> bool:
normalized = path.replace("\\", "/").lower()
if normalized in SUPPORT_PATHS:
return True
if any(normalized.startswith(prefix) for prefix in SUPPORT_PREFIXES):
return Path(normalized).suffix in UI_EXTENSIONS
if Path(normalized).suffix not in UI_EXTENSIONS:
return False
return any(normalized.startswith(prefix.lower()) for prefix in prefixes)
def cache_is_current(
game_root: Path,
output: Path,
*,
include_dynamic: bool,
prefixes: tuple[str, ...],
decode_png: bool,
) -> bool:
manifest_path = Path(output).resolve() / "manifest.json"
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
if (
manifest.get("format") != "skate3-native-ui-asset-cache"
or manifest.get("version") != CACHE_VERSION
or manifest.get("prefixes") != list(prefixes)
):
return False
if decode_png and manifest.get("summary", {}).get("textures_decoded", 0) == 0:
return False
big_directory = find_big_directory(game_root)
expected = list(CORE_ARCHIVES)
if include_dynamic:
expected.append(DYNAMIC_ARCHIVE)
records = {record["name"]: record for record in manifest.get("archives", [])}
if set(records) != set(expected):
return False
for name in expected:
source = big_directory / name
record = records[name]
if (
not source.is_file()
or source.stat().st_size != record.get("bytes")
or _sha256(source.read_bytes()) != record.get("sha256")
):
return False
return True
except (FileNotFoundError, KeyError, OSError, ValueError, json.JSONDecodeError):
return False
def _safe_output_path(root: Path, archive_path: str) -> Path:
relative = BigArchive._safe_relative(archive_path)
target = (root / relative).resolve()
if root != target and root not in target.parents:
raise FormatError(f"archive path escapes output: {archive_path!r}")
return target
def extract_project(
game_root: Path,
output: Path,
*,
include_dynamic: bool = False,
prefixes: tuple[str, ...] = DEFAULT_PREFIXES,
decode_png: bool = True,
force: bool = False,
update: bool = False,
) -> dict:
"""Build a local, exact-asset UI cache from an installed Skate 3 copy."""
game_root = Path(game_root).resolve()
output = Path(output).resolve()
if game_root == output or game_root in output.parents:
raise ValueError("output must be outside the read-only game-data tree")
if update and cache_is_current(
game_root,
output,
include_dynamic=include_dynamic,
prefixes=prefixes,
decode_png=decode_png,
):
manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8"))
manifest["cache_reused"] = True
return manifest
if update:
force = True
if output.exists() and any(output.iterdir()) and not force:
raise FileExistsError(f"{output} is not empty; pass --force to update it")
output.mkdir(parents=True, exist_ok=True)
big_directory = find_big_directory(game_root)
archive_names = list(CORE_ARCHIVES)
if include_dynamic:
archive_names.append(DYNAMIC_ARCHIVE)
archives: list[tuple[str, BigArchive]] = []
archive_records = []
for name in archive_names:
source = big_directory / name
if not source.is_file():
raise FileNotFoundError(f"required UI archive is missing: {source}")
archive = BigArchive(source)
archives.append((name, archive))
archive_records.append(
{
"name": name,
"source": str(source),
"bytes": source.stat().st_size,
"sha256": _sha256(source.read_bytes()),
"entries": len(archive.entries),
}
)
# Later archives are preferred. fedynamic.big is a superset when requested,
# while fetexture.big supplies RX2 files absent from fedata.big.
candidates: dict[str, list[SelectedEntry]] = {}
for archive_name, archive in archives:
for entry in archive.entries:
if is_selected_ui_path(entry.path, prefixes):
key = entry.path.replace("\\", "/").lower()
candidates.setdefault(key, []).append(
SelectedEntry(archive_name, archive, entry)
)
selected = {key: choices[-1] for key, choices in candidates.items()}
raw_root = output / "raw"
file_records = []
extracted: dict[str, Path] = {}
for key, source in sorted(selected.items()):
data = source.archive.read(source.entry)
target = _safe_output_path(raw_root, source.entry.path)
if target.exists() and not force:
raise FileExistsError(f"refusing to overwrite {target}; pass --force")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
extracted[key] = target
file_records.append(
{
"path": source.entry.path,
"archive": source.archive_name,
"bytes": len(data),
"sha256": _sha256(data),
"alternate_sources": [
choice.archive_name for choice in candidates[key][:-1]
],
}
)
bundles: dict[str, dict[str, Path]] = {}
display_names: dict[str, str] = {}
for key, path in extracted.items():
stem = str(Path(key).with_suffix("")).replace("\\", "/")
display_names.setdefault(
stem, str(Path(selected[key].entry.path).with_suffix("")).replace("\\", "/")
)
bundles.setdefault(stem, {})[path.suffix.lower()] = path
bundle_records = []
errors = []
for key, parts in sorted(bundles.items()):
display_name = display_names[key]
record: dict = {
"name": display_name,
"files": {
suffix.lstrip("."): path.relative_to(output).as_posix()
for suffix, path in sorted(parts.items())
},
}
metadata_base = _safe_output_path(output / "metadata", display_name)
asset_base = _safe_output_path(output / "assets", display_name)
if ".apt" in parts and ".const" in parts:
try:
apt_data = inspect_apt(parts[".apt"], parts[".const"])
apt_target = metadata_base.with_suffix(".apt.json")
_write_json(apt_target, apt_data, force)
movie = (apt_data.get("root") or {}).get("movie", {})
record["timeline"] = {
"metadata": apt_target.relative_to(output).as_posix(),
"frames": movie.get("frame_count", 0),
"milliseconds_per_frame": movie.get("milliseconds_per_frame", 0),
"characters": len(apt_data.get("characters", [])),
}
except (FormatError, OSError, ValueError) as error:
message = f"{display_name}: APT: {error}"
errors.append(message)
record.setdefault("errors", []).append(message)
if ".geo" in parts:
try:
geo_data = parse_geo(parts[".geo"])
geo_target = metadata_base.with_suffix(".geo.json")
_write_json(geo_target, geo_data, force)
record["geometry"] = {
"metadata": geo_target.relative_to(output).as_posix(),
"shapes": len(geo_data.get("shapes", [])),
}
except (FormatError, OSError, ValueError) as error:
message = f"{display_name}: GEO: {error}"
errors.append(message)
record.setdefault("errors", []).append(message)
if ".bmpfont" in parts:
try:
font_data = parse_bitmap_font(parts[".bmpfont"])
font_target = metadata_base.with_suffix(".font.json")
_write_json(font_target, font_data, force)
record["font"] = {
"metadata": font_target.relative_to(output).as_posix(),
"family": font_data["family"],
"glyphs": len(font_data["glyphs"]),
"characters": len(font_data["characters"]),
}
except (FormatError, OSError, ValueError) as error:
message = f"{display_name}: bitmap font: {error}"
errors.append(message)
record.setdefault("errors", []).append(message)
arena = parts.get(".rx2") or parts.get(".rps3")
if arena:
try:
texture_manifest = extract_textures(
arena, asset_base, force=force, decode_png=decode_png
)
texture_target = asset_base / "manifest.json"
_write_json(texture_target, texture_manifest, force)
record["textures"] = {
"manifest": texture_target.relative_to(output).as_posix(),
"platform": texture_manifest["platform"],
"count": len(texture_manifest["textures"]),
"decoded": sum(
"preview_file" in texture
for texture in texture_manifest["textures"]
),
}
for texture in texture_manifest["textures"]:
if "preview_error" in texture:
message = (
f"{display_name}: texture {texture['index']}: "
f"{texture['preview_error']}"
)
errors.append(message)
except (FormatError, OSError, ValueError) as error:
message = f"{display_name}: RW4: {error}"
errors.append(message)
record.setdefault("errors", []).append(message)
bundle_records.append(record)
labels_key = (
"data/fe/languages/labels/language_labels_global_skate3ng.bin"
)
english_key = (
"data/fe/languages/english/language_english_global_skate3ng.bin"
)
language_record = None
if labels_key in extracted and english_key in extracted:
try:
language_data = pair_language_tables(
extracted[labels_key], extracted[english_key]
)
language_target = output / "metadata" / "languages" / "english_global.json"
_write_json(language_target, language_data, force)
language_record = {
"metadata": language_target.relative_to(output).as_posix(),
"entries": language_data["pool_entries"],
"duplicate_labels": len(language_data["duplicate_labels"]),
}
except (FormatError, OSError, ValueError) as error:
message = f"English global language pair: {error}"
errors.append(message)
manifest = {
"format": "skate3-native-ui-asset-cache",
"version": CACHE_VERSION,
"notice": (
"Exact retail assets extracted locally from the user's installed game; "
"do not redistribute this cache."
),
"game_root": str(game_root),
"output": str(output),
"prefixes": list(prefixes),
"archives": archive_records,
"selected_files": file_records,
"bundles": bundle_records,
"languages": {"english_global": language_record}
if language_record
else {},
"errors": errors,
"summary": {
"files": len(file_records),
"bundles": len(bundle_records),
"timelines": sum("timeline" in bundle for bundle in bundle_records),
"geometry": sum("geometry" in bundle for bundle in bundle_records),
"bitmap_fonts": sum("font" in bundle for bundle in bundle_records),
"languages": int(language_record is not None),
"texture_arenas": sum("textures" in bundle for bundle in bundle_records),
"textures_decoded": sum(
bundle.get("textures", {}).get("decoded", 0)
for bundle in bundle_records
),
"errors": len(errors),
},
}
_write_json(output / "manifest.json", manifest, force)
return manifest
+93
View File
@@ -0,0 +1,93 @@
from __future__ import annotations
from .binary import FormatError
def _copy_backref(output: bytearray, distance: int, count: int) -> None:
if distance <= 0 or distance > len(output):
raise FormatError(
f"RefPack back-reference distance {distance} exceeds "
f"{len(output)} output bytes"
)
for _ in range(count):
output.append(output[-distance])
def decompress(data: bytes, expected_size: int | None = None) -> bytes:
"""Decode an EA RefPack stream, including its usual 0x10FB/0x90FB header."""
if len(data) < 2:
raise FormatError("RefPack stream is too short")
pos = 0
header_size = None
if data[1] == 0xFB and data[0] in (0x10, 0x90):
if data[0] & 0x80:
if len(data) < 6:
raise FormatError("truncated extended RefPack header")
header_size = int.from_bytes(data[2:6], "big")
pos = 6
else:
if len(data) < 5:
raise FormatError("truncated RefPack header")
header_size = int.from_bytes(data[2:5], "big")
pos = 5
if expected_size is None:
expected_size = header_size
elif header_size not in (None, expected_size):
raise FormatError(
f"RefPack header size {header_size} differs from archive size "
f"{expected_size}"
)
output = bytearray()
def literal(count: int) -> None:
nonlocal pos
if pos + count > len(data):
raise FormatError("truncated RefPack literal")
output.extend(data[pos : pos + count])
pos += count
while pos < len(data):
control = data[pos]
pos += 1
if control < 0x80:
if pos >= len(data):
raise FormatError("truncated two-byte RefPack command")
b1 = data[pos]
pos += 1
literal(control & 0x03)
distance = ((control & 0x60) << 3) + b1 + 1
count = ((control >> 2) & 0x07) + 3
_copy_backref(output, distance, count)
elif control < 0xC0:
if pos + 2 > len(data):
raise FormatError("truncated three-byte RefPack command")
b1, b2 = data[pos], data[pos + 1]
pos += 2
literal(b1 >> 6)
distance = ((b1 & 0x3F) << 8) + b2 + 1
_copy_backref(output, distance, (control & 0x3F) + 4)
elif control < 0xE0:
if pos + 3 > len(data):
raise FormatError("truncated four-byte RefPack command")
b1, b2, b3 = data[pos], data[pos + 1], data[pos + 2]
pos += 3
literal(control & 0x03)
distance = ((control & 0x10) << 12) + (b1 << 8) + b2 + 1
count = ((control & 0x0C) << 6) + b3 + 5
_copy_backref(output, distance, count)
elif control < 0xFC:
literal(((control & 0x1F) << 2) + 4)
else:
literal(control & 0x03)
break
if expected_size is not None and len(output) > expected_size:
raise FormatError("RefPack output exceeds declared size")
if expected_size is not None and len(output) != expected_size:
raise FormatError(
f"RefPack produced {len(output)} bytes; expected {expected_size}"
)
return bytes(output)
+161
View File
@@ -0,0 +1,161 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path, PurePath
from .binary import FormatError, Reader
from .xenos import decode_rgba, parse_fetch_constant, write_png
TOC_TYPE = 0x00EB000B
@dataclass
class Texture:
index: int
name: str
guid: int
data: bytes
platform: str
metadata: dict
def _safe_name(name: str, index: int) -> str:
leaf = PurePath(name.replace("\\", "/")).name
filtered = "".join(c if c.isalnum() or c in "._-" else "_" for c in leaf)
return filtered or f"texture_{index:04d}"
def read_textures(path: Path) -> tuple[dict, list[Texture]]:
path = Path(path)
raw = path.read_bytes()
r = Reader(raw, str(path))
r.require(0, 68)
if r.bytes(0, 4) != b"\x89RW4":
raise FormatError("not a RenderWare 4 arena")
platform_magic = r.bytes(4, 4)
if platform_magic == b"ps3\0":
platform = "PS3"
elif platform_magic == b"xb2\0":
platform = "XBOX360"
else:
raise FormatError(f"unknown RW4 platform magic {platform_magic!r}")
arena_offset = 32
entry_count = r.u32be(arena_offset)
dictionary_offset = r.u32be(arena_offset + 16)
resource_header_size = r.u32be(arena_offset + 36)
if entry_count > 1_000_000:
raise FormatError("unreasonable RW4 dictionary count")
r.require(dictionary_offset, entry_count * 24)
dictionary = []
for index in range(entry_count):
offset = dictionary_offset + index * 24
dictionary.append(
{
"pointer": r.u32be(offset),
"relocation": r.u32be(offset + 4),
"size": r.u32be(offset + 8),
"alignment": r.u32be(offset + 12),
"type_index": r.u32be(offset + 16),
"type_id": r.u32be(offset + 20),
}
)
toc = next((item for item in dictionary if item["type_id"] == TOC_TYPE), None)
if not toc:
raise FormatError("RW4 arena has no table of contents")
toc_offset = toc["pointer"]
r.require(toc_offset, 20)
item_count = r.u32be(toc_offset)
item_array = r.u32be(toc_offset + 4)
if item_count > 1_000_000:
raise FormatError("unreasonable RW4 TOC count")
r.require(toc_offset + item_array, item_count * 24)
textures = []
for item_index in range(item_count):
item = toc_offset + item_array + item_index * 24
name_offset = r.u32be(item)
name = r.cstring(toc_offset + name_offset)
guid = r.u64be(item + 8)
dictionary_index = r.u32be(item + 20) - 1
if dictionary_index < 0 or dictionary_index + 1 >= len(dictionary):
continue
payload = dictionary[dictionary_index]
info = dictionary[dictionary_index + 1]
data_offset = payload["pointer"] + resource_header_size
texture_data = r.bytes(data_offset, payload["size"])
metadata = {"raw_size": len(texture_data)}
if platform == "PS3":
r.require(info["pointer"], 40)
metadata.update(
{
"format": r.u8(info["pointer"]),
"mipmaps": r.u8(info["pointer"] + 1),
"remap": r.u32be(info["pointer"] + 4),
"width": r.u16be(info["pointer"] + 8),
"height": r.u16be(info["pointer"] + 10),
"texture_offset": r.u32be(info["pointer"] + 20),
"store_type": r.u32be(info["pointer"] + 28),
}
)
else:
r.require(info["pointer"] + 28, 24)
fetch_data = r.bytes(info["pointer"] + 28, 24)
metadata["fetch_constant_hex"] = fetch_data.hex()
metadata.update(parse_fetch_constant(fetch_data).manifest())
textures.append(
Texture(item_index, name, guid, texture_data, platform, metadata)
)
manifest = {
"format": "renderware4-arena",
"source": str(path),
"platform": platform,
"arena_id": r.u32be(28),
"dictionary_entries": entry_count,
}
return manifest, textures
def extract_textures(
path: Path, output: Path, force: bool = False, decode_png: bool = False
) -> dict:
manifest, textures = read_textures(path)
output = Path(output)
output.mkdir(parents=True, exist_ok=True)
entries = []
for texture in textures:
name = _safe_name(texture.name, texture.index)
target = output / f"{texture.index:04d}_{name}.texture.bin"
if target.exists() and not force:
raise FileExistsError(f"refusing to overwrite {target}")
target.write_bytes(texture.data)
entry = {
"index": texture.index,
"name": texture.name,
"guid": f"{texture.guid:016x}",
"raw_file": target.name,
**texture.metadata,
}
if decode_png and texture.platform == "XBOX360":
preview = output / f"{texture.index:04d}_{name}.png"
if preview.exists() and not force:
raise FileExistsError(f"refusing to overwrite {preview}")
try:
fetch = parse_fetch_constant(
bytes.fromhex(texture.metadata["fetch_constant_hex"])
)
rgba = decode_rgba(texture.data, fetch)
rgba_target = output / f"{texture.index:04d}_{name}.rgba"
if rgba_target.exists() and not force:
raise FileExistsError(f"refusing to overwrite {rgba_target}")
rgba_target.write_bytes(rgba)
write_png(preview, fetch.width, fetch.height, rgba)
entry["rgba_file"] = rgba_target.name
entry["preview_file"] = preview.name
except (FormatError, ValueError) as error:
entry["preview_error"] = str(error)
entries.append(entry)
manifest["textures"] = entries
return manifest
+430
View File
@@ -0,0 +1,430 @@
from __future__ import annotations
import json
from pathlib import Path, PurePosixPath
from typing import Callable
from .binary import FormatError
from .timeline import resolve_display_list
IDENTITY_MATRIX = [1.0, 0.0, 0.0, 1.0, 0.0, 0.0]
def compose_matrix(parent: list[float], local: list[float]) -> list[float]:
pa, pb, pc, pd, ptx, pty = parent
la, lb, lc, ld, ltx, lty = local
return [
pa * la + pc * lb,
pb * la + pd * lb,
pa * lc + pc * ld,
pb * lc + pd * ld,
pa * ltx + pc * lty + ptx,
pb * ltx + pd * lty + pty,
]
def transform_point(matrix: list[float], point: list[float]) -> list[float]:
return [
matrix[0] * point[0] + matrix[2] * point[1] + matrix[4],
matrix[1] * point[0] + matrix[3] * point[1] + matrix[5],
]
class AssetCache:
def __init__(self, root: Path):
self.root = Path(root).resolve()
self.manifest = json.loads(
(self.root / "manifest.json").read_text(encoding="utf-8")
)
if self.manifest.get("format") != "skate3-native-ui-asset-cache":
raise FormatError(f"{self.root}: not a native UI asset cache")
self.bundles = {
self.normalize_bundle(record["name"]): record
for record in self.manifest["bundles"]
}
self._loaded: dict[str, dict] = {}
self.font_aliases: dict[str, dict] = {}
def add_font_alias(self, source: str, target: str, provenance: dict) -> None:
self.font_aliases[source] = {
"target": target,
"provenance": provenance,
}
@staticmethod
def normalize_bundle(value: str) -> str:
normalized = value.replace("\\", "/").lower()
if normalized.startswith("source/"):
normalized = "data/fe/" + normalized
return str(PurePosixPath(normalized))
def load_bundle(self, name: str) -> dict:
key = self.normalize_bundle(name)
if key in self._loaded:
return self._loaded[key]
record = self.bundles.get(key)
if not record or "timeline" not in record:
raise FormatError(f"APT bundle is absent from cache: {name}")
apt = json.loads(
(self.root / record["timeline"]["metadata"]).read_text(encoding="utf-8")
)
geo = {"shapes": []}
if "geometry" in record:
geo = json.loads(
(self.root / record["geometry"]["metadata"]).read_text(
encoding="utf-8"
)
)
textures = {"textures": []}
if "textures" in record:
textures = json.loads(
(self.root / record["textures"]["manifest"]).read_text(
encoding="utf-8"
)
)
value = {
"key": key,
"record": record,
"apt": apt,
"characters": {item["id"]: item for item in apt["characters"]},
"imports": {
item["character_id"]: item for item in apt.get("imports", [])
},
"exports": {},
"shapes": {item["id"]: item for item in geo["shapes"]},
"textures": textures,
}
for item in apt.get("exports", []):
value["exports"].setdefault(item["name"], item["character_id"])
self._loaded[key] = value
return value
def resolve_character(self, bundle_name: str, character_id: int) -> tuple[dict, dict]:
bundle = self.load_bundle(bundle_name)
character = bundle["characters"].get(character_id)
if character:
return bundle, character
imported = bundle["imports"].get(character_id)
if not imported:
raise FormatError(
f"{bundle['key']}: character {character_id} is neither local nor imported"
)
imported_bundle = self.load_bundle(imported["file"])
exported_id = imported_bundle["exports"].get(imported["name"])
if exported_id is None:
raise FormatError(
f"{imported_bundle['key']}: export {imported['name']!r} is absent"
)
character = imported_bundle["characters"].get(exported_id)
if not character:
raise FormatError(
f"{imported_bundle['key']}: exported character {exported_id} is absent"
)
return imported_bundle, character
def font_asset(self, family: str) -> dict | None:
requested_family = family
alias = self.font_aliases.get(family)
if alias:
family = alias["target"]
matches = []
for record in self.manifest["bundles"]:
font = record.get("font")
if font and font["family"] == family:
matches.append(record)
if len(matches) != 1:
return None
record = matches[0]
metadata = json.loads(
(self.root / record["font"]["metadata"]).read_text(encoding="utf-8")
)
texture = json.loads(
(self.root / record["textures"]["manifest"]).read_text(encoding="utf-8")
)["textures"][0]
result = {
"bundle": record["name"],
"metadata": record["font"]["metadata"],
"texture": (
Path(record["textures"]["manifest"]).parent
/ texture["rgba_file"]
).as_posix(),
"preview": (
Path(record["textures"]["manifest"]).parent
/ texture["preview_file"]
).as_posix(),
"definition": metadata,
}
if alias:
result["requested_family"] = requested_family
result["alias"] = alias
return result
def texture_for_id(self, bundle: dict, texture_id: int) -> dict:
matches = []
for item in bundle["textures"]["textures"]:
basename = item["name"].replace("\\", "/").rsplit("/", 1)[-1]
stem = basename.rsplit(".", 1)[0]
if stem == str(texture_id):
matches.append(item)
if len(matches) != 1:
raise FormatError(
f"{bundle['key']}: expected one texture resource {texture_id}, "
f"found {len(matches)}"
)
item = matches[0]
manifest_dir = Path(bundle["record"]["textures"]["manifest"]).parent
return {
"resource_name": item["name"],
"width": item["width"],
"height": item["height"],
"preview": (manifest_dir / item["preview_file"]).as_posix(),
"rgba": (manifest_dir / item["rgba_file"]).as_posix(),
}
StateProvider = Callable[[str, dict, dict], dict]
class SceneFlattener:
def __init__(
self,
cache: AssetCache,
state_provider: StateProvider,
*,
include_zero_alpha: bool = False,
):
self.cache = cache
self.state_provider = state_provider
self.include_zero_alpha = include_zero_alpha
self.primitives: list[dict] = []
self.text: list[dict] = []
self.unresolved: list[dict] = []
self.draw_order = 0
def flatten(self, bundle_name: str, character_id: int) -> dict:
self._visit(
bundle_name,
character_id,
"/",
IDENTITY_MATRIX,
1.0,
[],
)
return {
"primitives": self.primitives,
"text": self.text,
"unresolved": self.unresolved,
}
def _visit(
self,
bundle_name: str,
character_id: int,
path: str,
world_matrix: list[float],
alpha: float,
stack: list[tuple[str, int]],
) -> None:
try:
bundle, character = self.cache.resolve_character(bundle_name, character_id)
except FormatError as error:
self.unresolved.append({"path": path, "reason": str(error)})
return
key = (bundle["key"], character["id"])
if key in stack:
self.unresolved.append({"path": path, "reason": "recursive character graph"})
return
state = self.state_provider(path, bundle, character)
if state.get("visible") is False:
return
alpha *= float(state.get("alpha", 1.0))
if alpha <= 0 and not self.include_zero_alpha:
return
type_name = character["type_name"]
if type_name == "shape":
self._shape(bundle, character, path, world_matrix, alpha)
return
if type_name in ("text", "static_text"):
self._text(bundle, character, path, world_matrix, alpha, state)
return
if type_name not in ("sprite", "animation"):
return
frame_count = character["movie"]["frame_count"]
frame = state.get("frame")
if frame is None:
if frame_count == 1:
frame = 0
elif self.include_zero_alpha and alpha <= 0:
# Retain a stable, non-rendering placeholder for authored
# layers whose parent colour transform currently hides them.
# Their frame cannot affect pixels until an extracted state
# explicitly makes the layer visible.
frame = 0
else:
self.unresolved.append(
{
"path": path,
"bundle": bundle["key"],
"character_id": character["id"],
"reason": "multi-frame character has no extracted runtime state",
}
)
return
try:
display = resolve_display_list(character, int(frame))
except FormatError as error:
self.unresolved.append({"path": path, "reason": str(error)})
return
self.unresolved.extend(
{"path": path, **item} for item in display["unresolved"]
)
for item in display["objects"]:
properties = item["properties"]
child_id = properties.get("character_id")
matrix = properties.get("matrix")
if child_id is None or matrix is None:
continue
name = properties.get("name") or f"depth_{item['depth']}"
child_path = path.rstrip("/") + "/" + name
try:
child_bundle, child_character = self.cache.resolve_character(
bundle["key"], child_id
)
child_state = self.state_provider(
child_path, child_bundle, child_character
)
except FormatError as error:
self.unresolved.append({"path": child_path, "reason": str(error)})
continue
matrix = list(child_state.get("matrix", matrix))
child_alpha = alpha
color = properties.get("color_transform_argb")
if color:
child_alpha *= color[0] / 255.0
child_alpha *= float(child_state.get("alpha", 1.0))
if child_state.get("visible") is False or (
child_alpha <= 0 and not self.include_zero_alpha
):
continue
self._visit(
bundle["key"],
child_id,
child_path,
compose_matrix(world_matrix, matrix),
child_alpha,
stack + [key],
)
def _shape(
self,
bundle: dict,
character: dict,
path: str,
matrix: list[float],
alpha: float,
) -> None:
shape = bundle["shapes"].get(character["id"])
if not shape:
self.unresolved.append(
{
"path": path,
"reason": f"shape {character['id']} has no GEO record",
}
)
return
for unit_index, unit in enumerate(shape["units"]):
texture = None
if unit["render_type_name"].startswith("texture"):
try:
_, bitmap = self.cache.resolve_character(
bundle["key"], unit["texture_id"]
)
texture = self.cache.texture_for_id(
bundle, bitmap["bitmap"]["texture_id"]
)
except (FormatError, KeyError) as error:
self.unresolved.append({"path": path, "reason": str(error)})
continue
triangles = []
for triangle in unit["triangles"]:
vertices = []
for point in triangle:
vertex = {"position": transform_point(matrix, point)}
if texture:
uv_point = transform_point(unit["uv_matrix"], point)
vertex["uv"] = [
uv_point[0] / texture["width"],
uv_point[1] / texture["height"],
]
vertices.append(vertex)
triangles.append(vertices)
self.primitives.append(
{
"draw_order": self.draw_order,
"path": path,
"bundle": bundle["key"],
"character_id": character["id"],
"unit": unit_index,
"render_type": unit["render_type_name"],
"matrix": matrix,
"color": unit["color"][:3] + [unit["color"][3] * alpha],
"texture": texture,
"triangles": triangles,
}
)
self.draw_order += 1
def _text(
self,
bundle: dict,
character: dict,
path: str,
matrix: list[float],
alpha: float,
state: dict,
) -> None:
if character["type_name"] != "text":
self.unresolved.append(
{"path": path, "reason": "static text flattening is not implemented"}
)
return
definition = character["text"]
font_character = bundle["characters"].get(definition["font_id"])
if not font_character or "font" not in font_character:
self.unresolved.append(
{"path": path, "reason": "text font declaration is absent"}
)
return
family = font_character["font"]["name"]
font_asset = self.cache.font_asset(family)
if not font_asset:
self.unresolved.append(
{
"path": path,
"reason": f"no exact packaged bitmap font for {family!r}",
}
)
self.text.append(
{
"draw_order": self.draw_order,
"path": path,
"bundle": bundle["key"],
"character_id": character["id"],
"matrix": matrix,
"bounds": character["bounds"],
"font_family": family,
"font_asset": font_asset,
"font_height": definition["font_height"],
"alignment": definition["alignment"],
"multiline": definition["multiline"],
"word_wrap": definition["word_wrap"],
"color_argb": state.get("color_argb", definition["color_argb"]),
"alpha": alpha,
"value": state.get("text", definition["initial_text"]),
}
)
self.draw_order += 1
+169
View File
@@ -0,0 +1,169 @@
from __future__ import annotations
from copy import deepcopy
from .binary import FormatError
PLACE_MOVE = 0x01
PLACE_HAS_CHARACTER = 0x02
PLACE_HAS_MATRIX = 0x04
PLACE_HAS_COLOR_TRANSFORM = 0x08
PLACE_HAS_RATIO = 0x10
PLACE_HAS_NAME = 0x20
PLACE_HAS_CLIP_DEPTH = 0x40
PLACE_HAS_CLIP_ACTIONS = 0x80
def frame_labels(character: dict) -> dict[str, int]:
labels = {}
for frame in character.get("frames", []):
for control in frame["controls"]:
if control["type_name"] == "frame_label":
label = control.get("label", "")
if not label:
continue
if label in labels:
raise FormatError(
f"character {character['id']}: duplicate frame label {label!r}"
)
labels[label] = frame["index"]
return labels
def first_stop_frame(character: dict, start_frame: int) -> int:
for frame in character.get("frames", [])[start_frame:]:
if any(
control["type_name"] == "do_action"
and control.get("first_action_opcode") == 0x07
for control in frame["controls"]
):
return frame["index"]
raise FormatError(
f"character {character['id']}: no stop action at/after frame {start_frame}"
)
def playback_frames(character: dict, label: str) -> list[int]:
labels = frame_labels(character)
if label not in labels:
raise FormatError(
f"character {character['id']}: frame label {label!r} is absent"
)
frames = character.get("frames", [])
if not frames:
raise FormatError(f"character {character['id']}: timeline has no frames")
start = labels[label]
result = []
frame_index = start
for _ in range(len(frames)):
result.append(frame_index)
frame = frames[frame_index]
if any(
control["type_name"] == "do_action"
and control.get("first_action_opcode") == 0x07
for control in frame["controls"]
):
return result
frame_index = (frame_index + 1) % len(frames)
raise FormatError(
f"character {character['id']}: playback from {label!r} never stops"
)
def playback_frame(character: dict, label: str, *, play: bool) -> int:
labels = frame_labels(character)
if label not in labels:
raise FormatError(
f"character {character['id']}: frame label {label!r} is absent"
)
start = labels[label]
return playback_frames(character, label)[-1] if play else start
def resolve_display_list(character: dict, through_frame: int) -> dict:
frames = character.get("frames", [])
if through_frame < 0 or through_frame >= len(frames):
raise FormatError(
f"character {character['id']}: frame {through_frame} is out of range"
)
objects: dict[int, dict] = {}
unresolved = []
for frame in frames[: through_frame + 1]:
frame_index = frame["index"]
for control in frame["controls"]:
kind = control["type_name"]
if kind in ("remove_object2", "remove_object3"):
objects.pop(control["depth"], None)
continue
if kind not in ("place_object2", "place_object3"):
continue
flags = control["flags"]
depth = control["depth"]
moving = bool(flags & PLACE_MOVE)
if moving:
if depth not in objects:
unresolved.append(
{
"frame": frame_index,
"control_offset": control["offset"],
"reason": "move references an empty depth",
"depth": depth,
}
)
current = {"depth": depth, "properties": {}, "provenance": {}}
objects[depth] = current
else:
current = objects[depth]
else:
current = {"depth": depth, "properties": {}, "provenance": {}}
objects[depth] = current
source = {
"frame": frame_index,
"control_offset": control["offset"],
}
def assign(name: str, value: object) -> None:
current["properties"][name] = deepcopy(value)
current["provenance"][name] = source
if flags & PLACE_HAS_CHARACTER:
assign("character_id", control["character_id"])
if flags & PLACE_HAS_MATRIX:
assign("matrix", control["matrix"])
if flags & PLACE_HAS_COLOR_TRANSFORM:
assign("color_transform_argb", control["color_transform"])
if flags & PLACE_HAS_RATIO:
assign("ratio", control["ratio"])
if flags & PLACE_HAS_NAME:
assign("name", control["name"])
if flags & PLACE_HAS_CLIP_DEPTH:
assign("clip_depth", control["clip_depth"])
if control["type"] == 9:
assign("blend_mode", control.get("blend_mode", -1))
assign("filter_pointer", control.get("filter_pointer", 0))
for depth, item in objects.items():
missing = [
name
for name in ("character_id", "matrix")
if name not in item["properties"]
]
if missing:
unresolved.append(
{
"depth": depth,
"reason": f"display object lacks {', '.join(missing)}",
}
)
return {
"character_id": character["id"],
"through_frame": through_frame,
"objects": [objects[depth] for depth in sorted(objects)],
"unresolved": unresolved,
}
+286
View File
@@ -0,0 +1,286 @@
from __future__ import annotations
import struct
import zlib
from dataclasses import asdict, dataclass
from pathlib import Path
from .binary import FormatError, align
XENOS_FORMATS = {
6: "A8R8G8B8",
18: "DXT1",
19: "DXT3",
20: "DXT5",
}
@dataclass(frozen=True)
class FetchConstant:
width: int
height: int
format_code: int
format_name: str
mipmaps: int
tiled: bool
endian: int
pitch: int = 0
def manifest(self) -> dict:
return asdict(self)
def reverse_bits(value: int, size: int) -> int:
result = 0
for bit in range(size):
result |= ((value >> bit) & 1) << (size - bit - 1)
return result
def parse_fetch_constant(data: bytes) -> FetchConstant:
"""Decode the 24-byte Xbox 360 texture fetch constant used by RX2."""
if len(data) != 24:
raise FormatError(f"Xbox texture fetch constant is {len(data)} bytes, expected 24")
encoded = struct.unpack(">6I", data)
words = [
reverse_bits(encoded[0], 32),
reverse_bits(encoded[1], 32),
encoded[2],
reverse_bits(encoded[3], 32),
reverse_bits(encoded[4], 32),
reverse_bits(encoded[5], 32),
]
width = (words[2] & 0x1FFF) + 1
height = ((words[2] >> 13) & 0x1FFF) + 1
format_code = reverse_bits((words[1] >> 26) & 0x3F, 6)
max_mip = reverse_bits((words[4] >> 22) & 0xF, 4)
return FetchConstant(
width=width,
height=height,
format_code=format_code,
format_name=XENOS_FORMATS.get(format_code, f"XENOS_{format_code}"),
mipmaps=max_mip + 1,
tiled=bool(reverse_bits(words[0] & 1, 1)),
endian=reverse_bits((words[1] >> 24) & 0x3, 2),
pitch=((encoded[0] >> 22) & 0x1FF) << 5,
)
def xbox360_tiled_offset(x: int, y: int, width: int, log_bpb: int) -> int:
aligned_width = align(width, 32)
macro = ((x >> 5) + (y >> 5) * (aligned_width >> 5)) << (log_bpb + 7)
micro = ((x & 7) + ((y & 0xE) << 2)) << log_bpb
offset = macro + ((micro & ~0xF) << 1) + (micro & 0xF) + ((y & 1) << 4)
address = (
((offset & ~0x1FF) << 3)
+ ((y & 16) << 7)
+ ((offset & 0x1C0) << 2)
+ (((((y & 8) >> 2) + (x >> 3)) & 3) << 6)
+ (offset & 0x3F)
)
return address >> log_bpb
def _layout(fetch: FetchConstant) -> tuple[int, int]:
if fetch.format_code == 18:
return 4, 8
if fetch.format_code in (19, 20):
return 4, 16
if fetch.format_code == 6:
return 1, 4
raise FormatError(f"unsupported Xbox texture format {fetch.format_name}")
def _endian_swap(block: bytes, compressed: bool) -> bytes:
if compressed:
swapped = bytearray(block)
for index in range(0, len(swapped) - 1, 2):
swapped[index], swapped[index + 1] = swapped[index + 1], swapped[index]
return bytes(swapped)
return block[::-1]
def linearize_base_level(data: bytes, fetch: FetchConstant) -> bytes:
"""Untile and endian-correct the base mip of an Xbox 360 texture."""
block_size, bytes_per_block = _layout(fetch)
block_width = max(1, (fetch.width + block_size - 1) // block_size)
block_height = max(1, (fetch.height + block_size - 1) // block_size)
linear = bytearray(block_width * block_height * bytes_per_block)
compressed = block_size == 4
if not fetch.tiled:
pitch_texels = fetch.pitch or fetch.width
pitch_blocks = max(
block_width, (pitch_texels + block_size - 1) // block_size
)
row_pitch = pitch_blocks * bytes_per_block
needed = (block_height - 1) * row_pitch + block_width * bytes_per_block
if len(data) < needed:
raise FormatError(f"texture payload is {len(data)} bytes, expected at least {needed}")
for y in range(block_height):
source_row = y * row_pitch
target_row = y * block_width * bytes_per_block
for x in range(block_width):
source = source_row + x * bytes_per_block
target = target_row + x * bytes_per_block
linear[target : target + bytes_per_block] = _endian_swap(
data[source : source + bytes_per_block], compressed
)
return bytes(linear)
tiled_width = align(fetch.width, 128 if compressed else 32) // block_size
log_bpb = bytes_per_block.bit_length() - 1
for y in range(block_height):
for x in range(block_width):
tiled_block = xbox360_tiled_offset(x, y, tiled_width, log_bpb)
source = tiled_block * bytes_per_block
target = (y * block_width + x) * bytes_per_block
if source + bytes_per_block > len(data):
raise FormatError(
f"tiled texture address {source:#x} exceeds {len(data):#x}-byte payload"
)
linear[target : target + bytes_per_block] = _endian_swap(
data[source : source + bytes_per_block], compressed
)
return bytes(linear)
def _expand_565(value: int) -> tuple[int, int, int]:
red = (value >> 11) & 0x1F
green = (value >> 5) & 0x3F
blue = value & 0x1F
return (
(red << 3) | (red >> 2),
(green << 2) | (green >> 4),
(blue << 3) | (blue >> 2),
)
def _color_table(c0: int, c1: int, allow_transparency: bool) -> list[tuple[int, ...]]:
color0 = _expand_565(c0)
color1 = _expand_565(c1)
if not allow_transparency or c0 > c1:
return [
(*color0, 255),
(*color1, 255),
tuple((2 * color0[i] + color1[i]) // 3 for i in range(3)) + (255,),
tuple((color0[i] + 2 * color1[i]) // 3 for i in range(3)) + (255,),
]
return [
(*color0, 255),
(*color1, 255),
tuple((color0[i] + color1[i]) // 2 for i in range(3)) + (255,),
(0, 0, 0, 0),
]
def _write_block(
output: bytearray,
width: int,
height: int,
block_x: int,
block_y: int,
colors: list[tuple[int, ...]],
color_bits: int,
alphas: list[int] | None = None,
) -> None:
for py in range(4):
for px in range(4):
x = block_x * 4 + px
y = block_y * 4 + py
if x >= width or y >= height:
continue
pixel = py * 4 + px
color = colors[(color_bits >> (pixel * 2)) & 3]
alpha = alphas[pixel] if alphas is not None else color[3]
offset = (y * width + x) * 4
output[offset : offset + 4] = bytes((color[0], color[1], color[2], alpha))
def _decode_dxt(width: int, height: int, data: bytes, format_code: int) -> bytes:
output = bytearray(width * height * 4)
block_bytes = 8 if format_code == 18 else 16
blocks_wide = max(1, (width + 3) // 4)
blocks_high = max(1, (height + 3) // 4)
source = 0
for by in range(blocks_high):
for bx in range(blocks_wide):
if source + block_bytes > len(data):
raise FormatError("compressed texture base mip is truncated")
alphas = None
color_offset = source
if format_code == 19:
alpha_bytes = data[source : source + 8]
alphas = []
for value in alpha_bytes:
alphas.extend(((value & 0xF) * 17, (value >> 4) * 17))
color_offset += 8
elif format_code == 20:
alpha0, alpha1 = data[source], data[source + 1]
palette = [alpha0, alpha1]
if alpha0 > alpha1:
palette.extend(
((7 - index) * alpha0 + index * alpha1) // 7
for index in range(1, 7)
)
else:
palette.extend(
((5 - index) * alpha0 + index * alpha1) // 5
for index in range(1, 5)
)
palette.extend((0, 255))
alpha_bits = int.from_bytes(data[source + 2 : source + 8], "little")
alphas = [palette[(alpha_bits >> (pixel * 3)) & 7] for pixel in range(16)]
color_offset += 8
c0, c1, color_bits = struct.unpack_from("<HHI", data, color_offset)
colors = _color_table(c0, c1, format_code == 18)
_write_block(output, width, height, bx, by, colors, color_bits, alphas)
source += block_bytes
return bytes(output)
def decode_rgba(data: bytes, fetch: FetchConstant) -> bytes:
linear = linearize_base_level(data, fetch)
if fetch.format_code in (18, 19, 20):
return _decode_dxt(fetch.width, fetch.height, linear, fetch.format_code)
if fetch.format_code == 6:
output = bytearray(fetch.width * fetch.height * 4)
for offset in range(0, len(output), 4):
blue, green, red, alpha = linear[offset : offset + 4]
output[offset : offset + 4] = bytes((red, green, blue, alpha))
return bytes(output)
raise FormatError(f"unsupported Xbox texture format {fetch.format_name}")
def _png_chunk(chunk_type: bytes, payload: bytes) -> bytes:
checksum = zlib.crc32(chunk_type)
checksum = zlib.crc32(payload, checksum)
return (
struct.pack(">I", len(payload))
+ chunk_type
+ payload
+ struct.pack(">I", checksum & 0xFFFFFFFF)
)
def encode_png(width: int, height: int, rgba: bytes) -> bytes:
expected = width * height * 4
if len(rgba) != expected:
raise FormatError(f"RGBA payload is {len(rgba)} bytes, expected {expected}")
scanlines = bytearray()
stride = width * 4
for y in range(height):
scanlines.append(0)
scanlines.extend(rgba[y * stride : (y + 1) * stride])
header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
return (
b"\x89PNG\r\n\x1a\n"
+ _png_chunk(b"IHDR", header)
+ _png_chunk(b"IDAT", zlib.compress(bytes(scanlines), 9))
+ _png_chunk(b"IEND", b"")
)
def write_png(path: Path, width: int, height: int, rgba: bytes) -> None:
Path(path).write_bytes(encode_png(width, height, rgba))