code: def prepare_mixer6(
      corpus_dir: Pathlike,
      output_dir: Optional[Pathlike] = None,
      dset_part="dev",
      mic: Optional[str] = "mdm",
      normalize_text: Optional[str] = "chime6",
      json_dir: Optional[
          Pathlike
      ] = None,  # alternative annotation e.g. from non-oracle diarization
      ignore_shorter: Optional[float] = 0.2,
  ) -> Dict[str, Dict[str, Union[RecordingSet, SupervisionSet]]]:
      assert mic in ["ihm", "mdm"], "mic must be one of 'ihm' or 'mdm'"
      if mic == "ihm":
          assert dset_part in [
              "train_intv",
              "train_call",
              "dev",
          ], "No close-talk microphones on evaluation set."
  
      normalize_text_func = choose_txt_normalization(normalize_text)
      transcriptions_dir = (
          os.path.join(corpus_dir, "transcriptions_scoring")
          if json_dir is None
          else json_dir
      )
      corpus_dir = Path(corpus_dir)
      assert corpus_dir.is_dir(), f"No such directory: {corpus_dir}"
      manifests = defaultdict(dict)
      if output_dir is not None:
          output_dir = Path(output_dir)
          output_dir.mkdir(parents=True, exist_ok=True)
  
      all_sessions = glob.glob(
          os.path.join(transcriptions_dir, dset_part, "*.json")
      )
      all_sessions = [Path(x).stem for x in all_sessions]
      audio_files = glob.glob(
          os.path.join(corpus_dir, "audio", dset_part, "*.flac")
      )
      assert (
          len(audio_files) > 0
      ), "Can't parse mixer6 audio files, is the path correct ?"
      sess2audio = {}
      for audio_f in audio_files:
          sess_name = "_".join(Path(audio_f).stem.split("_")[:-1])
          if sess_name not in sess2audio.keys():
              sess2audio[sess_name] = [audio_f]
          else:
              sess2audio[sess_name].append(audio_f)
  
      recordings = []
      supervisions = []
      for sess in all_sessions:
          with open(
              os.path.join(
                  transcriptions_dir, dset_part, f"{sess}.json"
              )
          ) as f:
              transcript = json.load(f)
          if mic == "ihm":
              if dset_part.startswith("train"):
                  if mic == "ihm" and dset_part.startswith("train"):
                      current_sess_audio = [
                          x
                          for x in sess2audio[sess]
                          if Path(x).stem.split("_")[-1] in ["CH02"]
                      ]  # only interview and call
  
              elif dset_part == "dev":
                  current_sess_audio = [
                      x
                      for x in sess2audio[sess]
                      if Path(x).stem.split("_")[-1] in ["CH02", "CH01"]
                  ]  #
              else:
                  raise NotImplementedError(
                      "No close-talk mics for eval set"
                  )
  
          elif mic == "mdm":
              current_sess_audio = [
                  x
                  for x in sess2audio[sess]
                  if Path(x).stem.split("_")[-1]
                  not in ["CH01", "CH02", "CH03"]
              ]
          else:
              raise NotImplementedError
  
          # recordings here
          sources = [
              AudioSource(
                  type="file", channels=[idx], source=str(audio_path)
              )
              for idx, audio_path in enumerate(current_sess_audio)
          ]
          audio_sf = sf.SoundFile(str(current_sess_audio[0]))
          recordings.append(
              Recording(
                  id=f"{sess}-{dset_part}-{mic}",
                  sources=sources,
                  sampling_rate=int(audio_sf.samplerate),
                  num_samples=audio_sf.frames,
                  duration=audio_sf.frames / audio_sf.samplerate,
              )
          )
  
          for idx, segment in enumerate(transcript):
              spk_id = segment["speaker"]
              start = float(segment["start_time"])
              end = float(segment["end_time"])
              if (
                  ignore_shorter is not None
                  and (end - start) < ignore_shorter
              ):
                  print(
                      "Ignored segment session {} speaker"
                      " {} seconds {} to {}, because shorter than {}"
                      "".format(
                          sess, spk_id, start, end, ignore_shorter
                      )
                  )
                  continue
  
              if mic == "ihm":  # and dset_part.startswith("train"):
                  rec_id = f"{sess}-{dset_part}-{mic}"
                  if mic == "ihm" and dset_part == "dev":
                      subject_id = sess.split("_")[-1]
                      if spk_id == subject_id:
                          channel = 0
                      else:
                          channel = 1
                  else:
                      channel = 0
              else:
                  rec_id = f"{sess}-{dset_part}-{mic}"
                  channel = list(range(len(current_sess_audio)))
  
              ex_id = (
                  f"{spk_id}_mixer6_{sess}_{dset_part}_{idx}-"
                  f"{round(100 * start):06d}_{round(100 * end):06d}-{mic}"
              )
              if "words" not in segment.keys():
                  assert json_dir is not None
                  segment["words"] = "placeholder"
              supervisions.append(
                  SupervisionSegment(
                      id=ex_id,
                      recording_id=rec_id,
                      start=start,
                      duration=add_durations(
                          end, -start, sampling_rate=16000
                      ),
                      channel=channel,
                      text=normalize_text_func(segment["words"]),
                      speaker=spk_id,
                  )
              )
  
      recording_set, supervision_set = fix_manifests(
          RecordingSet.from_recordings(recordings),
          SupervisionSet.from_segments(supervisions),
      )
      # Fix manifests
      validate_recordings_and_supervisions(
          recording_set, supervision_set
      )
  
      if output_dir is not None:
          supervision_set.to_file(
              os.path.join(
                  output_dir,
                  f"mixer6-{mic}_supervisions_{dset_part}.jsonl.gz",
              )
          )
          recording_set.to_file(
              os.path.join(
                  output_dir,
                  f"mixer6-{mic}_recordings_{dset_part}.jsonl.gz",
              )
          )
  
      manifests[dset_part] = {
          "recordings": recording_set,
          "supervisions": supervision_set,
      }
      return manifests
  
file_path: espnet/egs2/chime7_task1/asr1/local/get_lhotse_manifests.py
imports: import argparse
  import glob
  import json
  import os.path
  from collections import defaultdict
  from pathlib import Path
  from typing import Dict, Optional, Union
  import lhotse
  import soundfile
  from gen_task1_data import choose_txt_normalization
  from lhotse import fix_manifests, validate_recordings_and_supervisions
  from lhotse.audio import AudioSource, Recording, RecordingSet
  from lhotse.supervision import SupervisionSegment, SupervisionSet
  from lhotse.utils import Pathlike, add_durations, safe_extract, urlretrieve_progress
label_desc: Use image node_id 3 for calling a global function with example usage: prepare_mixer6(corpus_dir, output_dir, dset_part, mic, normalize_text, json_dir, ignore_shorter) and returns: manifests