XC3 data_sheet.bin readers

xc3_data_sheet.py

#  MIT License
#  
#  Copyright (c) 2023 RoccoDev
#  
#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this 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.

import struct
import json
import sys
import numpy

def u16(data): return struct.unpack('<H', data[0:2])[0]

def u32(data): return struct.unpack('<I', data[0:4])[0]

def u64(data): return struct.unpack('<Q', data[0:8])[0]

def read_string(data):
    len = data.index(0)
    return data[:len].decode('utf-8'), len + 1

def read_value(data, offset, keys):
    type = data[offset]
    type_high = type & 0xf
    type_low = type >> 4

    offset += 1

    if type_high == 1:
        # Integers
        num, size = read_varint(type_low, type_high, data[offset:], True)
        return num, size + offset

    if type_high == 2:
        # Float
        return struct.unpack('f', data[offset:offset+4])[0], offset + 4

    if type_high == 3:
        # Double
        return struct.unpack('d', data[offset:offset+8])[0], offset + 8

    if type_high == 4:
        # Nul-terminated string
        s, size = read_string(data[offset:])
        return s, offset + size

    if type_high == 5 or type_high == 13:
        # Repeated
        repeat_count, vi_size = read_varint(type_low, type_high, data[offset:], False)
        values = []
        start_offset = offset
        offset += vi_size
        for i in range(repeat_count):
            size, offset = read_value(data, offset, keys)
            val, offset = read_value(data, offset, keys)
            values.append(val)
        return values, offset

    if type_high == 6 or type_high == 14:
        # Struct
        start = offset
        fields = {}
        field_count, vi_size = read_varint(type_low, type_high, data[offset:], False)
        offset += vi_size
        for i in range(field_count):
            key, offset = read_value(data, offset, keys)
            size, offset = read_value(data, offset, keys)
            val, offset = read_value(data, offset, keys)
            fields[keys[key]] = val
        return fields, offset

    if type_high == 7:
        # Boolean (from lower bits)
        return type_low != 0, offset

    if type_high == 8:
        # Positive integer from lower bits
        return type_low, offset

    if type_high == 9:
        # Negative integer from lower bits
        return -type_low, offset

    if type_high == 10:
        # Double from lower bits
        return float(type_low), offset

    if type_high == 11:
        # Negative double from lower bits
        return -float(type_low), offset

    if type_high == 12:
        # String with known size
        return data[offset:offset+type_low].decode('utf-8'), offset + type_low

    raise Exception(f"Unknown type {type_high}")

def read_varint(type_low, type_high, data, with_negs):
    if type_high == 13 or type_high == 14:
        return (type_low, 0)

    # Positive size
    if type_low == 1:
        return (u32(data), 4)
    if type_low == 2:
        return (u16(data), 2)
    if type_low == 3:
        return (data[0], 1)

    if with_negs:
        # Negative size
        if type_low == 4:
            return (-data[0], 1)
        if type_low == 5:
            return (-u16(data), 2)
        if type_low == 6:
            return (-u32(data), 4)

    # Default
    return (u64(data), 8)

def parse_data_sheet(data):
    data = bytes(data)
    assert list(data[0:4]) == [0x56, 0x34, 0x12, 0x00]

    keys = []

    assert data[8] == 0xf
    key_count, offset = read_value(data, 9, keys)

    # Read keys
    for i in range(key_count):
        s, ln = read_string(data[offset:])
        offset += ln
        keys.append(s)

    values = []
    total_len = len(data)

    return read_value(data, offset, keys)

if __name__ == "__main__":
    file = sys.argv[1]
    file = open(file, "rb")
    data = list(file.read())
    file.close()
    data_sheet = parse_data_sheet(data)
    json.dump(data_sheet, sys.stdout, ensure_ascii=False, indent=1)

xc3_data_sheet.rs

#!/usr/bin/env rust-script

/// ```cargo
/// [dependencies]
/// serde = { version = "1.0", features = ["derive"] }
/// serde_json = "1.0"
/// serde-tuple-vec-map = "1.0.1"
/// ```

// MIT License
//
// Copyright (c) 2023 RoccoDev
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this 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.


use std::{convert::TryInto, io::BufWriter};
use serde::Serialize;
use serde_json::ser::{Serializer, PrettyFormatter};

struct Header<'a> {
    field_keys: Vec<&'a str>,
}

#[derive(Clone, Copy)]
struct Reader<'a> {
    type_high: u8,
    type_low: u8,
    next_data: &'a [u8],
}

struct ResultData<'a, T> {
    result: T,
    next_data: &'a [u8],
}

