chore(caul upgrade): upgrade caul to 0.10.3, caul-core to 0.4.0

workers/asr-worker/asr_worker/activities.py

@@ -9,15 +9,14 @@
 
 from aiofile import async_open
 from caul_core import (
-    ASRResult,
     InferenceRunner,
     InferenceRunnerConfig,
     Postprocessor,
     PostprocessorConfig,
-    PreprocessedInput,
     Preprocessor,
     PreprocessorConfig,
 )
+from caul_core.objects import ASRResult, PreprocessedInput
 from datashare_python.dependencies import lifespan_es_client, lifespan_worker_config
 from datashare_python.objects import DocRoute, Document
 from datashare_python.types_ import (
@@ -428,6 +427,14 @@ def _preprocess(
         logger.debug("writing batch to %s", batch_file)
         with batch_file.open("w") as f:
             for processed in batch:
+                if processed.metadata.error is not None:
+                    logger.error(
+                        "PreprocessedInput '%s' was not properly decoded with"
+                        "error '%s'. Skipping.",
+                        processed.metadata.input_file_path,
+                        processed.metadata.error,
+                    )
+                    continue
                 f.write(processed.model_dump_json() + "\n")
         yield batch_file
 

workers/asr-worker/asr_worker/objects.py

@@ -1,9 +1,10 @@
 import math
 from collections import defaultdict
 from functools import cache
-from typing import Any, ClassVar, Self
+from typing import Annotated, Any, ClassVar, Self
 
 from caul_core import ASRPipelineConfig, ASRResult
+from caul_core.config import BaseInferenceRunnerConfig
 from caul_core.objects import ASRLanguage, ASRModel
 from datashare_python.objects import (
     ArtifactType,
@@ -12,7 +13,16 @@
     ManifestEntry,
     TaskArgs,
 )
-from pydantic import Field, RootModel
+from icij_common.pydantic_utils import make_enum_discriminator, tagged_union
+from pydantic import Discriminator, Field, RootModel
+
+model_discriminator = make_enum_discriminator("model", ASRModel)
+InferenceRunnerConfig = Annotated[
+    tagged_union(
+        BaseInferenceRunnerConfig.__subclasses__(), lambda t: t.model.default.value
+    ),
+    Discriminator(model_discriminator),
+]
 
 DocumentSearchQuery = dict[str, Any]
 DocId = str

workers/asr-worker/pyproject.toml

@@ -11,7 +11,7 @@ readme = "README.md"
 requires-python = ">=3.11.0, <3.13"
 dependencies = [
   "datashare-python~=0.10.0",
-  "caul-core==0.3.1",
+  "caul-core==0.4.0",
 ]
 
 [project.scripts]
@@ -30,7 +30,7 @@ gpu = [
   "torchcodec==0.10.0+cu129; sys_platform == 'linux'",
 ]
 inference = [
-  "caul[nemo]==0.10.2",
+  "caul[nemo]==0.10.3",
   "kaldialign==0.9.3",
   "ml-dtypes==0.5.4",
   "numpy==2.3.0",

workers/asr-worker/tests/test_activities.py

@@ -64,6 +64,14 @@
         preprocessed_file_path=Path("preprocessed_2.wav"),
     )
 )
+PREPROCESSED_INPUT_ERROR = PreprocessedInput(
+    metadata=InputMetadata(
+        input_ordering=0,
+        duration_s=0.0,
+        preprocessed_file_path=Path("preprocessed_error.wav"),
+        error="failed to decode audio",
+    )
+)
 
 INFERENCE_RESULTS = [
     ASRResult(
@@ -121,6 +129,28 @@ def process(
             yield list(b)
 
 
+class MockErroringPreprocessor(Preprocessor):
+    def __init__(self, batch_size: int) -> None:
+        self._batch_size = batch_size
+
+    @classmethod
+    def cache_models(cls, cache_dir: Path | None = None) -> None: ...
+
+    @classmethod
+    def _from_config(cls, config: RegistrableConfig, **kwargs) -> Self:  # noqa: ARG003
+        return cls(**kwargs)
+
+    def process(
+        self,
+        audios: Iterable[Path],  # noqa: ARG002
+        **kwargs,  # noqa: ARG002
+    ) -> Iterable[list[PreprocessedInput]]:
+        outputs = cycle([PREPROCESSED_INPUT_ERROR, PREPROCESSED_INPUT_1])
+        outputs = [next(outputs) for _ in audios]
+        for b in batches(outputs, self._batch_size):
+            yield list(b)
+
+
 class MockInferenceRunner(InferenceRunner):
     @classmethod
     def _from_config(cls, config: RegistrableConfig, **kwargs) -> Self:  # noqa: ARG003
@@ -251,6 +281,50 @@ def test_preprocess_act(test_worker_config: ASRWorkerConfig, tmpdir: Path) -> No
     assert written_batches == expected_batches
 
 
+def test_preprocess_act_skips_input_with_error(
+    test_worker_config: ASRWorkerConfig, tmpdir: Path
+) -> None:
+    # Given
+    output_dir = Path(tmpdir)
+    n_audios = 2
+    batch_size = n_audios
+    audio_batch = tmpdir / "audio_batch.txt"
+    batch = [
+        Document(
+            id=f"doc-{i}",
+            language=DatashareLanguage("ENGLISH"),
+            path=Path(str(i)),
+            root_document=f"root-{i}",
+            index=TEST_PROJECT,
+            metadata={"tika_metadata_resourcename": f"doc-{i}.wav"},
+        )
+        for i in range(n_audios)
+    ]
+    with audio_batch.open("w") as f:
+        for fs_doc in batch:
+            f.write(fs_doc.model_dump_json() + "\n")
+    preprocessor = MockErroringPreprocessor(batch_size=batch_size)
+
+    # When
+    batch_files = preprocess_act(
+        preprocessor,
+        audio_batch=audio_batch,
+        worker_config=test_worker_config,
+        output_dir=output_dir,
+    )
+
+    # Then
+    assert len(batch_files) == 1
+    written_batches = [
+        [
+            PreprocessedInput.model_validate(d)
+            for d in read_jsonl_as(output_dir / f, PreprocessedInput)
+        ]
+        for f in batch_files
+    ]
+    assert written_batches == [[PREPROCESSED_INPUT_1]]
+
+
 async def test_infer_act(tmpdir: Path) -> None:
     # Given
     inference_runner = MockInferenceRunner()