| import typer |
| import torch |
| import subprocess |
| from pathlib import Path |
|
|
| from expert import UpstreamExpert |
|
|
| SUBMISSION_FILES = ["expert.py", "model.pt"] |
| SAMPLE_RATE = 16000 |
| SECONDS = [2, 1.8, 3.7] |
|
|
| app = typer.Typer() |
|
|
| @app.command() |
| def validate(): |
| |
| for file in SUBMISSION_FILES: |
| if not Path(file).is_file(): |
| raise ValueError(f"File {file} not found! Please include {file} in your submission") |
|
|
| try: |
| upstream = UpstreamExpert(ckpt="model.pt") |
| samples = [round(SAMPLE_RATE * sec) for sec in SECONDS] |
| wavs = [torch.rand(sample) for sample in samples] |
| results = upstream(wavs) |
|
|
| assert isinstance(results, dict) |
| tasks = ["PR", "SID", "ER", "ASR", "ASV", "SD", "QbE", "ST", "SS", "SE", "secret"] |
| for task in tasks: |
| hidden_states = results.get(task, results["hidden_states"]) |
| assert isinstance(hidden_states, list) |
|
|
| for state in hidden_states: |
| assert isinstance(state, torch.Tensor) |
| assert state.dim() == 3, "(batch_size, max_sequence_length_of_batch, hidden_size)" |
| assert state.shape == hidden_states[0].shape |
|
|
| downsample_rate = upstream.get_downsample_rates(task) |
| assert isinstance(downsample_rate, int) |
| assert abs(round(max(samples) / downsample_rate) - hidden_states[0].size(1)) < 5, "wrong downsample rate" |
|
|
| except: |
| print("Please check the Upstream Specification on https://superbbenchmark.org/challenge-slt2022/upstream") |
| raise |
|
|
| typer.echo("All submission files validated!") |
| typer.echo("Now you can upload these files to huggingface's Hub.") |
|
|
|
|
| @app.command() |
| def upload(commit_message: str): |
| subprocess.call("git pull origin main".split()) |
| subprocess.call(["git", "add", "."]) |
| subprocess.call(["git", "commit", "-m", f"Upload Upstream: {commit_message} "]) |
| subprocess.call(["git", "push"]) |
| typer.echo("Upload successful!") |
| typer.echo("Please go to https://superbbenchmark.org/submit to make a submission with the following information:") |
| typer.echo("1. Organization Name") |
| typer.echo("2. Repository Name") |
| typer.echo("3. Commit Hash (full 40 characters)") |
| typer.echo("These information can be shown by: python cli.py info") |
|
|
| @app.command() |
| def info(): |
| result = subprocess.run(["git", "config", "--get", "remote.origin.url"], capture_output=True) |
| url = result.stdout.decode("utf-8").strip() |
| organization = url.split("/")[-2] |
| repo = url.split("/")[-1] |
|
|
| result = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True) |
| commit_hash = result.stdout.decode("utf-8").strip() |
|
|
| typer.echo(f"Organization Name: {organization}") |
| typer.echo(f"Repository Name: {repo}") |
| typer.echo(f"Commit Hash: {commit_hash}") |
|
|
| if __name__ == "__main__": |
| app() |
|
|
