"""Inspect saved legacy Hume batch prosody and Oruk file responses locally. Python 3.10+; no packages, credentials, or network requests. python compare-legacy-hume.py --hume hume-predictions.json \ --oruk analysis.json --output migration-review.json Optionally add --audio recording.wav --manifest provenance.json to bind local bytes and declared time origins. Local hashes do not verify which recording a provider processed. This tool cannot align speakers, convert scores, or measure accuracy. It supports the legacy batch predictions export, not current Hume Tagger, EVI, or live messages. Code: MIT. Input recordings and responses retain their own permissions. """ from __future__ import annotations import argparse from bisect import bisect_left, bisect_right import hashlib import json import math import os from pathlib import Path import re import stat import sys MAX_JSON_BYTES = 32 * 1024 * 1024 MAX_MANIFEST_BYTES = 16 * 1024 MAX_AUDIO_BYTES = 512 * 1024 * 1024 MAX_OUTPUT_BYTES = 96 * 1024 * 1024 MAX_JSON_DEPTH = 64 MAX_JSON_NODES = 250_000 def bounded_tree(value): pending = [(value, 0)] count = 0 while pending: item, depth = pending.pop() count += 1 if depth > MAX_JSON_DEPTH or count > MAX_JSON_NODES: raise ValueError("JSON structure exceeds the depth or node limit") if isinstance(item, float) and not math.isfinite(item): raise ValueError("JSON contains a nonfinite number") if isinstance(item, dict): pending.extend((child, depth + 1) for child in item.values()) elif isinstance(item, list): pending.extend((child, depth + 1) for child in item) def exact_keys(value, fields, name): value = object_value(value, name) if set(value) != set(fields): raise ValueError(name + " has missing or unsupported fields") return value def digest(value): if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value): raise ValueError("manifest SHA-256 values must be lowercase hexadecimal") return value def manifest_contract(value): exact_keys(value, ("schema_version", "kind", "recording", "hume", "oruk", "provider_association"), "manifest") if type(value["schema_version"]) is not int or value["schema_version"] != 1 or value["kind"] != "oruk.hume-comparison.provenance": raise ValueError("unsupported provenance manifest") if value["provider_association"] != "customer_declared_same_recording": raise ValueError("manifest must distinguish customer declaration from provider verification") recording = exact_keys(value["recording"], ("sha256", "bytes"), "recording") digest(recording["sha256"]) if type(recording["bytes"]) is not int or not 1 <= recording["bytes"] <= MAX_AUDIO_BYTES: raise ValueError("recording bytes must be positive and at most 512 MiB") hume = exact_keys(value["hume"], ("response_sha256", "source_index", "file_index", "time_origin_seconds"), "Hume provenance") oruk = exact_keys(value["oruk"], ("response_sha256", "time_origin_seconds"), "Oruk provenance") for selected in (hume, oruk): digest(selected["response_sha256"]) number(selected["time_origin_seconds"], "time_origin_seconds") for name in ("source_index", "file_index"): item = hume[name] if item is None and name == "file_index": continue # A failed source can have no selected file. if type(item) is not int or not 0 <= item < 20_000: raise ValueError("manifest selection must be a bounded explicit index") return value def shifted(value, offset): return number(value + offset, "derived recording time") def shifted_interval(item, offset): start, end = shifted(item["start"], offset), shifted(item["end"], offset) if item["end"] > item["start"] and end <= start: raise ValueError("time origin loses interval precision") return {"start": start, "end": end} def name_value(value): if not isinstance(value, str) or not value or len(value.encode("utf-8")) > 256: raise ValueError("observed dimension names must be nonempty and at most 256 UTF-8 bytes") return value def observed_dimensions(value, key): """Inventory returned names only; do not infer a model's complete taxonomy.""" if value is None: return [] if isinstance(value, dict): return sorted(name_value(name) for name in value) names = set() for item in array(value, "observed dimensions"): if isinstance(item, str): names.add(name_value(item)) elif isinstance(item, dict) and key in item: names.add(name_value(item[key])) return sorted(names) def dimension_report(source, file_index, oruk): predictions = [] models = {} if file_index is not None: models = source["results"]["predictions"][file_index]["models"] prosody = models.get("prosody") if prosody is not None: predictions = [item for group in prosody["grouped_predictions"] for item in group["predictions"]] def inventory(values, key): return {"observed_names": sorted({name for value in values for name in observed_dimensions(value, key)}), "mapping": "unmapped", "complete_model_taxonomy": False} result = { "hume_prosody_emotions": inventory([item["emotions"] for item in predictions], "name"), "oruk_recording_emotions": inventory([oruk["emotions"]], "label"), "oruk_segment_emotions": inventory([item["emotions"] for item in oruk["segments"]], "label"), "oruk_recording_styles": inventory([oruk.get("styles")], "label"), "oruk_segment_styles": inventory([item.get("styles") for item in oruk["segments"]], "label"), } additional = [] for location, item in [("recording", oruk), *[(f"segment:{i}", s) for i, s in enumerate(oruk["segments"])]]: for field in ("scores", "axes", "unipolar", "labels"): if field in item: additional.append({"location": location, "field": field, "value_state": "null" if item[field] is None else "present", **inventory([item[field]], "label"), "interpretation": "unhandled_native_signal"}) result["additional_oruk_signals"] = additional # Metadata, transcripts and usage are preserved but are not signal dimensions. understood = {"id", "object", "task", "model", "duration", "emotions", "styles", "segments", "text", "tagged_text", "language", "usage", "scores", "axes", "unipolar", "labels"} return result, [{"name": name_value(name), "value_state": "null" if value is None else "present"} for name, value in models.items() if name != "prosody"], [name_value(name) for name in oruk if name not in understood] def object_value(value: object, field: str) -> dict: if not isinstance(value, dict): raise ValueError(f"{field} must be an object") return value def array(value: object, field: str) -> list: if not isinstance(value, list) or len(value) > 20_000: raise ValueError(f"{field} must be an array of at most 20,000 items") return value def number(value: object, field: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{field} must be finite and nonnegative") try: valid = math.isfinite(value) and value >= 0 except OverflowError: valid = False if not valid: raise ValueError(f"{field} must be finite and nonnegative") return value def interval(start: object, end: object, field: str) -> tuple[float, float]: start = number(start, field + ".start") end = number(end, field + ".end") if end < start: raise ValueError(f"{field} ends before it starts") return start, end def scores(value: object, key: str, field: str) -> None: names = set() for item in array(value, field): item = object_value(item, field + " item") name = item.get(key) if not isinstance(name, str) or not name or name in names: raise ValueError(f"{field} needs unique, nonempty {key} values") name_value(name) if number(item.get("score"), field + ".score") > 1: raise ValueError(f"{field} scores must be between 0 and 1") names.add(name) def choose(items: list, index: int | None, option: str) -> int: if index is None: if len(items) != 1: raise ValueError(f"found {len(items)} entries; select one with {option} (zero-based)") return 0 if isinstance(index, bool) or not isinstance(index, int) or not 0 <= index < len(items): raise ValueError(f"{option} is outside the available entries") return index def hume_intervals(export: object, source_index: int | None, file_index: int | None): sources = array(export, "Hume batch export") selected_source = choose(sources, source_index, "--source-index") source = object_value(sources[selected_source], "Hume source result") object_value(source.get("source"), "Hume source descriptor") error = source.get("error") if error is not None and not isinstance(error, str): raise ValueError("Hume source error must be a string or null") results = source.get("results") if results is None and error: if file_index is not None: raise ValueError("the failed Hume source contains no file to select") return source, selected_source, None, [], ["Hume source failed; its error is preserved."] results = object_value(results, "Hume results") errors = array(results.get("errors", []), "Hume results.errors") files = array(results.get("predictions"), "Hume results.predictions") notes = ["Hume returned errors; review the preserved source and results.errors."] if error or errors else [] if not files and (error or errors): if file_index is not None: raise ValueError("the failed Hume source contains no file to select") return source, selected_source, None, [], notes selected_file = choose(files, file_index, "--file-index") file = object_value(files[selected_file], "Hume file prediction") models = object_value(file.get("models"), "Hume models") prosody = models.get("prosody") if prosody is None: return source, selected_source, selected_file, [], notes + ["Selected Hume file has no prosody result."] groups = array(object_value(prosody, "Hume prosody").get("grouped_predictions"), "Hume groups") intervals = [] for group_index, group in enumerate(groups): group = object_value(group, "Hume group") for prediction_index, prediction in enumerate(array(group.get("predictions"), "Hume predictions")): prediction = object_value(prediction, "Hume prediction") time = object_value(prediction.get("time"), "Hume prediction.time") start, end = interval(time.get("begin"), time.get("end"), "Hume interval") scores(prediction.get("emotions"), "name", "Hume emotions") intervals.append({"group_index": group_index, "prediction_index": prediction_index, "start": start, "end": end}) if len(intervals) > 20_000: raise ValueError("Hume file exceeds this example's 20,000-interval limit") return source, selected_source, selected_file, intervals, notes def compare(hume: object, oruk: object, source_index=None, file_index=None, *, manifest=None) -> dict: bounded_tree(hume) bounded_tree(oruk) hume_offset = oruk_offset = 0 if manifest is not None: manifest_contract(manifest) for index, name in ((source_index, "source_index"), (file_index, "file_index")): if index is not None and (type(index) is not int or index != manifest["hume"][name]): raise ValueError("CLI selection conflicts with the provenance manifest") source_index, file_index = manifest["hume"]["source_index"], manifest["hume"]["file_index"] hume_offset, oruk_offset = manifest["hume"]["time_origin_seconds"], manifest["oruk"]["time_origin_seconds"] source, source_index, file_index, legacy, notes = hume_intervals(hume, source_index, file_index) if manifest is not None and (source_index, file_index) != (manifest["hume"]["source_index"], manifest["hume"]["file_index"]): raise ValueError("manifest selection does not identify the selected source and file") oruk = object_value(oruk, "Oruk response") if oruk.get("object") != "speech.result" or oruk.get("task") not in {"analysis", "affect", "emotions"}: raise ValueError("expected an Oruk analysis, affect, or emotions speech.result") if not isinstance(oruk.get("model"), str) or not oruk["model"]: raise ValueError("Oruk response is missing its model") duration = number(oruk.get("duration"), "Oruk duration") scores(oruk.get("emotions"), "label", "Oruk recording emotions") segments = array(oruk.get("segments"), "Oruk segments") for segment in segments: segment = object_value(segment, "Oruk segment") _, end = interval(segment.get("start"), segment.get("end"), "Oruk interval") # The file gateway accumulates floating-point chunk offsets separately # from total duration. Permit only a microsecond of numerical excess. if end > duration + 0.000001: raise ValueError("Oruk segment exceeds the returned recording duration") scores(segment.get("emotions"), "label", "Oruk segment emotions") # Only derived coordinates change. Native responses and interval values remain intact. timed_legacy = [shifted_interval(item, hume_offset) for item in legacy] timed_segments = [shifted_interval(item, oruk_offset) for item in segments] ordered = sorted(enumerate(timed_legacy), key=lambda item: item[1]["start"]) starts = [item["start"] for _, item in ordered] latest_end = [] for _, item in ordered: latest_end.append(max(latest_end[-1] if latest_end else 0, item["end"])) alignment, matched = [], set() comparisons = link_count = 0 for index, segment in enumerate(timed_segments): links = [] first = bisect_right(latest_end, segment["start"]) last = bisect_left(starts, segment["end"]) for legacy_index, item in ordered[first:last]: comparisons += 1 if comparisons > 2_000_000: raise ValueError("too many interval comparisons; review a smaller recording") overlap = min(segment["end"], item["end"]) - max(segment["start"], item["start"]) if overlap > 0: link_count += 1 if link_count > 100_000: raise ValueError("too many overlapping intervals (100,000 links)") matched.add(legacy_index) links.append({"hume_interval_index": legacy_index, "overlap_seconds": overlap}) links.sort(key=lambda item: item["hume_interval_index"]) alignment.append({"oruk_segment_index": index, "hume_intervals": links}) if not legacy: notes.append("No timed Hume prosody predictions; no score or timing equivalence established.") if not segments: notes.append("No Oruk segments; recording scores were not assigned to guessed intervals.") dimensions, unsupported_modalities, unhandled_fields = dimension_report(source, file_index, oruk) return { "schema_version": 2, "scope": "legacy Hume batch prosody versus Oruk saved file response; inspection only", "audio_identity_verified": False, "provenance": {"local_bytes_verified": False, "provider_association_verified": False, "customer_declaration": manifest, "manifest_sha256": None}, "time_alignment": { "basis": "customer_declared_recording_offsets" if manifest is not None else "unverified_zero_origin_assumption", "provider_verified": False, "unit": "seconds", "normalization": "recording_seconds = response_seconds + time_origin_seconds", "hume_time_origin_seconds": hume_offset, "oruk_time_origin_seconds": oruk_offset, }, "unmapped_dimensions": dimensions, "unsupported_hume_modalities": unsupported_modalities, "unhandled_oruk_fields": unhandled_fields, "notes": [ "Confirm the common recording and each declared response time origin; local hashes do not prove provider association." if manifest is not None else "Confirm that both requests used the same untrimmed recording and time origin; the zero-origin alignment is unverified.", "Positive time overlap does not establish speaker identity or label agreement.", "Scores are preserved, not converted, averaged, normalized, or thresholded.", "Every observed dimension remains unmapped, including identical names. No shared calibration or complete taxonomy is inferred.", "Unsupported modalities and unhandled fields are preserved, not evaluated. An unobserved label is not an unsupported model class.", "An absent Oruk label has no returned score; it is not zero. Fallback selection is not inferred.", "No current Hume Tagger, Prosody, EVI, or accuracy evaluation is performed.", *notes, ], "hume_selection": {"source_index": source_index, "file_index": file_index}, "hume_source": source, "oruk": oruk, "hume_intervals": legacy, "alignment": alignment, "unmatched_hume_interval_indices": [index for index in range(len(legacy)) if index not in matched], "unmatched_oruk_segment_indices": [item["oruk_segment_index"] for item in alignment if not item["hume_intervals"]], } def read_file(path: Path, limit: int, *, hash_only=False): """Read one explicitly selected regular file; never decode audio or follow URLs.""" def identity(info): return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns) selected = os.lstat(path) if not stat.S_ISREG(selected.st_mode) or not 0 < selected.st_size <= limit: raise ValueError("input is not a nonempty regular file within its byte limit") flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0) fd = os.open(path, flags) with os.fdopen(fd, "rb") as file: before = os.fstat(file.fileno()) if not stat.S_ISREG(before.st_mode) or identity(selected) != identity(before): raise ValueError("input changed before it could be read") digest_value = hashlib.sha256() size = 0 data = bytearray() while True: chunk = file.read(min(1024 * 1024, limit + 1 - size)) if not chunk: break size += len(chunk) if size > limit: raise ValueError("input exceeds its byte limit") digest_value.update(chunk) if not hash_only: data.extend(chunk) after = os.fstat(file.fileno()) final_path = os.lstat(path) if (identity(before) != identity(after) or size != before.st_size or not stat.S_ISREG(final_path.st_mode) or identity(after) != identity(final_path)): raise ValueError("input changed while it was being read") return data, digest_value.hexdigest(), size def read_json(path: Path, limit=MAX_JSON_BYTES): data, digest_value, _ = read_file(path, limit) def pairs(items): result = {} for name, value in items: if name in result: raise ValueError("JSON contains a duplicate key") result[name] = value return result def reject_constant(_): raise ValueError("JSON contains a nonfinite number") value = json.loads(data, object_pairs_hook=pairs, parse_constant=reject_constant) bounded_tree(value) return value, digest_value def serialized_report(result): output = bytearray() for item in json.JSONEncoder(ensure_ascii=False, allow_nan=False, indent=2).iterencode(result): encoded = item.encode("utf-8") if len(output) + len(encoded) + 1 > MAX_OUTPUT_BYTES: raise ValueError("report exceeds the 96 MiB output limit") output.extend(encoded) output.extend(b"\n") return output class SafeParser(argparse.ArgumentParser): def error(self, message): self.exit(2, "Invalid arguments. Use --help; argument values are not echoed.\n") def main() -> int: parser = SafeParser(description=__doc__) parser.add_argument("--hume", required=True, type=Path) parser.add_argument("--oruk", required=True, type=Path) parser.add_argument("--output", required=True, type=Path, help="new JSON file; existing files are never overwritten") parser.add_argument("--source-index", type=int, help="zero-based source index; required for multi-source exports") parser.add_argument("--file-index", type=int, help="zero-based file index; required for multi-file sources") parser.add_argument("--audio", type=Path, help="explicit local recording, streamed for SHA-256; requires --manifest") parser.add_argument("--manifest", type=Path, help="strict recording/response/time-origin declaration; requires --audio") args = parser.parse_args() try: hume, hume_hash = read_json(args.hume) oruk, oruk_hash = read_json(args.oruk) manifest = manifest_hash = None if (args.audio is None) != (args.manifest is None): raise ValueError("--audio and --manifest must be supplied together") if args.manifest is not None: manifest, manifest_hash = read_json(args.manifest, MAX_MANIFEST_BYTES) manifest_contract(manifest) if hume_hash != manifest["hume"]["response_sha256"] or oruk_hash != manifest["oruk"]["response_sha256"]: raise ValueError("response bytes do not match the manifest SHA-256 values") _, audio_hash, audio_size = read_file(args.audio, MAX_AUDIO_BYTES, hash_only=True) if (audio_hash, audio_size) != (manifest["recording"]["sha256"], manifest["recording"]["bytes"]): raise ValueError("local recording bytes do not match the manifest") result = compare(hume, oruk, args.source_index, args.file_index, manifest=manifest) result["input_sha256"] = {"hume_export": hume_hash, "oruk_response": oruk_hash} if manifest is not None: result["provenance"].update(local_bytes_verified=True, manifest_sha256=manifest_hash) serialized = serialized_report(result) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) fd = os.open(args.output, flags, 0o600) with os.fdopen(fd, "wb") as file: file.write(serialized) print("Saved local inspection. Provider recording association, alignment and score agreement are not verified.") return 0 except (OSError, ValueError, TypeError, RecursionError, OverflowError): print("Cannot compare responses: check the selected files, bounds, manifest hashes and selection. No provider call was made.", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())