"""Inspect a same-size V2C1 deployment candidate without importing TensorFlow. python same_size_verify.py --candidate candidate.tflite python same_size_verify.py --candidate candidate.tflite --baseline baseline.tflite The standard-library-only reader checks the serialized TFLite graph, not runtime accuracy, firmware compatibility, tensor-arena use, or inference latency. Passing this gate is not evidence that the detector improved. INT32 bias and reduction-axis tensors are allowed; every activation and weight must be quantized INT8. Allocated I/O shapes are fixed; the existing optional dynamic batch signature is reported and accepted only when the allocated batch is one. No dimension is resized. The byte ceiling is the smaller of the actual baseline size and the frozen 40,200 bytes. Neither model nor deployment metadata is changed. Exit status is 0 on pass, 1 on verification failure, and 2 for unreadable files or malformed FlatBuffers. """ from __future__ import annotations import argparse import hashlib import json import math import struct import sys from collections import Counter from pathlib import Path FROZEN_SIZE_CEILING = 40_200 DEFAULT_BASELINE = Path(__file__).resolve().parent / "v2_deploy" / "drone_model_int8.tflite" OP_NAMES = {3: "CONV_2D", 9: "FULLY_CONNECTED", 17: "MAX_POOL_2D", 25: "SOFTMAX", 40: "MEAN"} OPTIONS_TYPES = {3: 1, 9: 8, 17: 5, 25: 9, 40: 27} INT8, INT32 = 9, 2 class ModelFormatError(ValueError): """The file cannot be safely decoded by this narrow TFLite schema reader.""" class FlatBuffer: """Bounds-checked access to the TFLite schema fields needed by this gate. Field numbers and enum values match TensorFlow's schema_py_generated.py. This deliberately never sends candidate bytes to a native interpreter. """ def __init__(self, raw: bytes): self.raw = raw if len(raw) < 8 or raw[4:8] != b"TFL3": raise ModelFormatError("missing TFL3 file identifier") self.root = self.scalar(0, "I") if self.root < 8: raise ModelFormatError("invalid root table offset") self.table(self.root) def bounds(self, pos: int, size: int) -> None: if pos < 0 or size < 0 or pos > len(self.raw) - size: raise ModelFormatError(f"out-of-bounds FlatBuffer field at {pos} ({size} bytes)") def scalar(self, pos: int, fmt: str): self.bounds(pos, struct.calcsize("<" + fmt)) return struct.unpack_from("<" + fmt, self.raw, pos)[0] def table(self, pos: int) -> tuple[int, int, int]: vtable = pos - self.scalar(pos, "i") length, object_size = self.scalar(vtable, "H"), self.scalar(vtable + 2, "H") if length < 4 or length % 2 or object_size < 4: raise ModelFormatError("invalid table or vtable size") self.bounds(vtable, length) self.bounds(pos, object_size) return vtable, length, object_size def field(self, pos: int, index: int, width: int = 4) -> int | None: vtable, length, object_size = self.table(pos) entry = 4 + 2 * index offset = self.scalar(vtable + entry, "H") if entry + 2 <= length else 0 if not offset: return None if offset < 4 or offset + width > object_size: raise ModelFormatError("table field extends beyond its object") return pos + offset def value(self, pos: int, index: int, fmt: str, default=0): field = self.field(pos, index, struct.calcsize("<" + fmt)) return default if field is None else self.scalar(field, fmt) def pointer(self, field: int) -> int: offset = self.scalar(field, "I") if offset < 4: raise ModelFormatError("invalid relative pointer") target = field + offset self.bounds(target, 4) return target def child(self, pos: int, index: int) -> int | None: field = self.field(pos, index) if field is None: return None target = self.pointer(field) self.table(target) return target def vector_location(self, pos: int, index: int, width: int) -> tuple[int, int]: field = self.field(pos, index) if field is None: return 0, 0 vector = self.pointer(field) count = self.scalar(vector, "I") self.bounds(vector + 4, count * width) return vector + 4, count def vector(self, pos: int, index: int, fmt: str) -> list: width = struct.calcsize("<" + fmt) start, count = self.vector_location(pos, index, width) return [self.scalar(start + i * width, fmt) for i in range(count)] def tables(self, pos: int, index: int) -> list[int]: start, count = self.vector_location(pos, index, 4) result = [self.pointer(start + i * 4) for i in range(count)] for table in result: self.table(table) return result def string(self, pos: int, index: int) -> str: start, count = self.vector_location(pos, index, 1) if not start: return "" self.bounds(start + count, 1) if self.raw[start + count] != 0: raise ModelFormatError("unterminated FlatBuffer string") try: return self.raw[start:start + count].decode("utf-8") except UnicodeDecodeError as exc: raise ModelFormatError("non-UTF-8 FlatBuffer string") from exc def inspect_model(raw: bytes, path: str = "") -> dict: """Return a machine-readable graph inspection and all failed gate checks.""" fb = FlatBuffer(raw) failures: list[str] = [] def check(ok: bool, message: str) -> None: if not ok: failures.append(message) model = fb.root check(fb.value(model, 0, "I") == 3, "TFLite schema version must be 3") codes = [] for table in fb.tables(model, 1): old_code = fb.value(table, 0, "b") new_code = fb.value(table, 3, "i") # TFLite reads the maximum for backwards-compatible builtin enums. code = max(old_code, new_code) check(code in OP_NAMES, f"unsupported builtin operator code {code}") check(not fb.string(table, 1), "custom/Select TF operator code is forbidden") check(old_code == new_code or new_code == 0, f"inconsistent builtin operator codes ({old_code}, {new_code})") version = fb.value(table, 2, "i", 1) check(version > 0, "operator version must be positive") codes.append((code, version)) buffers = [] for table in fb.tables(model, 4): start, count = fb.vector_location(table, 0, 1) check(fb.value(table, 1, "Q") == 0 and fb.value(table, 2, "Q") == 0, "external buffer storage is forbidden") buffers.append(raw[start:start + count] if start else b"") check(bool(buffers), "model must have a buffer table") if buffers: check(not buffers[0], "buffer zero must be empty") graphs = fb.tables(model, 2) check(len(graphs) == 1, "model must have exactly one subgraph") report = {"path": path, "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest(), "schema_version": fb.value(model, 0, "I"), "subgraphs": len(graphs), "operator_counts": {}, "operator_versions": {}, "inputs": [], "outputs": [], "tensor_dtypes": {}, "failures": failures} if len(graphs) != 1: return report graph = graphs[0] tensors = [] for ti, table in enumerate(fb.tables(graph, 0)): shape = fb.vector(table, 0, "i") signature = fb.vector(table, 7, "i") dtype = fb.value(table, 1, "b") buffer_index = fb.value(table, 2, "I") check(dtype in (INT8, INT32), f"tensor {ti} has non-INT8/INT32 dtype {dtype}") check(bool(shape) and all(d > 0 for d in shape), f"tensor {ti} shape is not fixed and positive") signature_matches = signature == shape or ( len(signature) == len(shape) and bool(shape) and shape[0] == 1 and signature[0] == -1 and signature[1:] == shape[1:]) check(not signature or signature_matches, f"tensor {ti} has a dynamic/different non-batch shape signature") check(not fb.value(table, 5, "?"), f"tensor {ti} is variable") check(fb.child(table, 6) is None, f"tensor {ti} has sparse storage") check(not fb.tables(table, 9), f"tensor {ti} has variant subtypes") check(buffer_index < len(buffers), f"tensor {ti} references nonexistent buffer {buffer_index}") data = buffers[buffer_index] if buffer_index < len(buffers) else b"" if data and dtype in (INT8, INT32) and shape and all(d > 0 for d in shape): expected_size = math.prod(shape) * (1 if dtype == INT8 else 4) check(len(data) == expected_size, f"tensor {ti} constant data size does not match shape/dtype") quant = fb.child(table, 4) scales = fb.vector(quant, 2, "f") if quant else [] zero_points = fb.vector(quant, 3, "q") if quant else [] axis = fb.value(quant, 6, "i") if quant else 0 if quant: check(fb.value(quant, 4, "B") == 0 and fb.child(quant, 5) is None, f"tensor {ti} has custom quantization") if dtype == INT8 or scales or zero_points: check(bool(scales) and len(scales) == len(zero_points), f"tensor {ti} lacks matching quantization parameters") check(all(math.isfinite(s) and s > 0 for s in scales), f"tensor {ti} quantization scale is invalid") check(all(-128 <= zp <= 127 for zp in zero_points) if dtype == INT8 else all(zp == 0 for zp in zero_points), f"tensor {ti} quantization zero point is invalid") if len(scales) > 1: check(0 <= axis < len(shape) and len(scales) == shape[axis], f"tensor {ti} per-axis quantization does not match its shape") tensors.append({"index": ti, "name": fb.string(table, 3), "shape": shape, "shape_signature": signature, "dtype": {INT8: "int8", INT32: "int32"}.get(dtype, f"type_{dtype}"), "scales": scales, "zero_points": zero_points, "quantized_dimension": axis, "constant": bool(data), "_data": data}) inputs, outputs = fb.vector(graph, 1, "i"), fb.vector(graph, 2, "i") for label, indices, expected_shape in (("input", inputs, [1, 32, 32, 1]), ("output", outputs, [1, 2])): check(len(indices) == 1, f"model must have exactly one {label}") for index in indices: check(0 <= index < len(tensors), f"{label} references nonexistent tensor {index}") if not 0 <= index < len(tensors): continue tensor = tensors[index] check(tensor["shape"] == expected_shape, f"{label} shape must be {expected_shape}") check(tensor["dtype"] == "int8", f"{label} dtype must be int8") check(len(tensor["scales"]) == 1, f"{label} must use per-tensor quantization") check(not tensor["constant"], f"{label} must be a runtime tensor") report[label + "s"].append({k: tensor[k] for k in ("index", "name", "shape", "shape_signature", "dtype", "scales", "zero_points")}) integer_roles: set[int] = set() weight_roles: set[int] = set() produced = set() available = set(inputs) | {t["index"] for t in tensors if t["constant"]} operators = fb.tables(graph, 3) check(bool(operators), "model must have operators") counts: Counter = Counter() versions: dict[str, set[int]] = {} last_op = None for oi, table in enumerate(operators): opcode_index = fb.value(table, 0, "I") check(opcode_index < len(codes), f"operator {oi} references nonexistent opcode") if opcode_index >= len(codes): continue code, version = codes[opcode_index] name = OP_NAMES.get(code, f"builtin_{code}") counts[name] += 1 versions.setdefault(name, set()).add(version) last_op = code op_inputs, op_outputs = fb.vector(table, 1, "i"), fb.vector(table, 2, "i") check(len(op_outputs) == 1, f"operator {oi} must have one output") expected_inputs = 3 if code in (3, 9) else 2 if code == 40 else 1 check(len(op_inputs) == expected_inputs, f"operator {oi} {name} has wrong input count") check(not fb.vector(table, 5, "B") and not fb.value(table, 9, "Q") and not fb.value(table, 10, "Q"), f"operator {oi} has custom options") check(not any(fb.vector(table, 7, "?")), f"operator {oi} mutates a tensor") check(not fb.vector(table, 8, "i"), f"operator {oi} has unsupported intermediate tensors") check(fb.value(table, 11, "B") == 0 and fb.child(table, 12) is None, f"operator {oi} uses unsupported secondary builtin options") check(fb.value(table, 3, "B") == OPTIONS_TYPES.get(code), f"operator {oi} has wrong builtin options type") check(fb.child(table, 4) is not None, f"operator {oi} lacks builtin options") for ii, index in enumerate(op_inputs): if code in (3, 9) and ii == 2 and index == -1: continue # a missing convolution/FC bias is a valid TFLite input check(0 <= index < len(tensors), f"operator {oi} has nonexistent/optional input {index}") check(index in available, f"operator {oi} reads unavailable tensor {index}") for index in op_outputs: check(0 <= index < len(tensors), f"operator {oi} has nonexistent output {index}") check(index not in available and index not in produced, f"operator {oi} overwrites tensor {index}") if 0 <= index < len(tensors): check(tensors[index]["dtype"] == "int8", f"operator {oi} output must be int8") check(len(tensors[index]["scales"]) == 1, f"operator {oi} activation output must use per-tensor quantization") produced.add(index) available.update(op_outputs) if len(op_inputs) != expected_inputs or any( not 0 <= index < len(tensors) and not (code in (3, 9) and ii == 2 and index == -1) for ii, index in enumerate(op_inputs)): continue check(tensors[op_inputs[0]]["dtype"] == "int8", f"operator {oi} activation input must be int8") check(len(tensors[op_inputs[0]]["scales"]) == 1, f"operator {oi} activation input must use per-tensor quantization") if code in (3, 9): weight = tensors[op_inputs[1]] weight_roles.add(op_inputs[1]) check(weight["dtype"] == "int8" and weight["constant"], f"operator {oi} weights must be constant int8") check(all(zp == 0 for zp in weight["zero_points"]), f"operator {oi} weights must have symmetric quantization") if op_inputs[2] != -1: bias = tensors[op_inputs[2]] integer_roles.add(op_inputs[2]) check(bias["dtype"] == "int32" and bias["constant"], f"operator {oi} bias must be constant int32") check(bool(bias["scales"]) and all(zp == 0 for zp in bias["zero_points"]), f"operator {oi} bias must be quantized") input_scales = tensors[op_inputs[0]]["scales"] expected_scales = [input_scales[0] * s for s in weight["scales"]] if len(input_scales) == 1 else [] check(len(bias["scales"]) == len(expected_scales) and all( math.isclose(a, b, rel_tol=1e-5, abs_tol=0.0) for a, b in zip(bias["scales"], expected_scales)), f"operator {oi} bias scale must equal input scale times weight scale") elif code == 40: axes = tensors[op_inputs[1]] integer_roles.add(op_inputs[1]) check(axes["dtype"] == "int32" and axes["constant"] and not axes["scales"] and not axes["zero_points"], f"operator {oi} mean axes must be unquantized constant int32") if axes["dtype"] == "int32" and len(axes["_data"]) % 4 == 0: axes_values = [v[0] for v in struct.iter_unpack(" dict: """Read both files and enforce graph and actual-byte-count gates.""" baseline_raw, candidate_raw = baseline.read_bytes(), candidate.read_bytes() ceiling = min(len(baseline_raw), FROZEN_SIZE_CEILING) baseline_report = inspect_model(baseline_raw, str(baseline.resolve())) candidate_report = inspect_model(candidate_raw, str(candidate.resolve())) failures = [f"baseline: {s}" for s in baseline_report["failures"]] failures.extend(f"candidate: {s}" for s in candidate_report["failures"]) if len(baseline_raw) > FROZEN_SIZE_CEILING: failures.append(f"baseline exceeds frozen {FROZEN_SIZE_CEILING}-byte ceiling") if len(candidate_raw) > ceiling: failures.append(f"candidate is {len(candidate_raw)} bytes; ceiling is {ceiling}") return {"passed": not failures, "size_ceiling_bytes": ceiling, "frozen_ceiling_bytes": FROZEN_SIZE_CEILING, "baseline": baseline_report, "candidate": candidate_report, "failures": failures, "scope": "Serialized graph and file-size gate only; accuracy, runtime, and hardware performance are unmeasured."} def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--candidate", type=Path, required=True) parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE) parser.add_argument("--json", action="store_true", help="print machine-readable inspection") args = parser.parse_args() try: result = verify(args.candidate, args.baseline) except (OSError, ModelFormatError) as exc: if args.json: print(json.dumps({"passed": False, "error": str(exc)})) else: print(f"ERROR: {exc}", file=sys.stderr) return 2 if args.json: print(json.dumps(result, indent=2)) else: for label in ("baseline", "candidate"): model = result[label] print(f"{label}: {model['path']}\n bytes={model['bytes']} sha256={model['sha256']}") print(f" operators={model['operator_counts']} versions={model['operator_versions']}") print(f" tensor_dtypes={model['tensor_dtypes']}") for io in ("inputs", "outputs"): for tensor in model[io]: print(f" {io}: shape={tensor['shape']} signature={tensor['shape_signature']} dtype={tensor['dtype']} " f"scale={tensor['scales']} zero_point={tensor['zero_points']}") print(f"size ceiling: {result['size_ceiling_bytes']} bytes") for failure in result["failures"]: print(f"FAIL: {failure}") print("PASS" if result["passed"] else "FAIL") print(result["scope"]) return 0 if result["passed"] else 1 if __name__ == "__main__": sys.exit(main())