| 299 | counter = functools.partial(beam.metrics.Metrics.counter, _BEAM_NAMESPACE) |
| 300 | |
| 301 | def _process_join(el): |
| 302 | try: |
| 303 | (_, values) = el |
| 304 | examples = values["examples"] |
| 305 | if len(examples) != 1: |
| 306 | counter(f"{len(examples)}_examples").inc() |
| 307 | return |
| 308 | indexed_example = examples[0] |
| 309 | |
| 310 | clips = values["clips"] |
| 311 | if len(clips) != 1: |
| 312 | counter("{len(clips)}_clips").inc() |
| 313 | return |
| 314 | clip = clips[0] |
| 315 | |
| 316 | except ValueError as e: |
| 317 | raise ValueError(f"Unpacking element {el} failed") from e |
| 318 | example = indexed_example.example.copy() |
| 319 | example["voice"] = clip |
| 320 | yield indexed_example.index, example |
| 321 | |
| 322 | examples_pipeline = pipeline | "Collection of examples" >> beam.Create( |
| 323 | [[ex.clip_filename, ex] for ex in examples] |