Skip to content

JoshTalks/Signaljt

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

8 Commits
.github/workflows
.github/workflows
 
 
signaljt
signaljt
 
 
.gitignore
.gitignore
 
 
API.md
API.md
 
 
LICENSE
LICENSE
 
 
README.md
README.md
 
 
pyproject.toml
pyproject.toml
 
 
release.py
release.py
 
 
uv.lock
uv.lock
 
 

Repository files navigation

signaljt

Python client library for the Signal validation set — Signal, by Josh Talks.

signaljt gives you two things through one small, dependency-light package:

  1. Datasets — download the ground-truth evaluation audio for any of 15 Indic languages, with resumable parallel downloads, integrity checks, a shared cache, and ready-to-use absolute file paths.
  2. Scoring — submit your ASR model's predictions and get OI-WER back: overall, per-language, and per-utterance, as a dot-accessible object that drops straight into wandb / TensorBoard.

Everything is available both as a Python API and a signaljt CLI.

Not using Python? The three underlying REST endpoints are documented separately in API.md, with curl examples you can use from any language.

Table of contents

Requirements

  • Python 3.8+
  • Runtime dependencies: httpx, filelock (installed automatically)
  • Optional: ipywidgets (for the notebook login widget)

Installation

pip install signaljt

Notebook login widget:

pip install "signaljt[notebook]"

From source (development):

git clone <repo> && cd meta_validation_package
pip install -e .            # or: pip install -e ".[notebook,dev]"

Getting started (5 minutes)

import signaljt

# 1. Authenticate once (or set the SIGNALJT_API_KEY env var)
signaljt.login("your-api-key")

# 2. Download a language's ground-truth audio
ds = signaljt.fetch_dataset("hi")           # Hindi
print(len(ds), "utterances")
print(ds[0].index, ds[0].audio_path)        # absolute path, ready to load

# 3. Run YOUR model over the audio -> {index: transcript}
predictions = {item.index: my_asr_model(item.audio_path) for item in ds}

# 4. Score it — returns overall + per-language + per-chunk OI-WER
result = signaljt.evaluate(predictions, eval_name="my-model-v1")
print("OI-WER:", result.oiwer)              # e.g. 0.1033
print("Hindi:", result.per_language["hi"].oiwer)

# 5. (optional) log to your experiment tracker
import wandb; wandb.log(result.metrics())

The same flow on the command line:

signaljt login
signaljt download hi
# ... produce predictions.json ...
signaljt eval predictions.json              # prints the signed result URL

Authentication

Signal authenticates every request with a long-lived API key (X-API-Key). signaljt stores it once and resolves it automatically for every call.

Resolution order (highest priority first):

  1. An explicit api_key=... argument to any function
  2. The SIGNALJT_API_KEY environment variable
  3. The saved token file at ~/.cache/signaljt/token

Log in

Three interchangeable ways — use whichever fits your context:

# CLI (interactive, hidden prompt) — best on a terminal / HPC login node
signaljt login
# In a script
import signaljt
signaljt.login("your-api-key")       # or signaljt.login() to be prompted
# In a Jupyter / Colab notebook — masked widget, never echoed into the notebook
from signaljt import notebook_login
notebook_login()
# Non-interactive (CI, Slurm batch) — just set the env var, no login needed
export SIGNALJT_API_KEY="your-api-key"

The token is written to ~/.cache/signaljt/token with 0600 permissions.

Check and log out

signaljt whoami      # validates the saved key against the server
signaljt logout      # deletes the saved token
signaljt.whoami()    # raises AuthenticationError if the key is invalid
signaljt.logout()

Note: API keys have an expiry (1h / 1d / 2d / 7d / never, set on the dashboard). An expired key returns a clear AuthenticationError — just log in again with a fresh key.

Downloading datasets

fetch_dataset fetches the signed URL, downloads the zip, verifies it, extracts it, and returns a friendly object with absolute audio paths. It's cached, so the second call (from any process or node) is instant.

