windlass.providers.loaders.media

media

Image (OCR) and audio (transcription) loaders.

Both turn non-text media into text so it can be chunked, embedded and retrieved exactly like a document.

Install with::

pip install "windlass[ocr]"     # images — also needs the Tesseract binary
pip install "windlass[audio]"   # audio

Example

from windlass import Windlass # doctest: +SKIP Windlass.loader("image").load("./scans") # doctest: +SKIP Windlass.loader("audio").load("./meeting.mp3") # doctest: +SKIP

ImageLoader

ImageLoader(
    *,
    language: str = "eng",
    psm: int = 3,
    preprocess: bool = True,
    min_chars: int = 5,
    **config: Any
)

Bases: Loader

Extracts text from images using Tesseract.

Parameters:

Name	Type	Description	Default
language	str	Tesseract language code(s), e.g. "eng" or "eng+deu".	'eng'
psm	int	Tesseract page segmentation mode. 3 (fully automatic) suits documents; 6 (assume a uniform block) suits screenshots.	3
preprocess	bool	Convert to greyscale and auto-contrast before OCR, which measurably improves accuracy on photographs and low-contrast scans.	True
min_chars	int	Return no document when fewer characters were recognised — below this it is noise, not text.	5
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Raises:

Type	Description
MissingDependencyError	When pytesseract or pillow is missing.
IngestionError	When the Tesseract binary is not installed, or the image cannot be read.

Note

pytesseract is a wrapper, not an OCR engine. The Tesseract binary must be installed separately (apt install tesseract-ocr, brew install tesseract, or the Windows installer). The error message says so if it is missing.

Source code in src\windlass\providers\loaders\media.py

