Skip to content

train

birdnet_stm32.cli.train

CLI entry point for training.

AdaptiveLoaderTuner

Bases: Callback

Tune loader in-flight queue online using throughput and free RAM.

Source code in birdnet_stm32/cli/train.py
class AdaptiveLoaderTuner(tf.keras.callbacks.Callback):
    """Tune loader in-flight queue online using throughput and free RAM."""

    def __init__(
        self,
        control: dict,
        batch_size: int,
        adjust_every: int = 200,
        min_inflight: int = 128,
        max_inflight: int = 4096,
        target_free_gb: float = 8.0,
    ):
        super().__init__()
        self.control = control
        self.batch_size = batch_size
        self.adjust_every = adjust_every
        self.min_inflight = min_inflight
        self.max_inflight = max_inflight
        self.target_free_gb = target_free_gb
        self._t0 = 0.0
        self._steps = 0
        self._durations: list[float] = []
        self._prev_throughput: float | None = None
        self._direction = 1
        self._step_size = max(32, min(512, min_inflight // 2))

    def on_train_batch_begin(self, batch, logs=None):
        self._t0 = time.perf_counter()

    def on_train_batch_end(self, batch, logs=None):
        dt = time.perf_counter() - self._t0
        self._durations.append(dt)
        self._steps += 1
        if self._steps % self.adjust_every != 0:
            return

        step_sec = float(np.mean(self._durations)) if self._durations else 0.0
        self._durations.clear()
        throughput = self.batch_size / max(step_sec, 1e-6)
        _total_gb, avail_gb = _read_meminfo_gb()

        cur = int(self.control.get("max_inflight_files", self.min_inflight))
        new = cur
        reason = "hold"

        if avail_gb > 0 and avail_gb < self.target_free_gb:
            new = max(self.min_inflight, int(cur * 0.8))
            self._direction = -1
            self._step_size = max(16, self._step_size // 2)
            reason = "memory-pressure"
        elif self._prev_throughput is None:
            new = min(self.max_inflight, cur + self._step_size)
            self._direction = 1
            reason = "initial-probe"
        elif throughput >= self._prev_throughput * 1.01:
            if self._direction >= 0:
                new = min(self.max_inflight, cur + self._step_size)
                reason = "throughput-up"
            else:
                new = max(self.min_inflight, cur - self._step_size)
                reason = "throughput-up-down-dir"
        else:
            self._step_size = max(16, self._step_size // 2)
            if self._direction >= 0:
                new = max(self.min_inflight, cur - self._step_size)
                self._direction = -1
                reason = "reverse-down"
            else:
                new = min(self.max_inflight, cur + self._step_size)
                self._direction = 1
                reason = "reverse-up"

        self._prev_throughput = throughput
        if new != cur:
            self.control["max_inflight_files"] = int(new)
        self.control["last_tuning_event"] = {
            "step": int(self._steps),
            "throughput": float(throughput),
            "available_gb": float(avail_gb),
            "previous_inflight": int(cur),
            "current_inflight": int(self.control.get("max_inflight_files", cur)),
            "reason": reason,
        }

HostMemoryGuard

Bases: Callback

Abort training before host memory pressure makes the machine unusable.

Source code in birdnet_stm32/cli/train.py
class HostMemoryGuard(tf.keras.callbacks.Callback):
    """Abort training before host memory pressure makes the machine unusable."""

    def __init__(
        self,
        check_every: int = 25,
        reserve_gb: float = 12.0,
        reserve_fraction: float = 0.20,
    ):
        super().__init__()
        self.check_every = max(1, int(check_every))
        self.reserve_gb = float(reserve_gb)
        self.reserve_fraction = float(reserve_fraction)
        self._steps = 0

    def _memory_state(self) -> tuple[float, float, float]:
        total_gb, available_gb = _read_meminfo_gb()
        adaptive_reserve_gb = min(self.reserve_gb, max(2.0, total_gb * self.reserve_fraction))
        return total_gb, available_gb, adaptive_reserve_gb

    def on_train_batch_end(self, batch, logs=None):
        self._steps += 1
        if self._steps % self.check_every != 0:
            return

        _total_gb, available_gb, adaptive_reserve_gb = self._memory_state()
        if available_gb > 0 and available_gb < adaptive_reserve_gb:
            raise MemoryError(
                "Host memory guard stopped training: "
                f"{available_gb:.1f} GiB available is below the "
                f"{adaptive_reserve_gb:.1f} GiB reserve. "
                "The last completed-epoch checkpoint is still usable."
            )

    def on_epoch_end(self, epoch, logs=None):
        _total_gb, available_gb, adaptive_reserve_gb = self._memory_state()
        if available_gb > 0:
            print(f"[memory] epoch={epoch + 1} available={available_gb:.1f} GiB reserve={adaptive_reserve_gb:.1f} GiB")

get_args()

Parse command-line arguments for training.

Sensible defaults are chosen so that most users only need to specify --data_path_train. SpecAugment, deterministic seeding, and gradient clipping are enabled by default.

Source code in birdnet_stm32/cli/train.py
def get_args() -> argparse.Namespace:
    """Parse command-line arguments for training.

    Sensible defaults are chosen so that most users only need to specify
    ``--data_path_train``. SpecAugment, deterministic seeding, and gradient
    clipping are enabled by default.
    """
    parser = argparse.ArgumentParser(description="Train STM32N6 audio classifier")

    # -- Data -----------------------------------------------------------------
    parser.add_argument("--data_path_train", type=str, required=True, help="Path to train dataset")
    parser.add_argument(
        "--data_path_val",
        type=str,
        default=None,
        help="Separate validation dataset root; disables the random --val_split",
    )
    parser.add_argument(
        "--classes_file",
        type=str,
        default=None,
        help="Ordered one-label-per-line output schema; noise remains an all-zero folder",
    )
    parser.add_argument(
        "--model_config",
        type=str,
        default="",
        help=(
            "Architecture config for --qat and --linear_probe. Defaults to the "
            "checkpoint's sibling _model_config.json; pass it explicitly when fine-tuning a "
            "checkpoint that inherited its architecture from an earlier step, such as a QAT "
            "checkpoint, which writes no config of its own."
        ),
    )
    parser.add_argument("--max_classes", type=int, default=None, help="Use top N classes by sample count")
    parser.add_argument("--max_samples", type=int, default=None, help="Max samples per class")
    parser.add_argument("--upsample_ratio", type=float, default=0.5, help="Upsample ratio for minority classes")

    # -- Audio ----------------------------------------------------------------
    parser.add_argument("--sample_rate", type=int, default=24000, help="Audio sample rate (Hz)")
    parser.add_argument("--num_mels", type=int, default=64, help="Number of mel bins")
    parser.add_argument("--spec_width", type=int, default=256, help="Spectrogram width (frames)")
    parser.add_argument("--fft_length", type=int, default=512, help="FFT length")
    parser.add_argument("--chunk_duration", type=float, default=2.5, help="Audio chunk duration (seconds)")
    parser.add_argument(
        "--max_duration",
        type=int,
        default=60,
        help=(
            "Maximum seconds to read per file. The loader still reads only the bytes it needs "
            "for the candidate chunks (smart-crop bounded by --max_chunks_per_file)."
        ),
    )
    parser.add_argument(
        "--audio_frontend",
        type=str,
        default="raw",
        choices=["raw", "hybrid", "librosa"],
        help="Audio frontend mode",
    )
    parser.add_argument(
        "--mag_scale",
        type=str,
        default="pwl",
        choices=["pwl", "none"],
        help="Magnitude scaling: learned hinge sum (pwl) or none",
    )
    parser.add_argument(
        "--input_compression",
        type=str,
        default="none",
        choices=["none", "sqrt", "log"],
        help="Compress a librosa/hybrid spectrogram input before it is quantized (computed on host/M55)",
    )

    # -- Model architecture ---------------------------------------------------
    parser.add_argument("--embeddings_size", type=int, default=512, help="Embeddings layer size")
    parser.add_argument("--alpha", type=float, default=1.0, help="Width multiplier")
    parser.add_argument("--depth_multiplier", type=int, default=1, help="Depth multiplier")
    parser.add_argument(
        "--head_pooling",
        type=str,
        default="gap",
        choices=list(HEAD_POOLINGS),
        help="Pooling head: global average (gap), or mean over frequency then max + mean over time",
    )
    parser.add_argument(
        "--dw_kernel_size",
        type=int,
        default=3,
        choices=list(DW_KERNEL_SIZES),
        help="Depthwise kernel size in stages 2-4 (stage 1 stays 3x3)",
    )
    parser.add_argument(
        "--stage_widths",
        type=int,
        nargs=4,
        default=list(STAGE_WIDTHS),
        metavar="W",
        help="Base output channels of the four stages, before --alpha",
    )
    parser.add_argument("--frontend_trainable", action="store_true", default=False)

    # -- Augmentation ---------------------------------------------------------
    parser.add_argument("--no_spec_augment", action="store_true", default=False, help="Disable SpecAugment")
    parser.add_argument("--freq_mask_max", type=int, default=8, help="Max frequency mask width (bins)")
    parser.add_argument("--time_mask_max", type=int, default=25, help="Max time mask width (frames)")
    parser.add_argument("--mixup_alpha", type=float, default=0.2, help="Mixup alpha")
    parser.add_argument("--mixup_probability", type=float, default=0.25, help="Mixup batch fraction")

    # -- Training -------------------------------------------------------------
    parser.add_argument("--batch_size", type=int, default=32, help="Batch size")
    parser.add_argument("--num_workers", type=int, default=8, help="Parallel data loading workers (0 = sequential)")
    parser.add_argument(
        "--max_chunks_per_file",
        type=int,
        default=1,
        help="Max salient chunks to extract per file open (reduces redundant I/O for long recordings)",
    )
    parser.add_argument(
        "--prefetch_batches",
        type=int,
        default=2,
        help="Loader prefetch queue depth in batches (higher = faster, but more RAM)",
    )
    parser.add_argument(
        "--epochs",
        type=int,
        default=None,
        help=f"Number of epochs (default: {TRAIN_EPOCHS}; {QAT_EPOCHS} with --qat)",
    )
    parser.add_argument(
        "--learning_rate",
        type=float,
        default=None,
        help=(
            f"Initial learning rate (default: {TRAIN_LEARNING_RATE:g}; {QAT_LEARNING_RATE:g} with --qat, "
            f"{PROBE_LEARNING_RATE:g} with --linear_probe)"
        ),
    )
    parser.add_argument("--dropout", type=float, default=0.5, help="Dropout rate before classifier head")
    parser.add_argument("--optimizer", type=str, default="adam", choices=["adam", "sgd", "adamw"], help="Optimizer")
    parser.add_argument("--weight_decay", type=float, default=0.0, help="Weight decay (adamw only)")
    parser.add_argument("--val_split", type=float, default=0.2, help="Validation split ratio")
    parser.add_argument(
        "--checkpoint_path", type=str, default="checkpoints/best_model.keras", help="Output checkpoint path (.keras)"
    )
    parser.add_argument("--grad_clip", type=float, default=1.0, help="Max gradient norm for clipping (0 = disabled)")
    parser.add_argument(
        "--mixed_precision", action="store_true", default=False, help="Enable FP16 mixed precision training"
    )
    parser.add_argument("--resume", action="store_true", default=False, help="Resume training from checkpoint")
    parser.add_argument("--seed", type=int, default=42, help="Random seed for deterministic training")

    parser.add_argument(
        "--validation_overlap",
        type=float,
        default=None,
        help="File validation overlap in seconds (default: half the model chunk duration).",
    )
    parser.add_argument(
        "--validation_pooling",
        choices=["max", "avg", "lme"],
        default="max",
        help="File validation pooling for exact cMAP checkpoint selection.",
    )
    parser.add_argument(
        "--validation_subset",
        type=int,
        default=0,
        help=(
            "Score checkpoint selection (standard training and QAT) on a fixed stratified subset of "
            "this many validation files instead of all of them. Classes are drawn round-robin with "
            "seed 1234, so the subset is as class-balanced as the folders allow and identical across "
            "runs and epochs. Subset cMAP is biased upward and not comparable to full-manifest "
            "numbers. 0 uses all."
        ),
    )

    # -- QAT ------------------------------------------------------------------
    parser.add_argument(
        "--qat",
        action="store_true",
        default=False,
        help="Quantization-aware fine-tuning (requires pretrained --checkpoint_path)",
    )
    parser.add_argument(
        "--qat_calibration_samples",
        type=int,
        default=1024,
        help="Exact stratified samples used for QAT ranges and final INT8 calibration",
    )
    parser.add_argument(
        "--qat_distillation_weight",
        type=float,
        default=1.0,
        help="QAT teacher Bernoulli-KL loss weight",
    )
    parser.add_argument(
        "--qat_cosine_weight",
        type=float,
        default=0.10,
        help="QAT mean teacher/student cosine-loss weight",
    )
    parser.add_argument(
        "--qat_cosine_tail_weight",
        type=float,
        default=0.75,
        help="QAT worst-sample teacher/student cosine-loss weight",
    )
    parser.add_argument(
        "--qat_cosine_tail_fraction",
        type=float,
        default=0.10,
        help="Fraction of each QAT batch included in the worst-sample loss",
    )
    parser.add_argument(
        "--qat_range_refresh",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Recalibrate QAT activation ranges on the current weights after every epoch",
    )
    # -- Linear probing -------------------------------------------------------
    parser.add_argument(
        "--linear_probe",
        action="store_true",
        default=False,
        help="Freeze backbone and train only the classifier head (requires pretrained --checkpoint_path)",
    )

    args = parser.parse_args()

    # Derive positive flags from --no_* flags
    args.spec_augment = not args.no_spec_augment
    args.deterministic = True  # always deterministic

    if args.validation_subset < 0:
        parser.error("--validation_subset must be non-negative")

    # The compression steps run one at a time against a converged checkpoint;
    # the documented order is QAT, then convert.
    exclusive = [name for name in ("qat", "linear_probe") if getattr(args, name)]
    if len(exclusive) > 1:
        raise SystemExit(f"Options are mutually exclusive, run them as separate steps: {exclusive}")

    # Each step has its own best schedule: fine-tuning at the training rate
    # destroys a converged model (measured for QAT at 2e-4).
    if args.epochs is None:
        args.epochs = QAT_EPOCHS if args.qat else TRAIN_EPOCHS
    if args.learning_rate is None:
        args.learning_rate = (
            QAT_LEARNING_RATE if args.qat else PROBE_LEARNING_RATE if args.linear_probe else TRAIN_LEARNING_RATE
        )

    return args

main()

Train a DS-CNN model on a class-structured audio dataset.

Source code in birdnet_stm32/cli/train.py
def main():
    """Train a DS-CNN model on a class-structured audio dataset."""
    args = get_args()
    args.audio_frontend = normalize_frontend_name(args.audio_frontend)

    # Enable dynamic GPU memory growth so we don't reserve all VRAM
    for gpu in tf.config.list_physical_devices("GPU"):
        tf.config.experimental.set_memory_growth(gpu, True)

    # Deterministic mode: seed all RNGs and enable TF deterministic ops
    if args.deterministic:
        seed = args.seed
        random.seed(seed)
        np.random.seed(seed)
        tf.random.set_seed(seed)
        os.environ["TF_DETERMINISTIC_OPS"] = "1"
        os.environ["PYTHONHASHSEED"] = str(seed)
        print(f"Deterministic mode enabled (seed={seed}).")

    # Early warning for STM32N6 raw frontend constraint
    if args.audio_frontend == "raw":
        T = int(args.sample_rate * args.chunk_duration)
        if T >= (1 << 16):
            print(f"[WARN] STM32N6 constraint: raw input length {T} >= 65536.")
            print("       Use --sample_rate 16000 or --chunk_duration 2, or --audio_frontend hybrid.")

    # Mixed precision
    if args.mixed_precision:
        tf.keras.mixed_precision.set_global_policy("mixed_float16")
        print("Mixed precision enabled (float16 compute, float32 accumulation).")

    # Quantization-aware fine-tuning
    if args.qat:
        from birdnet_stm32.training.qat import run_qat

        run_qat(args)
        return

    # Linear probing: freeze backbone, retrain head only
    if args.linear_probe:
        from birdnet_stm32.training.linear_probe import run_linear_probe

        run_linear_probe(args)
        return

    hop_length = compute_hop_length(args.sample_rate, args.chunk_duration, args.spec_width)

    # Load file paths
    top_classes = load_classes_file(args.classes_file) if args.classes_file else None
    if top_classes is not None and args.max_classes is not None:
        raise ValueError("--classes_file and --max_classes are mutually exclusive")
    if args.max_classes is not None:
        top_classes = get_classes_with_most_samples(args.data_path_train, n_classes=args.max_classes)
        print(f"Selected top {len(top_classes)} classes by sample count.")
    file_paths, classes = load_file_paths_from_directory(
        args.data_path_train, classes=top_classes, max_samples=args.max_samples
    )
    if top_classes is not None and classes != top_classes:
        missing = [class_name for class_name in top_classes if class_name not in classes]
        raise ValueError(f"Training dataset is missing configured classes: {missing}")

    # Train/val split
    if args.data_path_val:
        train_paths = file_paths
        val_paths, val_classes = load_file_paths_from_directory(
            args.data_path_val, classes=classes, max_samples=args.max_samples
        )
        if val_classes != classes:
            missing = [class_name for class_name in classes if class_name not in val_classes]
            raise ValueError(f"Validation dataset is missing configured classes: {missing}")
        print("Using separate validation root; --val_split is ignored.")
    else:
        split_idx = int(len(file_paths) * (1 - args.val_split))
        train_paths = file_paths[:split_idx]
        val_paths = file_paths[split_idx:]
    print(f"Training on {len(train_paths)} files, validating on {len(val_paths)} files.")

    # Upsample
    if args.upsample_ratio and 0 < args.upsample_ratio < 1.0:
        train_paths = upsample_minority_classes(train_paths, classes, args.upsample_ratio)
        print(f"After upsampling: {len(train_paths)} training files.")

    # Datasets
    common_kwargs = dict(
        sample_rate=args.sample_rate,
        max_duration=args.max_duration,
        chunk_duration=args.chunk_duration,
        spec_width=args.spec_width,
        mel_bins=args.num_mels,
        fft_length=args.fft_length,
        mag_scale=args.mag_scale,
        input_compression=args.input_compression,
        prefetch_batches=args.prefetch_batches,
    )

    initial_inflight = max(256, args.num_workers * 64)
    common_kwargs["max_inflight_files"] = initial_inflight

    train_loader_control: dict | None = {"max_inflight_files": int(initial_inflight)} if args.num_workers > 0 else None
    extra_callbacks: list[tf.keras.callbacks.Callback] = [HostMemoryGuard()]
    if train_loader_control is not None:
        extra_callbacks.append(
            AdaptiveLoaderTuner(
                control=train_loader_control,
                batch_size=args.batch_size,
                adjust_every=_LOADER_TUNE_ADJUST_EVERY,
                min_inflight=max(64, args.num_workers * 8),
                max_inflight=max(512, args.num_workers * 256),
                target_free_gb=_LOADER_TARGET_FREE_GB,
            )
        )
        print(f"Loader auto-tuning enabled (initial max_inflight_files={initial_inflight}).")

    train_kwargs = dict(common_kwargs)
    if train_loader_control is not None:
        train_kwargs["loader_control"] = train_loader_control

    val_kwargs = dict(common_kwargs)
    train_dataset = load_dataset(
        train_paths,
        classes,
        audio_frontend=args.audio_frontend,
        batch_size=args.batch_size,
        num_workers=args.num_workers,
        max_chunks_per_file=args.max_chunks_per_file,
        mixup_alpha=args.mixup_alpha,
        mixup_probability=args.mixup_probability,
        random_offset=True,
        snr_threshold=0.1,
        spec_augment=args.spec_augment,
        freq_mask_max=args.freq_mask_max,
        time_mask_max=args.time_mask_max,
        **train_kwargs,
    )
    from birdnet_stm32.training.validation import VALIDATION_SUBSET_SEED, FileCmap, stratified_validation_subset

    selection_paths = stratified_validation_subset(val_paths, args.validation_subset)
    if len(selection_paths) < len(val_paths):
        print(
            f"Selecting checkpoints on a fixed stratified subset of {len(selection_paths)} "
            f"of {len(val_paths)} validation files (seed {VALIDATION_SUBSET_SEED})"
        )
    # Keras' own per-epoch validation pass (val_loss and the chunk metrics in the
    # history) reads the same files as selection. Over the full manifest it
    # re-decoded every validation file each epoch for numbers that select nothing.
    val_dataset = load_dataset(
        selection_paths,
        classes,
        audio_frontend=args.audio_frontend,
        batch_size=args.batch_size,
        num_workers=args.num_workers,
        max_chunks_per_file=1,
        mixup_alpha=0.0,
        mixup_probability=0.0,
        random_offset=False,
        snr_threshold=0.5,
        spec_augment=False,
        **val_kwargs,
    )

    steps_per_epoch = max(
        1, math.ceil(estimate_samples_per_epoch(len(train_paths), args.max_chunks_per_file) / float(args.batch_size))
    )
    val_steps = max(1, math.ceil(len(selection_paths) / float(args.batch_size)))

    # Build model
    print("Building model...")
    model = build_dscnn_model(
        num_mels=args.num_mels,
        spec_width=args.spec_width,
        sample_rate=args.sample_rate,
        chunk_duration=args.chunk_duration,
        audio_frontend=args.audio_frontend,
        num_classes=len(classes),
        alpha=args.alpha,
        depth_multiplier=args.depth_multiplier,
        head_pooling=args.head_pooling,
        dw_kernel_size=args.dw_kernel_size,
        stage_widths=args.stage_widths,
        embeddings_size=args.embeddings_size,
        fft_length=args.fft_length,
        mag_scale=args.mag_scale,
        frontend_trainable=args.frontend_trainable,
        dropout_rate=args.dropout,
    )
    # Per-layer MACs and N6 compatibility, rather than a plain Keras summary:
    # on this target the MAC budget and op support decide whether the model is
    # deployable at all.
    print_profile(model)

    # Save model config
    cfg = ModelConfig(
        sample_rate=args.sample_rate,
        num_mels=args.num_mels,
        spec_width=args.spec_width,
        fft_length=args.fft_length,
        chunk_duration=args.chunk_duration,
        hop_length=hop_length,
        audio_frontend=args.audio_frontend,
        mag_scale=args.mag_scale,
        input_compression=args.input_compression,
        embeddings_size=args.embeddings_size,
        alpha=args.alpha,
        depth_multiplier=args.depth_multiplier,
        head_pooling=args.head_pooling,
        dw_kernel_size=args.dw_kernel_size,
        stage_widths=list(args.stage_widths),
        num_classes=len(classes),
        class_names=classes,
        frontend_trainable=args.frontend_trainable,
        dropout_rate=args.dropout,
    )
    cfg_path = os.path.splitext(args.checkpoint_path)[0] + "_model_config.json"
    cfg.save(cfg_path)
    print(f"Saved model config to '{cfg_path}'")

    # Write the labels before training starts. The class list is already fixed
    # here, and the checkpoint callback can leave a usable .keras behind at any
    # epoch, so the contract files must not depend on the run reaching its end:
    # an interrupt or a crash would otherwise strand a checkpoint with no labels.
    labels_file = args.checkpoint_path.replace(".keras", "_labels.txt")
    with open(labels_file, "w") as f:
        for cls in classes:
            f.write(f"{cls}\n")
    print(f"Saved labels to '{labels_file}'")

    extra_callbacks.append(
        FileCmap(
            selection_paths,
            classes,
            cfg.to_dict(),
            overlap=args.validation_overlap,
            pooling=args.validation_pooling,
            batch_size=args.batch_size,
        )
    )

    # Train
    print("Starting training...")
    try:
        train_model(
            model,
            train_dataset,
            val_dataset,
            epochs=args.epochs,
            learning_rate=args.learning_rate,
            batch_size=args.batch_size,
            checkpoint_path=args.checkpoint_path,
            steps_per_epoch=steps_per_epoch,
            val_steps=val_steps,
            optimizer=args.optimizer,
            weight_decay=args.weight_decay,
            gradient_clip_norm=args.grad_clip,
            resume=args.resume,
            extra_callbacks=extra_callbacks,
        )
        print(f"Training complete. Best model saved to '{args.checkpoint_path}'.")
    except KeyboardInterrupt:
        print(f"\nTraining interrupted. Best checkpoint so far: '{args.checkpoint_path}'")