ds = signaljt.fetch_dataset("bn")     # Bengali

Supported language codes (via signaljt.list_languages()):

Code Language Code Language Code Language
hi Hindi bn Bengali ta Tamil
te Telugu mr Marathi gu Gujarati
kn Kannada ml Malayalam pa Punjabi
as Assamese or Odia mai Maithili
bho Bhojpuri hne Chattisgarhi ur Urdu

One language, many, or all

The first argument accepts a single code, a list, or the string "all":

ds   = signaljt.fetch_dataset("hi")           # -> a single Dataset
data = signaljt.fetch_dataset(["hi", "bn"])   # -> {"hi": Dataset, "bn": Dataset}
data = signaljt.fetch_dataset("all")          # -> all 15 languages, as a dict

When you request multiple languages they download concurrently (bounded by max_parallel, default 4); each language keeps its own cache/lock/resume, so a partial run resumes cleanly.

The Dataset object

ds = signaljt.fetch_dataset("bn")

len(ds)                    # number of utterances
ds.language_code           # "bn"
ds.root                    # directory containing the extracted files

# Iterate
for item in ds:
    item.index             # utterance id (str), from the manifest
    item.audio_path        # ABSOLUTE path to the .flac, ready to load
    item.extra             # dict of any extra manifest columns

# Index access
ds[0]                      # by position -> DatasetItem
ds["100825"]               # by Index    -> DatasetItem
"100825" in ds             # membership test
ds.get("999", default=None)

# Bulk helpers
ds.indices                 # ["100825", "100826", ...]
ds.paths()                 # {"100825": "/abs/.../100825.flac", ...}
ds.items                   # list[DatasetItem]

The paths are computed at load time from wherever the data actually lives, so they stay correct even if you move or copy the folder — always go through the Dataset object rather than hand-parsing the manifest file.

Placing files where you want (local_dir)

