172 lines
7.5 KiB
Python
172 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Inspect SERiF archive data without compiling SERiF or installing dependencies.
|
|
|
|
This is a data reader, not a physics/model factory. The C++ loader additionally
|
|
checks SERiF version tolerance, component schemas and discretization compatibility.
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import shlex
|
|
import struct
|
|
import sys
|
|
|
|
|
|
def fingerprint(data):
|
|
"""Match the archive's versioned corruption-detection fingerprint."""
|
|
value = 14695981039346656037
|
|
for byte in data:
|
|
value = ((value ^ byte) * 1099511628211) & ((1 << 64) - 1)
|
|
return f"fnv1a64-v1:{value:016x}"
|
|
|
|
|
|
def unquote(data):
|
|
"""Decode std::quoted text, whose escapes differ from JSON escapes."""
|
|
if len(data) < 2 or data[0] != 34 or data[-1] != 34:
|
|
raise ValueError("Invalid quoted string")
|
|
result = bytearray()
|
|
index = 1
|
|
while index < len(data) - 1:
|
|
if data[index] == 92:
|
|
index += 1
|
|
if index >= len(data) - 1:
|
|
raise ValueError("Unfinished string escape")
|
|
elif data[index] == 34:
|
|
raise ValueError("Unescaped quote inside string")
|
|
result.append(data[index])
|
|
index += 1
|
|
return bytes(result)
|
|
|
|
|
|
class Archive:
|
|
"""Read the ledger once and seek directly to individual data sections."""
|
|
|
|
def __init__(self, path, maximum_bytes=1024 * 1024 * 1024):
|
|
self.file = open(path, "rb")
|
|
self.maximum_bytes = maximum_bytes
|
|
try:
|
|
self.file.seek(0, 2)
|
|
size = self.file.tell()
|
|
self.file.seek(0)
|
|
header = self.file.read(512).decode("ascii")
|
|
lines = header.split("END_HEADER\n", 1)[0].splitlines()
|
|
if not lines or lines[0] != "SERIF_ARCHIVE 1":
|
|
raise ValueError("Unsupported SERiF archive format")
|
|
self.header = dict(shlex.split(line) for line in lines[1:])
|
|
offset = int(self.header["ledger_offset"])
|
|
count = int(self.header["ledger_bytes"])
|
|
if (offset != 512 or count < 0 or count > min(maximum_bytes, 64 * 1024 * 1024)
|
|
or offset + count > size or int(self.header["file_bytes"]) != size
|
|
or int(self.header["data_offset"]) != offset + count):
|
|
raise ValueError("Invalid archive section bounds")
|
|
ledger = self.file.read(count)
|
|
if fingerprint(ledger) != self.header["ledger_checksum"]:
|
|
raise ValueError("Damaged archive ledger")
|
|
self.entries = {}
|
|
previous_end = offset + count
|
|
previous_path = ""
|
|
for line in ledger.decode("utf-8").splitlines()[1:]:
|
|
path, schema, version, kind, extent, start, length, checksum = shlex.split(line)
|
|
entry = dict(schema=schema, version=int(version), kind=int(kind), count=int(extent),
|
|
offset=int(start), bytes=int(length), checksum=checksum)
|
|
if path <= previous_path or not 0 <= entry["kind"] <= 8 or entry["count"] < 0:
|
|
raise ValueError("Invalid or duplicate ledger entry")
|
|
if entry["kind"]:
|
|
if (entry["offset"] < previous_end or entry["bytes"] < 0
|
|
or entry["bytes"] > maximum_bytes or entry["offset"] + entry["bytes"] > size):
|
|
raise ValueError("Invalid or overlapping payload bounds")
|
|
previous_end = entry["offset"] + entry["bytes"]
|
|
elif entry["offset"] or entry["bytes"] or entry["count"]:
|
|
raise ValueError("Directory contains a payload")
|
|
self.entries[path] = entry
|
|
previous_path = path
|
|
if len(self.entries) != int(self.header["entries"]):
|
|
raise ValueError("Incorrect ledger extent")
|
|
if self.header["format"] not in ("binary", "text"):
|
|
raise ValueError("Unknown numeric encoding")
|
|
except Exception:
|
|
self.file.close()
|
|
raise
|
|
|
|
def __enter__(self):
|
|
"""Allow a with-statement to own the open file."""
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
"""Close the reader after analysis, including when decoding fails."""
|
|
self.file.close()
|
|
|
|
def read(self, path):
|
|
"""Return one string, number or numerical array without reading sibling fields."""
|
|
entry = self.entries[path]
|
|
kind, count = entry["kind"], entry["count"]
|
|
self.file.seek(entry["offset"])
|
|
data = self.file.read(entry["bytes"])
|
|
if len(data) != entry["bytes"] or fingerprint(data) != entry["checksum"]:
|
|
raise ValueError(f"Damaged payload: {path}")
|
|
binary = self.header["format"] == "binary"
|
|
if kind == 0:
|
|
return None
|
|
if kind == 1:
|
|
allowed = (b"\x00", b"\x01") if binary else (b"0", b"1")
|
|
if count != 1 or data not in allowed:
|
|
raise ValueError("Invalid boolean")
|
|
return data == allowed[1]
|
|
if kind == 5:
|
|
text = data if binary else unquote(data)
|
|
if len(text) != count:
|
|
raise ValueError("Invalid string length")
|
|
return text.decode("utf-8", errors="surrogateescape")
|
|
if kind in (2, 3, 4) and count != 1:
|
|
raise ValueError("Invalid scalar extent")
|
|
if count > len(data) or count > self.maximum_bytes // 8:
|
|
raise ValueError("Invalid numerical extent")
|
|
code = {2: "q", 3: "Q", 4: "d", 6: "d", 7: "q", 8: "Q"}[kind]
|
|
if binary:
|
|
if len(data) != 8 * count:
|
|
raise ValueError("Invalid binary array length")
|
|
values = list(struct.unpack(f"<{count}{code}", data))
|
|
else:
|
|
tokens = data.decode("ascii").split()
|
|
if len(tokens) != count:
|
|
raise ValueError("Invalid text array length")
|
|
values = []
|
|
for token in tokens:
|
|
if code == "d":
|
|
value = struct.unpack("<d", bytes.fromhex(token[5:])[::-1])[0] if token.startswith("bits:") else float(token)
|
|
else:
|
|
value = int(token)
|
|
if not (-(1 << 63) if code == "q" else 0) <= value < (1 << (63 if code == "q" else 64)):
|
|
raise ValueError("Integer out of range")
|
|
values.append(value)
|
|
return values[0] if kind in (2, 3, 4) else values
|
|
|
|
|
|
def main():
|
|
"""Print available data, or export one numerical field for ordinary astronomical analysis."""
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("archive")
|
|
parser.add_argument("--read", metavar="FIELD", help="Read one ledger path")
|
|
parser.add_argument("--csv", action="store_true", help="Emit index,value rows for the selected numerical array")
|
|
arguments = parser.parse_args()
|
|
with Archive(arguments.archive) as archive:
|
|
if arguments.read:
|
|
value = archive.read(arguments.read)
|
|
if arguments.csv:
|
|
if not isinstance(value, list):
|
|
parser.error("--csv requires an array field")
|
|
output = csv.writer(sys.stdout)
|
|
output.writerow(("index", "value"))
|
|
output.writerows(enumerate(value))
|
|
else:
|
|
print(json.dumps(value, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"SERiF {archive.header['serif_version']}; written {archive.header['written_utc']}")
|
|
for path, entry in archive.entries.items():
|
|
print(f"{path}\t{entry['schema']}\t{entry['count']}\t{entry['offset']}\t{entry['bytes']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|