| """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 = "<memory>") -> 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") |
| |
| 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 |
| 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("<i", axes["_data"])] |
| rank = len(tensors[op_inputs[0]]["shape"]) |
| check(bool(axes_values) and len(set(axes_values)) == len(axes_values) and all(-rank <= a < rank for a in axes_values), |
| f"operator {oi} mean axes are invalid") |
| for tensor in tensors: |
| index = tensor["index"] |
| if tensor["dtype"] == "int32": |
| check(index in integer_roles, f"int32 tensor {index} is not a bias or reduction axis") |
| elif index not in weight_roles: |
| check(len(tensor["scales"]) == 1, f"activation tensor {index} must use per-tensor quantization") |
| check(all(index in produced for index in outputs), "model output must be produced by the graph") |
| check(last_op == 25, "model must end with SOFTMAX") |
| if operators: |
| check(fb.vector(operators[-1], 2, "i") == outputs, "SOFTMAX must produce the model output") |
| for output in report["outputs"]: |
| check(output["scales"] == [1.0 / 256] and output["zero_points"] == [-128], |
| "int8 SOFTMAX output must have scale 1/256 and zero point -128") |
| report["operator_counts"] = dict(sorted(counts.items())) |
| report["operator_versions"] = {name: sorted(v) for name, v in sorted(versions.items())} |
| report["tensor_dtypes"] = dict(Counter(t["dtype"] for t in tensors)) |
| return report |
|
|
|
|
| def verify(candidate: Path, baseline: Path) -> 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()) |
|
|
