| """Command-line interface. |
| |
| bleep find audio.wav --recall high |
| bleep redact audio.wav # writes a review plan, not audio |
| bleep redact audio.wav --apply --out red.wav # needs a decided plan |
| bleep union audio.wav --asr-spans spans.json |
| bleep info |
| |
| **No command in this module can print what was said.** ``find`` emits |
| category + time; ``redact`` emits a plan of categories + times; ``union`` takes |
| someone else's spans and returns spans. :mod:`bleep.asr` and |
| :mod:`bleep.textpii` are not imported here at all, so no code path reachable |
| from the CLI can produce a transcript. |
| ``tests/test_no_content_path.py`` asserts both the import property and that the |
| rendered output contains no transcript-like text. |
| |
| Claim demonstrated by this module: no-content, and the reviewability |
| requirement (``redact`` defaults to plan-only). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| from pathlib import Path |
|
|
| from .audio import load_audio |
| from .redact import MODES, RedactionPlan, Redactor, verify_redaction |
| from .spans import ANY, CATEGORIES, Span |
| from .tagger import load_model, measure_rtf, predict_scores |
| from .threshold import ( |
| PRESETS, |
| OperatingPoint, |
| RecallFirstThresholding, |
| get_operating_point, |
| load_operating_point, |
| ) |
| from .union import UnionCombiner |
|
|
| DEFAULT_CKPT = Path("checkpoints/bleep-base.pt") |
|
|
| |
| |
| |
| from . import DISCLAIMER |
|
|
|
|
| |
|
|
|
|
| def _resolve_model(path: str | None): |
| p = Path(path) if path else DEFAULT_CKPT |
| if not p.exists(): |
| raise SystemExit( |
| f"no checkpoint at {p}.\n" |
| f"Train one: python scripts/train.py --data data --preset bleep-base\n" |
| f"Or point at one: bleep ... --model /path/to/bleep-base.pt" |
| ) |
| return load_model(p) |
|
|
|
|
| def _operating_points_file(model_path: str | None) -> Path: |
| """Where a calibrated operating-point set lives: next to the checkpoint.""" |
| base = Path(model_path) if model_path else DEFAULT_CKPT |
| return base.parent / "operating_points.json" |
|
|
|
|
| def _operating_point(args, frame_rate: float): |
| """Resolve the decision rule, preferring a calibrated threshold. |
| |
| Resolution order: an explicit ``--operating-point-file``, then the |
| ``operating_points.json`` written next to the checkpoint by |
| ``run_benchmarks.py --calibrate-only``, then failure with the command to |
| run. A preset's *threshold* is not portable across checkpoints, so |
| inventing one here would silently produce a decision rule nobody measured. |
| """ |
| if getattr(args, "operating_point_file", None): |
| op = load_operating_point(args.operating_point_file) |
| else: |
| op = get_operating_point(args.recall) |
| calibrated = _operating_points_file(getattr(args, "model", None)) |
| if calibrated.exists(): |
| try: |
| blob = json.loads(calibrated.read_text(encoding="utf-8")) |
| except json.JSONDecodeError as e: |
| raise SystemExit(f"{calibrated} is not valid JSON: {e}") |
| if args.recall in blob: |
| op = OperatingPoint.from_dict(blob[args.recall]) |
| if op.threshold is None: |
| raise SystemExit( |
| f"operating point {op.name!r} has no calibrated threshold for this " |
| f"checkpoint.\n\nCalibrate it on the held-out `calib` split:\n" |
| f" python scripts/run_benchmarks.py --data data " |
| f"--checkpoint {args.model or DEFAULT_CKPT} --calibrate-only\n\n" |
| f"That writes {_operating_points_file(getattr(args, 'model', None))}, " |
| f"which this command picks up automatically." |
| ) |
| if getattr(args, "pre_margin", None) is not None: |
| op.pre_margin_s = args.pre_margin |
| if getattr(args, "post_margin", None) is not None: |
| op.post_margin_s = args.post_margin |
| return RecallFirstThresholding(op, frame_rate) |
|
|
|
|
| def _detect(args) -> tuple[list[Span], dict]: |
| model, feat, extra = _resolve_model(args.model) |
| audio = load_audio(args.audio) |
| rf = _operating_point(args, model.cfg.frame_rate) |
| scores = predict_scores(model, audio.samples, feat, device=args.device, |
| chunk_seconds=args.chunk_seconds) |
| typed = rf.decode_typed(scores, duration=audio.duration) |
| anyv = rf.decode_any(scores, duration=audio.duration) |
| meta = { |
| "audio": str(args.audio), |
| "duration_s": round(audio.duration, 3), |
| "sample_rate": audio.sample_rate, |
| "model": extra.get("preset", model.cfg.name), |
| "param_count": model.param_count(), |
| "operating_point": rf.op.as_dict(), |
| "frame_rate_hz": model.cfg.frame_rate, |
| } |
| |
| |
| spans = list(typed) |
| for s in anyv: |
| if not any(t.overlap(s) > 0.5 * s.duration for t in typed): |
| spans.append(s) |
| return sorted(spans, key=lambda s: (s.start, s.label)), meta |
|
|
|
|
| |
| |
| |
|
|
|
|
| def cmd_find(args) -> int: |
| """Report sensitive regions: category + time only. |
| |
| Claim: no-content -- the output is category and time, nothing else. |
| """ |
| spans, meta = _detect(args) |
| if args.json: |
| print(json.dumps({"meta": meta, "spans": [s.as_dict() for s in spans], |
| "disclaimer": DISCLAIMER}, indent=2)) |
| return 0 |
| print(f"{meta['audio']} ({meta['duration_s']:.2f}s, " |
| f"model {meta['model']}, operating point " |
| f"{meta['operating_point']['name']} @ thr " |
| f"{meta['operating_point']['threshold']})") |
| if not spans: |
| print("\n no regions flagged") |
| else: |
| print(f"\n {'start':>8} {'end':>8} {'dur':>6} {'score':>5} category") |
| print(" " + "-" * 56) |
| for s in spans: |
| print(f" {s.start:8.2f} {s.end:8.2f} {s.duration:6.2f} " |
| f"{s.score:5.2f} {s.label}") |
| total = sum(s.duration for s in spans) |
| print(" " + "-" * 56) |
| print(f" {len(spans)} region(s), {total:.2f}s " |
| f"({total/meta['duration_s']:.1%} of the recording)") |
| print(f"\n{DISCLAIMER}") |
| return 0 |
|
|
|
|
| def cmd_redact(args) -> int: |
| """Plan a redaction (default) or apply a decided plan. |
| |
| Claim: the reviewability requirement -- plan by default, audio only from a |
| decided plan. |
| """ |
| audio = load_audio(args.audio) |
| red = Redactor(args.mode) |
|
|
| if args.apply: |
| if not args.plan: |
| raise SystemExit( |
| "--apply needs --plan PATH (the reviewed plan).\n" |
| "Produce one first: bleep redact audio.wav --plan-out plan.json" |
| ) |
| plan = RedactionPlan.load(args.plan) |
| if args.decisions_csv: |
| plan.apply_csv_decisions(Path(args.decisions_csv).read_text(encoding="utf-8")) |
| try: |
| out_audio, report = red.apply(audio, plan, |
| require_review=not args.no_review) |
| except ValueError as e: |
| raise SystemExit(f"refused: {e}") |
| out_path = Path(args.out or (Path(args.audio).with_suffix("")).name + ".redacted.wav") |
| Redactor.write(out_path, out_audio, source_path=args.audio) |
| v = verify_redaction(audio, out_audio, plan) |
| print(json.dumps({"written": str(out_path), "report": report, |
| "verification": v}, indent=2)) |
| if not v["ok"]: |
| print("WARNING: redaction verification failed", file=sys.stderr) |
| return 1 |
| return 0 |
|
|
| spans, meta = _detect(args) |
| plan = red.plan(audio, spans, source_path=str(args.audio), |
| operating_point=meta["operating_point"], |
| auto_approve=args.auto_approve) |
| plan_path = Path(args.plan_out or |
| Path(args.audio).with_suffix(".bleep-plan.json").name) |
| plan.save(plan_path) |
| csv_path = plan_path.with_suffix(".csv") |
| csv_path.write_text(plan.to_csv(), encoding="utf-8") |
| print(json.dumps(plan.summary(), indent=2)) |
| print(f"\nplan: {plan_path}\nreview: {csv_path}") |
| print("\nNo audio was modified. Review the CSV (set each `decision` to " |
| "approved/rejected), then:") |
| print(f" bleep redact {args.audio} --apply --plan {plan_path} " |
| f"--decisions-csv {csv_path} --out redacted.wav") |
| print(f"\n{DISCLAIMER}") |
| return 0 |
|
|
|
|
| def cmd_union(args) -> int: |
| """Union Bleep's detections with an existing detector's spans. |
| |
| Claim: complementarity and containment; also no-content, since text on the |
| incoming spans is dropped. |
| """ |
| spans, meta = _detect(args) |
| payload = json.loads(Path(args.asr_spans).read_text(encoding="utf-8")) |
| raw = payload.get("spans", payload) if isinstance(payload, dict) else payload |
| asr = [] |
| for d in raw: |
| |
| |
| |
| asr.append(Span(float(d["start"]), float(d["end"]), |
| d.get("label") if d.get("label") in CATEGORIES else ANY, |
| float(d.get("score", 1.0)), "asr")) |
| uc = UnionCombiner(args.gap_tolerance) |
| bd = uc.combine(asr, spans, duration=meta["duration_s"]) |
| ok, missing = uc.verify_containment(asr, bd) |
| out = { |
| "meta": meta, |
| "union": [s.as_dict() for s in bd.union], |
| "breakdown": bd.as_dict(), |
| "containment_holds": ok, |
| "containment_violations": [s.as_dict() for s in missing], |
| "disclaimer": DISCLAIMER, |
| } |
| print(json.dumps(out, indent=2)) |
| return 0 if ok else 1 |
|
|
|
|
| def cmd_info(args) -> int: |
| """Model size, receptive field, measured speed, and the size-band check. |
| |
| Claim: on-device speed -- size, receptive field, measured real-time |
| factor. |
| """ |
| from .tagger import build, in_target_band |
|
|
| if args.model or DEFAULT_CKPT.exists(): |
| model, feat, extra = _resolve_model(args.model) |
| info = { |
| "checkpoint": str(args.model or DEFAULT_CKPT), |
| "preset": extra.get("preset", model.cfg.name), |
| "trained_step": extra.get("step"), |
| "validation": extra.get("val"), |
| } |
| else: |
| model = build(args.preset) |
| feat = None |
| info = {"checkpoint": None, "preset": args.preset, |
| "note": "no checkpoint found; reporting the untrained architecture"} |
| from .features import DEFAULT_FEATURES |
| info.update( |
| param_count=model.param_count(), |
| param_count_millions=round(model.param_count() / 1e6, 2), |
| in_target_size_band_0p05_0p15B=in_target_band(model), |
| receptive_field_seconds=round(model.cfg.receptive_field_seconds, 2), |
| output_frame_rate_hz=model.cfg.frame_rate, |
| gmac_per_audio_second=round( |
| model.macs_per_output_frame() * model.cfg.frame_rate / 1e9, 3), |
| ) |
| if args.speed: |
| info["speed"] = measure_rtf(model, feat or DEFAULT_FEATURES, |
| seconds=args.speed_seconds, |
| device=args.device, repeats=3) |
| info["operating_points"] = {k: v.as_dict() for k, v in PRESETS.items()} |
| info["categories"] = list(CATEGORIES) |
| info["disclaimer"] = DISCLAIMER |
| print(json.dumps(info, indent=2)) |
| return 0 |
|
|
|
|
| |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| """Build the argument parser. |
| |
| Claim: no-content -- there is no subcommand that could emit a transcript. |
| """ |
| ap = argparse.ArgumentParser( |
| prog="bleep", |
| description="Transcript-free detection of sensitive speech regions " |
| "(category + time only).", |
| epilog=DISCLAIMER, |
| ) |
| sub = ap.add_subparsers(dest="cmd", required=True) |
|
|
| def common(p): |
| """Add the arguments every audio-taking subcommand shares. |
| |
| Claim: recall at fixed false-positive rate -- the operating point is selectable on every command. |
| """ |
| p.add_argument("audio", type=Path) |
| p.add_argument("--model", default=None, help="checkpoint path") |
| p.add_argument("--recall", default="high", choices=sorted(PRESETS), |
| help="operating point (default: high = recall-first)") |
| p.add_argument("--operating-point-file", default=None, |
| help="calibrated operating point JSON") |
| p.add_argument("--pre-margin", type=float, default=None, |
| help="override leading margin, seconds") |
| p.add_argument("--post-margin", type=float, default=None) |
| p.add_argument("--device", default="cpu") |
| p.add_argument("--chunk-seconds", type=float, default=30.0) |
|
|
| p = sub.add_parser("find", help="report sensitive regions (category + time)") |
| common(p) |
| p.add_argument("--json", action="store_true") |
| p.set_defaults(func=cmd_find) |
|
|
| p = sub.add_parser("redact", help="plan a redaction; --apply to write audio") |
| common(p) |
| p.add_argument("--mode", default="silence", choices=MODES) |
| p.add_argument("--plan-out", default=None) |
| p.add_argument("--auto-approve", action="store_true", |
| help="mark every region approved in the plan (skips review)") |
| p.add_argument("--apply", action="store_true", help="write redacted audio") |
| p.add_argument("--plan", default=None, help="reviewed plan to apply") |
| p.add_argument("--decisions-csv", default=None) |
| p.add_argument("--no-review", action="store_true", |
| help="apply pending regions without review (recorded in the plan)") |
| p.add_argument("--out", default=None) |
| p.set_defaults(func=cmd_redact) |
|
|
| p = sub.add_parser("union", help="union Bleep with an existing detector's spans") |
| common(p) |
| p.add_argument("--asr-spans", required=True, |
| help="JSON list of {start,end[,label,score]} from your detector") |
| p.add_argument("--gap-tolerance", type=float, default=0.08) |
| p.set_defaults(func=cmd_union) |
|
|
| p = sub.add_parser("info", help="model size, receptive field, speed") |
| p.add_argument("--model", default=None) |
| p.add_argument("--preset", default="bleep-base") |
| p.add_argument("--speed", action="store_true", help="measure real-time factor") |
| p.add_argument("--speed-seconds", type=float, default=60.0) |
| p.add_argument("--device", default="cpu") |
| p.set_defaults(func=cmd_info) |
| return ap |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| """CLI entry point. |
| |
| Claim: no-content. |
| """ |
| args = build_parser().parse_args(argv) |
| return args.func(args) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