enum AnyReader<'a, 'h> {
    Integer(VarintReader<'a>),
    Bool(BoolReader<'a>),
    Floating(FloatingReader<'a>),
    String(StringReader<'a>),
    List(ListReader<'a, 'h>),
    Struct(StructReader<'a, 'h>),
}

#[derive(Debug, Serialize)]
#[serde(untagged)]
enum Value<'a> {
    Integer(i64),
    Bool(bool),
    Floating(f64),
    String(&'a str),
    List(Vec<Value<'a>>),
    #[serde(with = "tuple_vec_map")]
    Struct(Vec<(&'a str, Value<'a>)>),
}

struct StringReader<'a> {
    reader: Reader<'a>,
}

struct VarintReader<'a> {
    reader: Reader<'a>,
}

struct BoolReader<'a> {
    reader: Reader<'a>,
}

struct FloatingReader<'a> {
    reader: Reader<'a>,
}

struct ListReader<'a, 'h> {
    reader: Reader<'a>,
    elem_count: usize,
    header: &'h Header<'a>,
}

struct StructReader<'a, 'h> {
    reader: Reader<'a>,
    field_count: usize,
    header: &'h Header<'a>,
}

impl<'a> Header<'a> {
    fn read(data: &'a [u8]) -> (Self, &[u8]) {
        assert_eq!([0x56, 0x34, 0x12, 0x00], data[0..4]);
        assert_eq!(0xf, data[8]);

        let mut header = Header { field_keys: vec![] };

        let ResultData {
            result,
            mut next_data,
        } = VarintReader::new(Reader::new(&data[9..])).read();
        let key_count = result as usize;

        header.field_keys.reserve_exact(key_count);

        for _ in 0..key_count {
            let name = StringReader::global_read_nul_terminated(next_data);
            header.field_keys.push(name);
            next_data = &next_data[name.len() + 1..];
        }

        (header, next_data)
    }
}

impl<'a> Reader<'a> {
    fn new(data: &'a [u8]) -> Self {
        if data.is_empty() {
            return Self {
                type_high: 0,
                type_low: 0,
                next_data: data,
            };
        }

        let ty = data[0];
        Self {
            type_high: ty & 0xf,
            type_low: ty >> 4,
            next_data: &data[1..],
        }
    }
}

impl<'a, 'h> AnyReader<'a, 'h> {
    fn new(data: &'a [u8], header: &'h Header<'a>) -> Self {
        let reader = Reader::new(data);
        (match reader.type_high {
            1 | 8 | 9 => |r, _| Self::Integer(VarintReader::new(r)),
            7 => |r, _| Self::Bool(BoolReader::new(r)),
            2 | 3 | 10 | 11 => |r, _| Self::Floating(FloatingReader::new(r)),
            4 | 12 => |r, _| Self::String(StringReader::new(r)),
            5 | 13 => |r, h| Self::List(ListReader::new(r, h)),
            6 | 14 => |r, h| Self::Struct(StructReader::new(r, h)),
            t => panic!("unknown type {t}"),
        })(reader, header)
    }

    fn read(&self) -> Value<'a> {
        match self {
            AnyReader::Integer(r) => Value::Integer(r.read().result),
            AnyReader::Bool(r) => Value::Bool(r.read().result),
            AnyReader::Floating(r) => Value::Floating(r.read().result),
            AnyReader::String(r) => Value::String(r.read().result),
            AnyReader::List(r) => Value::List(
                r.read()
                    .result
                    .into_iter()
                    .map(|r| r.read())
                    .collect::<Vec<_>>(),
            ),
            AnyReader::Struct(r) => Value::Struct(
                r.read()
                    .result
                    .into_iter()
                    .map(|(k, v)| (k, v.read()))
                    .collect::<Vec<_>>(),
            ),
        }
    }
}

impl<'a> StringReader<'a> {
    fn new(reader: Reader<'a>) -> Self {
        Self { reader }
    }

    fn read(&self) -> ResultData<&'a str> {
        match self.reader.type_high {
            4 => self.read_nul_terminated(),
            12 => self.read_embedded(),
            _ => unreachable!(),
        }
    }

    fn read_nul_terminated(&self) -> ResultData<&'a str> {
        let nul_pos = self
            .reader
            .next_data
            .iter()
            .position(|i| *i == 0)
            .expect("no nul byte");
        let result = std::str::from_utf8(&self.reader.next_data[..nul_pos]).expect("invalid utf8");
        ResultData {
            result,
            next_data: &self.reader.next_data[nul_pos + 1..],
        }
    }

    fn global_read_nul_terminated(data: &[u8]) -> &str {
        let nul_pos = data.iter().position(|i| *i == 0).expect("no nul byte");
        std::str::from_utf8(&data[..nul_pos]).expect("invalid utf8")
    }

    fn read_embedded(&self) -> ResultData<&'a str> {
        let result = std::str::from_utf8(&self.reader.next_data[..self.reader.type_low as usize])
            .expect("invalid utf8");
        ResultData {
            result,
            next_data: &self.reader.next_data[self.reader.type_low as usize..],
        }
    }
}