def __init__(
    self,
    *,
    language: str = "eng",
    psm: int = 3,
    preprocess: bool = True,
    min_chars: int = 5,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.language = language
    self.psm = psm
    self.preprocess = preprocess
    self.min_chars = min_chars

aload_source async

aload_source(source: SourceLike) -> list[Document]

OCR one image.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path or a bytes payload.	required

Returns:

Type	Description
list[Document]	A single-element list, or empty when too little text was recognised.

Raises:

Type	Description
IngestionError	When OCR fails or Tesseract is unavailable.

Source code in src\windlass\providers\loaders\media.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """OCR one image.

    Args:
        source: A path or a bytes payload.

    Returns:
        A single-element list, or empty when too little text was recognised.

    Raises:
        IngestionError: When OCR fails or Tesseract is unavailable.
    """
    pytesseract = require("pytesseract", extra="ocr", feature="The image OCR loader")
    pil = require("PIL.Image", extra="ocr", feature="The image OCR loader")
    raw = self._read_bytes(source)
    path = str(source) if not isinstance(source, bytes) else None
    base = self._base_metadata(source) if not isinstance(source, bytes) else {}

    def _ocr() -> tuple[str, dict[str, Any]]:
        try:
            image = pil.open(io.BytesIO(raw))
            meta = {"width": image.width, "height": image.height, "format": image.format}
            if self.preprocess:
                image = _prepare(image)
            text = pytesseract.image_to_string(
                image, lang=self.language, config=f"--psm {self.psm}"
            )
        except Exception as exc:
            if "tesseract" in str(exc).lower():
                raise IngestionError(
                    "The Tesseract OCR binary is not installed.",
                    hint="Install it: apt install tesseract-ocr | "
                    "brew install tesseract | choco install tesseract",
                ) from exc
            raise IngestionError(f"OCR failed for {path or 'the image'}: {exc}") from exc
        return text, meta

    text, meta = await to_thread(_ocr)
    cleaned = normalize_whitespace(text)
    if len(cleaned) < self.min_chars:
        self._log.debug("OCR produced only %d characters for %s", len(cleaned), path)
        return []
    return [
        Document(
            content=cleaned,
            metadata={**base, **meta, "ocr_language": self.language},
            source=path,
            mimetype="text/plain",
        )
    ]

AudioLoader

AudioLoader(
    *,
    model: str = "base",
    language: str | None = None,
    device: str = "auto",
    compute_type: str = "int8",
    per_segment: bool = False,
    vad_filter: bool = True,
    **config: Any
)

Bases: Loader

Transcribes audio using faster-whisper.

Parameters:

Name	Type	Description	Default
model	str	Whisper model size — tiny, base, small, medium, large-v3. Bigger is more accurate and slower.	'base'
language	str | None	Force a language code, or None to auto-detect.	None
device	str	"cpu", "cuda", or "auto".	'auto'
compute_type	str	Quantisation, e.g. "int8" on CPU or "float16" on GPU.	'int8'
per_segment	bool	One document per transcript segment, each carrying its timestamp — the right choice for anything you want to cite.	False
vad_filter	bool	Skip silence with voice-activity detection, which speeds up transcription of sparse recordings considerably.	True
**config	Any	Forwarded to :class:~windlass.interfaces.loader.Loader.	{}

Raises:

Type	Description
MissingDependencyError	When faster-whisper is not installed.
IngestionError	When the audio cannot be transcribed.

Performance

Transcription is slow and CPU-heavy. It runs on a worker thread so it never blocks the event loop, but expect roughly real-time on CPU with the base model.

Source code in src\windlass\providers\loaders\media.py

def __init__(
    self,
    *,
    model: str = "base",
    language: str | None = None,
    device: str = "auto",
    compute_type: str = "int8",
    per_segment: bool = False,
    vad_filter: bool = True,
    **config: Any,
) -> None:
    super().__init__(**config)
    self.model_size = model
    self.language = language
    self.device = device
    self.compute_type = compute_type
    self.per_segment = per_segment
    self.vad_filter = vad_filter

aload_source async

aload_source(source: SourceLike) -> list[Document]

Transcribe one audio file.

Parameters:

Name	Type	Description	Default
source	SourceLike	A path. Byte payloads are written to a temporary file first, because the decoder needs a seekable path.	required

Returns:

Type	Description
list[Document]	One document per segment, or one for the whole transcript.

Raises:

Type	Description
IngestionError	When transcription fails.

Source code in src\windlass\providers\loaders\media.py

async def aload_source(self, source: SourceLike) -> list[Document]:
    """Transcribe one audio file.

    Args:
        source: A path. Byte payloads are written to a temporary file first,
            because the decoder needs a seekable path.

    Returns:
        One document per segment, or one for the whole transcript.

    Raises:
        IngestionError: When transcription fails.
    """
    path = str(source) if not isinstance(source, bytes) else None
    base = self._base_metadata(source) if not isinstance(source, bytes) else {}

    if path is None:
        import tempfile

        with tempfile.NamedTemporaryFile(suffix=".audio", delete=False) as handle:
            handle.write(source)  # type: ignore[arg-type]
            path = handle.name

    def _transcribe() -> tuple[list[Any], Any]:
        try:
            segments, info = self._model().transcribe(
                path,
                language=self.language,
                vad_filter=self.vad_filter,
                beam_size=5,
            )
            return list(segments), info
        except Exception as exc:
            raise IngestionError(
                f"Could not transcribe {path}: {exc}",
                hint="Check the file is valid audio and that ffmpeg is installed.",
            ) from exc

    segments, info = await to_thread(_transcribe)
    meta = {
        **base,
        "language": getattr(info, "language", self.language or "unknown"),
        "duration_seconds": round(getattr(info, "duration", 0.0)),
        "whisper_model": self.model_size,
    }

    if self.per_segment:
        documents = []
        for segment in segments:
            text = (segment.text or "").strip()
            if not text:
                continue
            start = float(segment.start)
            documents.append(
                Document(
                    content=text,
                    metadata={
                        **meta,
                        "start_seconds": round(start, 1),
                        "end_seconds": round(float(segment.end), 1),
                        "timestamp": f"{int(start) // 60:02d}:{int(start) % 60:02d}",
                    },
                    source=path,
                    mimetype="text/plain",
                )
            )
        return documents

    text = normalize_whitespace(" ".join((s.text or "").strip() for s in segments))
    if not text:
        return []
    return [Document(content=text, metadata=meta, source=path, mimetype="text/plain")]