By default the data lives in the cache. To also materialize it into a directory you choose (like Hugging Face's local_dir), without re-downloading:

ds = signaljt.fetch_dataset("bn", local_dir="./bn_data")
ds.root            # ./bn_data
ds[0].audio_path   # absolute path inside ./bn_data

# Materialize an already-fetched dataset afterwards:
signaljt.fetch_dataset("bn").export("./bn_data", mode="copy")

# For multiple languages, each lands under local_dir/<code>/
signaljt.fetch_dataset("all", local_dir="./data")   # ./data/hi, ./data/bn, ...

Placement modes (local_dir_mode=):

Mode Behavior Use when
auto (default) hardlink if on the same filesystem (instant, no extra disk); copy across filesystems almost always
copy a real independent copy you'll move / rsync / edit it
symlink symlink back into the cache smallest footprint; breaks if the cache is cleared
hardlink force a hardlink (errors across filesystems) you want to guarantee no copy

Caching, refresh, and force

The dataset is cached under ~/.cache/signaljt/datasets/<code> (override with cache_dir= or SIGNALJT_CACHE_DIR). Subsequent calls are instant and do no network I/O.

signaljt.fetch_dataset("bn")                 # instant if cached
signaljt.fetch_dataset("bn", refresh=True)   # one cheap probe; re-download only if the remote zip changed
signaljt.fetch_dataset("bn", force=True)     # always re-download from scratch
  • Integrity: every download is verified against the server's MD5 (x-goog-hash) when present, size otherwise.
  • Versioning: the cached zip's identity is recorded; refresh=True detects a re-published dataset and refreshes automatically.
  • Resume: a killed download continues where it stopped (segment-granular), which makes it safe under Slurm preemption / time limits.

Performance tuning

Every knob is a per-call parameter. Defaults are conservative (single-threaded), which is fastest on low-latency links; raise them for large files or high-latency / on-prem → cloud transfers.

Parameter Default What it does
connections 1 parallel ranged download streams per file
extract_workers 1 threads used to unzip
chunk_size 1 MiB streaming buffer size
min_segment 8 MiB don't split a download below this
max_segment 64 MiB cap per-segment size (bounds resume loss)
retries 4 per-segment retry attempts
max_parallel 4 languages downloaded concurrently (multi-language calls) 
signaljt.fetch_dataset(
    "bn",
    connections=8,        # parallelize the download
    extract_workers=4,    # parallelize the unzip
    chunk_size=4 * 1024 * 1024,
    max_segment=32 * 1024 * 1024,
    retries=6,
)

Tip: on a machine co-located with the storage (same-region cloud) a single connection already saturates the link, so the defaults are ideal. On a cluster pulling over the public internet, connections=8 is usually a solid win — benchmark once on your actual nodes.

CLI: download / languages

signaljt languages                      # list the 15 supported codes

signaljt download hi                    # one language
signaljt download hi bn te              # several
signaljt download all                   # everything

# Options
signaljt download bn \
  --dir ./cache \                       # cache directory
  --local-dir ./bn_data --mode auto \   # also place files here
  --connections 8 --extract-workers 4 \
  --parallel 4 \                        # languages at once (for multi/all)
  --chunk-size 4M --min-segment 8M --max-segment 32M --retries 6 \
  --force                               # or --refresh

Size flags accept K/M/G suffixes (e.g. --chunk-size 4M).

Evaluating (OI-WER scoring)

Scoring is an asynchronous job on the server: you submit predictions, it scores them against the ground truth, and you fetch the result. evaluate() does the whole thing synchronously and hands you a parsed result object.

result = signaljt.evaluate("predictions.json", eval_name="conformer-ckpt3800")
print(result.oiwer)                 # weighted overall OI-WER

evaluate() blocks until scoring finishes and returns silently (no console output). Pass progress=True to print poll status to stderr.

Prediction input formats

predictions (or predictions_url=) can be any of:

signaljt.evaluate("preds.json")                       # JSON file path
signaljt.evaluate("preds.jsonl")                      # JSONL file path (streamed on upload)
signaljt.evaluate(predictions_url="https://.../p.json")  # a public / GCS URL (no upload)
signaljt.evaluate({"100000": "predicted text", ...})  # in-memory dict {index: hypothesis}
signaljt.evaluate([{"index": "100000", "hypothesis": "..."}, ...])  # in-memory list

The prediction set may span any subset of languages — the backend matches each index to its language's ground truth and reports per-language scores.

Large prediction files: the file is streamed on upload (not loaded into memory). For very large sets you can host the file on GCS / any public URL and pass predictions_url= to skip the upload entirely.

The EvalResult object

Everything is reachable by attribute (.) or key ([...]) — nested dicts are wrapped lazily, so it's cheap even with tens of thousands of chunks.

r = signaljt.evaluate("predictions.json")

r.oiwer                          # shortcut for r.overall.weighted_oiwer
r.eval_id                        # server eval id
r.result_url                     # signed URL to the full result JSON (valid ~2 days)

# Overall
r.overall.weighted_oiwer
r.overall.macro_oiwer
r.overall.word.errors            # errors / count / sub / ins / del
r.overall.total_utterances
r.overall.total_duration_sec

# Per language (keyed by ISO code, same codes as fetch_dataset)
r.per_language["hi"].oiwer
r.per_language.hi.name           # "hindi"
r.per_language.hi.coverage.scored, r.per_language.hi.coverage.missing
for iso, lang in r.per_language.items():
    print(iso, lang.oiwer, dict(lang.word))

# Per utterance (lazy list — only built when you touch it)
c = r.per_chunk[0]
c.index, c.language, c.duration
c.word.errors
c.alignment.hyp                  # ["word1", "word2", ...]
c.alignment.ops                  # ["c", "s", "i", "d", ...]  (correct / sub / ins / del)
c.tags                           # ["correct"], ["wrong_script"], ...

# Run metadata
r.meta.penalize_missing, r.meta.tag_rows, r.meta.strip_bracketed

# Export
r.to_dict()                      # the raw result dict
r.save("eval_result.json")       # write the full result to disk

Result schema

Section Fields
overall weighted_oiwer, macro_oiwer, weighted_by, word{errors,count,sub,ins,del}, total_utterances, total_duration_sec, languages_scored, extra_predictions
per_language[iso] iso, name, oiwer, word{...}, utterances, coverage{scored,missing,extra}
per_chunk[i] index, language, duration, cps, p808_mos, native_district, native_state, gender, speaker_id, predicted, status, word{...}, alignment{hyp[],ops[]}, tags[]
meta scored_at, audio_version, lattice_version, penalize_missing, tag_rows, strip_bracketed

alignment.ops codes: c = correct, s = substitution, i = insertion, d = deletion.

Scoring options

Option Default Effect
penalize_missing True Count ground-truth utterances you didn't predict as full errors. Set False to score only what you predicted (missing ones are reported in coverage but not penalized).
tag_rows True Add error-type tags to each per-chunk entry (correct, wrong_script, internal_deletion, ...).
strip_bracketed False Strip bracketed tokens like <noise> / [laughter] from the hypothesis before scoring. 
# "How good is my model only on what it transcribed?"
signaljt.evaluate(preds, penalize_missing=False)

# "How good over the whole benchmark, including gaps?" (default)
signaljt.evaluate(preds, penalize_missing=True)

Non-blocking submit / poll

For fire-and-forget or batch pipelines, submit without waiting:

job = signaljt.submit("predictions.json", eval_name="run-42")
job.eval_id                      # save this
job.status                       # "queued"

# ... later, same or different process ...
result = job.wait()                              # block until completed
result = signaljt.get_result(job.eval_id, wait=True)   # by id, from anywhere

EvalJob methods:

  • job.refresh() — one status poll; updates job.status, returns raw metadata
  • job.wait(poll_interval=3.0, timeout=1800.0, progress=False) — block, then return an EvalResult
  • job.result() — fetch now; raises if not yet completed

Logging to wandb / TensorBoard

result.metrics() returns a flat, slash-namespaced {name: value} dict — overall and per-language — that both tools group nicely:

metrics = result.metrics()
# {
#   "oiwer/weighted": 0.1033, "oiwer/macro": 0.103,
#   "words/errors": 18417, "words/sub": 9586, "words/ins": 3181,
#   "words/del": 5650, "words/count": 178265,
#   "oiwer_by_language/hi": 0.104, "oiwer_by_language/bn": 0.104, ...
# }

# Weights & Biases
import wandb
wandb.log(result.metrics())

# TensorBoard
for name, value in result.metrics().items():
    writer.add_scalar(name, value, step)

Control what's included:

result.metrics(per_language=True, words=True, prefix="oiwer")

Distributed training (DDP)

When you score inside distributed training (e.g. 4 GPUs), the framework runs your eval/metrics code on every rank, so a naive evaluate() submits the dataset N times — N duplicate entries per epoch. Two flags fix this.

ddp=True (recommended) — one submission, identical result on every rank. Rank 0 submits; every other rank fetches that same run from the API by eval_name (read-only, no new entry), so all ranks return an identical EvalResult:

# inside HuggingFace Trainer's compute_metrics (or any per-rank eval hook)
def compute_metrics(pred):
    predictions = {idx: text for idx, text in zip(indices, decode(pred))}
    result = signaljt.evaluate(
        predictions,
        eval_name=f"my-run-epoch-{epoch}",
        progress=False,
        ddp=True,
    )
    return {"oiwer": result.oiwer}      # identical on every rank; ONE eval submitted

Requirements:

  • eval_name must be identical across ranks (derive it from the epoch/step — it already is the same on all ranks).
  • eval_name must be unique per submission, so a re-run doesn't fetch a stale run of the same name. Include the epoch, plus a run id if you re-use output dirs: 
    run_id = os.environ.get("TORCHELASTIC_RUN_ID", "")   # shared across ranks under torchrun/accelerate
eval_name = f"my-run-{run_id}-epoch-{epoch}"

Because the non-main ranks block until rank 0's eval completes, ddp=True also acts as a natural barrier that keeps the ranks in sync across the eval step.

main_process_only=True (lighter) — rank 0 submits, others get None. No extra API calls, but you must return matching metric keys yourself so the trainer doesn't choke on an inconsistent dict across ranks:

    result = signaljt.evaluate(predictions, eval_name=name, main_process_only=True)
    if result is None:                       # non-main ranks
        return {"oiwer": float("inf")}       # placeholder; non-rank-0 metrics are ignored anyway
    return {"oiwer": result.oiwer}

Detect the main process yourself: signaljt.is_main_process() returns True only on rank 0.

Works with any DDP framework. Rank detection uses only the standard RANK / LOCAL_RANK environment variables that torchrun, accelerate launch, HuggingFace Trainer, PyTorch Lightning, and NeMo all set — so ddp=True behaves the same everywhere; just call it from wherever your per-rank validation/metrics code runs. (Multi-node: this relies on the global RANK, which torchrun/accelerate export. A launcher that only sets a per-node LOCAL_RANK would submit once per node — use torchrun/accelerate to avoid that.)

CLI: eval / result

The CLI takes a predictions file or URL and prints the result URL on stdout (progress goes to stderr, so URL=$(...) captures only the URL):

signaljt eval predictions.json                 # submit + wait, prints result URL
signaljt eval https://.../predictions.jsonl    # from a URL
signaljt eval predictions.json --name my-run --strip-bracketed
signaljt eval predictions.json --no-penalize-missing --no-tag-rows
signaljt eval predictions.json --no-wait       # submit only, prints eval_id
signaljt eval predictions.json --quiet         # no progress on stderr

# Fetch an existing eval later
signaljt result <eval_id>                       # prints result URL if completed, else status
signaljt result <eval_id> --wait                # block until completed

Using with ML frameworks

ds is just an Index → absolute audio path mapping, so it drops into any stack. The core eval loop needs no adapter:

# transformers — pipelines accept a file path directly
from transformers import pipeline
asr = pipeline("automatic-speech-recognition", model="openai/whisper-large-v3")
preds = {it.index: asr(it.audio_path)["text"] for it in ds}
result = signaljt.evaluate(preds)

Thin adapters when you want a framework-native object (no extra deps in this package):

# PyTorch DataLoader
import torch, torchaudio
class TorchDS(torch.utils.data.Dataset):
    def __init__(self, ds): self.items = ds.items
    def __len__(self): return len(self.items)
    def __getitem__(self, i):
        wav, sr = torchaudio.load(self.items[i].audio_path)
        return self.items[i].index, wav, sr
loader = torch.utils.data.DataLoader(TorchDS(ds), batch_size=8, collate_fn=list)

# HuggingFace datasets (lazy audio decoding)
import datasets
hf = datasets.Dataset.from_dict(
    {"index": ds.indices, "audio": [it.audio_path for it in ds]}
).cast_column("audio", datasets.Audio())

# fairseq / wav2vec TSV manifest
import soundfile as sf, os
with open("eval.tsv", "w") as f:
    print(ds.root, file=f)
    for it in ds:
        print("%s\t%d" % (os.path.relpath(it.audio_path, ds.root),
                          sf.info(it.audio_path).frames), file=f)

Configuration reference

Environment variables (all optional):

Variable Purpose
SIGNALJT_API_KEY API key for non-interactive use (CI / Slurm batch)
SIGNALJT_CACHE_DIR Override the cache root (default ~/.cache/signaljt)
SIGNALJT_API_BASE Override the API base URL (staging / testing)
HTTPS_PROXY / NO_PROXY Standard proxy settings, honoured on all requests

Cache layout (~/.cache/signaljt/ by default):

~/.cache/signaljt/
├── token                        # your API key (chmod 600)
└── datasets/
    └── <code>/                  # e.g. bn/
        ├── <Language>_manifest.jsonl
        └── audio/*.flac

Error handling

All exceptions derive from signaljt.SignaljtError:

Exception Raised when
SignaljtError base class — catch this to handle any library error
NotLoggedInError no API key found (arg / env / saved token)
AuthenticationError the server rejected the key (invalid / expired)
ApiError the API returned an unexpected response
DownloadError a transfer failed after retries, or a checksum mismatch
ManifestError the dataset manifest was missing or malformed 
import signaljt
try:
    ds = signaljt.fetch_dataset("bn")
except signaljt.NotLoggedInError:
    signaljt.login()
except signaljt.SignaljtError as e:
    print("signaljt failed:", e)

Full API reference

Auth

signaljt.login(api_key=None, *, validate=True) -> None
signaljt.logout() -> None
signaljt.whoami(api_key=None) -> dict
signaljt.get_api_key(api_key=None, *, required=True) -> str | None
signaljt.notebook_login(validate=True) -> None

Datasets

signaljt.list_languages(api_key=None) -> list[str]

signaljt.fetch_dataset(
    language,                      # str code | list[str] | "all"
    *, api_key=None, cache_dir=None,
    local_dir=None, local_dir_mode="auto",
    connections=1, extract_workers=1,
    chunk_size=1048576, min_segment=8388608, max_segment=67108864, retries=4,
    force=False, refresh=False,
    max_parallel=4, continue_on_error=True, progress=True,
) -> Dataset | dict[str, Dataset]

# Dataset
len(ds); iter(ds); ds[int]; ds[index]; index in ds
ds.get(index, default=None); ds.indices; ds.paths(); ds.items
ds.root; ds.language_code
ds.export(local_dir, *, mode="auto") -> Dataset

# DatasetItem
item.index; item.audio_path; item.extra

Scoring

signaljt.evaluate(
    predictions=None, *, predictions_url=None, eval_name=None,
    penalize_missing=True, tag_rows=True, strip_bracketed=False,
    api_key=None, poll_interval=3.0, timeout=1800.0, progress=False,
) -> EvalResult

signaljt.submit(
    predictions=None, *, predictions_url=None, eval_name=None,
    penalize_missing=True, tag_rows=True, strip_bracketed=False, api_key=None,
) -> EvalJob

signaljt.get_result(
    eval_id=None, *, eval_name=None, api_key=None,
    wait=False, poll_interval=3.0, timeout=1800.0,
) -> EvalResult

# EvalJob
job.eval_id; job.eval_name; job.status
job.refresh() -> dict
job.wait(*, poll_interval=3.0, timeout=1800.0, progress=False) -> EvalResult
job.result() -> EvalResult

# EvalResult
r.oiwer; r.overall; r.per_language; r.per_chunk; r.meta
r.eval_id; r.eval_name; r.result_url
r.metrics(*, per_language=True, words=True, prefix="oiwer") -> dict
r.to_dict() -> dict
r.save(path) -> None

End-to-end example

import signaljt

signaljt.login("your-api-key")                # once (or SIGNALJT_API_KEY)

# Score a model across every language and log per-language OI-WER
scores = {}
for code in signaljt.list_languages():
    ds = signaljt.fetch_dataset(code)         # cached after first pull
    preds = {it.index: my_asr_model(it.audio_path) for it in ds}
    scores.update(preds)

result = signaljt.evaluate(scores, eval_name="whisper-large-v3")
print("Overall OI-WER:", result.oiwer)
for iso, lang in result.per_language.items():
    print(f"  {iso}: {lang.oiwer}")

result.save("eval_result.json")

import wandb
wandb.init(project="asr-eval")
wandb.log(result.metrics())

CLI equivalent for a single scored file:

export SIGNALJT_API_KEY="your-api-key"
signaljt download all
# ... run your model, write predictions.json ...
RESULT_URL=$(signaljt eval predictions.json --name whisper-large-v3)
echo "Full result: $RESULT_URL"

About

No description, website, or topics provided.

Resources

Readme

License

MIT license
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

2 tags

Packages

 
 
 

Contributors

Languages