code: def write_on_batch_end(
      self,
      trainer,
      pl_module: "LightningModule",
      prediction: Tuple[int, List[FrameCtmUnit]],
      batch_indices: List[int],
      batch: Any,
      batch_idx: int,
      dataloader_idx: int,
  ):
      for sample_id, units in prediction:
          sample = self.dataset.get_manifest_sample(sample_id)
          with_ctm = True
          if len(units) == 0:
              logging.warning(
                  f"""Do not producing CTM output for item `{sample.audio_file}`.
                      Check if text is empty or if duration is too short: `{sample.text_raw}`, {sample.duration}"""
              )
              with_ctm = False
          item = {}
          item["audio_filepath"] = sample.audio_file
          item["duration"] = sample.duration
          item["text"] = sample.text_raw
          if with_ctm:
              utt_name = Path(sample.audio_file).stem
              ctm_filepath = (
                  os.path.join(self.output_ctm_dir, utt_name) + ".ctm"
              )
              self.write_ctm(utt_name, ctm_filepath, units)
              item["ctm_filepath"] = ctm_filepath
          else:
              item["ctm_filepath"] = ""
          self.outf.write(json.dumps(item) + "\n")
          self.samples_num += 1
      return
  
file_path: NeMo/nemo/collections/asr/data/audio_to_ctm_dataset.py
imports: import json
  import os
  from dataclasses import dataclass
  from pathlib import Path
  from typing import Any, List, Tuple
  from nemo.collections.asr.data.audio_to_text_dataset import ASRPredictionWriter
  from nemo.utils import logging
label_desc: Use image node_id 3 for calling the ASRCTMPredictionWriter obj's underlying member method code with example usage: obj.write_on_batch_end(trainer, pl_module, prediction, batch_indices, batch, batch_idx, dataloader_idx) without return types