impl<'a> VarintReader<'a> {
    fn new(reader: Reader<'a>) -> Self {
        Self { reader }
    }

    fn read(&self) -> ResultData<'a, i64> {
        let data = &self.reader.next_data;
        let (num, size): (i64, usize) = match self.reader.type_low {
            _ if self.reader.type_high == 8
                || self.reader.type_high == 13
                || self.reader.type_high == 14 =>
            {
                (i64::from(self.reader.type_low), 0)
            }
            _ if self.reader.type_high == 9 => (-i64::from(self.reader.type_low), 0),
            1 => (u32::from_le_bytes(data[0..4].try_into().unwrap()).into(), 4),
            2 => (u16::from_le_bytes(data[0..2].try_into().unwrap()).into(), 2),
            3 => (i64::from(data[0]), 1),
            4 => (
                -i64::from(u32::from_le_bytes(data[0..4].try_into().unwrap())),
                4,
            ),
            5 => (
                -i64::from(u16::from_le_bytes(data[0..2].try_into().unwrap())),
                2,
            ),
            6 => (-i64::from(data[0]), 1),
            _ => (i64::from_le_bytes(data[0..8].try_into().unwrap()), 8),
        };
        ResultData {
            result: num,
            next_data: &data[size..],
        }
    }
}

impl<'a> BoolReader<'a> {
    fn new(reader: Reader<'a>) -> Self {
        Self { reader }
    }

    fn read(&self) -> ResultData<bool> {
        ResultData {
            result: self.reader.type_low != 0,
            next_data: self.reader.next_data,
        }
    }
}

impl<'a> FloatingReader<'a> {
    fn new(reader: Reader<'a>) -> Self {
        Self { reader }
    }

    fn read(&self) -> ResultData<f64> {
        let data = &self.reader.next_data;
        let (num, size) = match self.reader.type_high {
            2 => (
                f32::from_bits(u32::from_le_bytes(data[0..4].try_into().unwrap())).into(),
                4,
            ),
            3 => (
                f64::from_bits(u64::from_le_bytes(data[0..8].try_into().unwrap())),
                8,
            ),
            10 => (f64::from(self.reader.type_low), 0),
            11 => (-f64::from(self.reader.type_low), 0),
            _ => unreachable!(),
        };
        ResultData {
            result: num,
            next_data: &data[size..],
        }
    }
}

impl<'a, 'h> ListReader<'a, 'h> {
    fn new(r: Reader<'a>, header: &'h Header<'a>) -> Self {
        let ResultData {
            result: elem_count,
            next_data,
        } = VarintReader::new(r).read();
        Self {
            elem_count: elem_count as usize,
            reader: Reader::new(next_data),
            header,
        }
    }

    fn read(&self) -> ResultData<Vec<AnyReader<'a, 'h>>> {
        let mut entries = Vec::with_capacity(self.elem_count);
        let mut reader = self.reader;

        for _ in 0..self.elem_count {
            let ResultData {
                result: size,
                next_data,
            } = VarintReader::new(reader).read();
            entries.push(AnyReader::new(next_data, self.header));
            reader = Reader::new(&next_data[size as usize..]);
        }

        ResultData {
            result: entries,
            next_data: reader.next_data,
        }
    }
}

impl<'a, 'h> StructReader<'a, 'h> {
    fn new(r: Reader<'a>, header: &'h Header<'a>) -> Self {
        let ResultData {
            result: field_count,
            next_data,
        } = VarintReader::new(r).read();
        Self {
            field_count: field_count as usize,
            reader: Reader::new(next_data),
            header,
        }
    }

    fn read(&self) -> ResultData<Vec<(&'a str, AnyReader<'a, 'h>)>> {
        let mut fields = Vec::with_capacity(self.field_count);
        let mut reader = self.reader;

        for _ in 0..self.field_count {
            let ResultData {
                result: key_idx,
                next_data,
            } = VarintReader::new(reader).read();
            reader = Reader::new(next_data);
            let ResultData {
                result: value_size,
                next_data,
            } = VarintReader::new(reader).read();

            fields.push((
                self.header.field_keys[key_idx as usize],
                AnyReader::new(next_data, self.header),
            ));
            reader = Reader::new(&next_data[value_size as usize..]);
        }

        ResultData {
            result: fields,
            next_data: reader.next_data,
        }
    }
}

fn main() {
    let file = std::env::args().nth(1).expect("no file arg");
    let file = std::fs::read(&file).unwrap();

    let (header, bytes) = Header::read(&file);
    let value = AnyReader::new(bytes, &header).read();

    let writer = BufWriter::new(std::io::stdout());
    let mut json = Serializer::with_formatter(writer, PrettyFormatter::with_indent(b" "));
    value.serialize(&mut json).unwrap();
}