text: #!/usr/bin/env/python3
  """This minimal example trains a CTC-based speech recognizer on a tiny dataset.
  The encoder is based on a combination of convolutional, recurrent, and
  feed-forward networks (CRDNN) that predict phonemes.  A greedy search is used on
  top of the output probabilities.
  Given the tiny dataset, the expected behavior is to overfit the training dataset
  (with a validation performance that stays high).
  """
  
  import pathlib
  import speechbrain as sb
  from hyperpyyaml import load_hyperpyyaml
  
  
  class CTCBrain(sb.Brain):
      def compute_forward(self, batch, stage):
          "Given an input batch it computes the output probabilities."
          wavs, lens = batch.sig
          feats = self.hparams.compute_features(wavs)
          feats = self.modules.mean_var_norm(feats, lens)
          x = self.modules.model(feats)
          x = self.modules.lin(x)
          outputs = self.hparams.softmax(x)
  
          return outputs, lens
  
      def compute_objectives(self, predictions, batch, stage):
          "Given the network predictions and targets computed the CTC loss."
          predictions, lens = predictions
          phns, phn_lens = batch.phn_encoded
          loss = self.hparams.compute_cost(predictions, phns, lens, phn_lens)
  
          if stage != sb.Stage.TRAIN:
              seq = sb.decoders.ctc_greedy_decode(
                  predictions, lens, blank_id=self.hparams.blank_index
              )
              self.per_metrics.append(batch.id, seq, phns, target_len=phn_lens)
  
          return loss
  
      def on_stage_start(self, stage, epoch=None):
          "Gets called when a stage (either training, validation, test) starts."
          if stage != sb.Stage.TRAIN:
              self.per_metrics = self.hparams.per_stats()
  
      def on_stage_end(self, stage, stage_loss, epoch=None):
          """Gets called at the end of a stage."""
          if stage == sb.Stage.TRAIN:
              self.train_loss = stage_loss
  
          if stage == sb.Stage.VALID and epoch is not None:
              print("Epoch %d complete" % epoch)
              print("Train loss: %.2f" % self.train_loss)
  
          if stage != sb.Stage.TRAIN:
              print(stage, "loss: %.2f" % stage_loss)
              print(stage, "PER: %.2f" % self.per_metrics.summarize("error_rate"))
  
  
  def data_prep(data_folder, hparams):
      "Creates the datasets and their data processing pipelines."
  
      # 1. Declarations:
      train_data = sb.dataio.dataset.DynamicItemDataset.from_json(
          json_path=data_folder / "train.json",
          replacements={"data_root": data_folder},
      )
      valid_data = sb.dataio.dataset.DynamicItemDataset.from_json(
          json_path=data_folder / "dev.json",
          replacements={"data_root": data_folder},
      )
      datasets = [train_data, valid_data]
      label_encoder = sb.dataio.encoder.CTCTextEncoder()
  
      # 2. Define audio pipeline:
      @sb.utils.data_pipeline.takes("wav")
      @sb.utils.data_pipeline.provides("sig")
      def audio_pipeline(wav):
          sig = sb.dataio.dataio.read_audio(wav)
          return sig
  
      sb.dataio.dataset.add_dynamic_item(datasets, audio_pipeline)
  
      # 3. Define text pipeline:
      @sb.utils.data_pipeline.takes("phn")
      @sb.utils.data_pipeline.provides("phn_list", "phn_encoded")
      def text_pipeline(phn):
          phn_list = phn.strip().split()
          yield phn_list
          phn_encoded = label_encoder.encode_sequence_torch(phn_list)
          yield phn_encoded
  
      sb.dataio.dataset.add_dynamic_item(datasets, text_pipeline)
  
      # 3. Fit encoder:
      # NOTE: In this minimal example, also update from valid data
      label_encoder.insert_blank(index=hparams["blank_index"])
      label_encoder.update_from_didataset(train_data, output_key="phn_list")
      label_encoder.update_from_didataset(valid_data, output_key="phn_list")
  
      # 4. Set output:
      sb.dataio.dataset.set_output_keys(datasets, ["id", "sig", "phn_encoded"])
  
      return train_data, valid_data
  
  
  def main():
      experiment_dir = pathlib.Path(__file__).resolve().parent
      hparams_file = experiment_dir / "hyperparams_complex_net.yaml"
      data_folder = "../../../../samples/audio_samples/nn_training_samples"
      data_folder = (experiment_dir / data_folder).resolve()
  
      # Load model hyper parameters:
      with open(hparams_file) as fin:
          hparams = load_hyperpyyaml(fin)
  
      # Dataset creation
      train_data, valid_data = data_prep(data_folder, hparams)
  
      # Trainer initialization
      ctc_brain = CTCBrain(hparams["modules"], hparams["opt_class"], hparams)
  
      # Training/validation loop
      ctc_brain.fit(
          range(hparams["N_epochs"]),
          train_data,
          valid_data,
          train_loader_kwargs=hparams["dataloader_options"],
          valid_loader_kwargs=hparams["dataloader_options"],
      )
      # Evaluation is run separately (now just evaluating on valid data)
      ctc_brain.evaluate(valid_data)
  
      # Check if model overfits for integration test
      assert ctc_brain.train_loss < 1.0
  
  
  if __name__ == "__main__":
      main()
  
  
  def test_error():
      main()
  