File size: 3,867 Bytes
7381480 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """Question rendering, calibrated answers and audio preprocessing."""
import json
import math
from typing import Literal
import numpy as np
import torch
from pydantic import BaseModel, ConfigDict, Field
MARKER = "\n<spev_score>\n"
class ChoiceQuestion(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
type: Literal["choice"] = "choice"
instructions: str
criteria: dict[str, str] = Field(min_length=2, max_length=255)
class ChoiceQuestions(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
questions: dict[str, ChoiceQuestion] = Field(min_length=1)
def literal(value):
return (
json.dumps(value, ensure_ascii=False)
.replace("<", "\\u003c")
.replace(">", "\\u003e")
)
def render(state, questions):
if not questions:
raise ValueError("At least one question is required")
schema = []
mapping = []
for key, question in questions.items():
criteria = question["criteria"]
if not 2 <= len(criteria) <= 255:
raise ValueError("Each question requires 2–255 criteria")
schema.append(
{
"instructions": question["instructions"],
"criteria": list(criteria.values()),
}
)
mapping.extend((key, name) for name in criteria)
# IDs only route outputs. Descriptions carry the criterion semantics.
text = "Use the audio and state as evidence for the supplied questions.\n"
text += "State: " + literal(state) + "\nQuestions: " + literal(schema) + "\n"
for question in questions.values():
for description in question["criteria"].values():
text += (
literal(
{"question": question["instructions"], "criterion": description}
)
+ MARKER
)
return text, mapping
def locate_markers(ids, marker, expected):
positions = [
i + len(marker) - 1
for i in range(len(ids) - len(marker) + 1)
if ids[i : i + len(marker)] == marker
]
if len(positions) != expected:
raise ValueError("Scoring anchor count mismatch")
return positions
def answers(logits, mapping, temperature=1.0):
if not math.isfinite(temperature) or temperature <= 0:
raise ValueError("Temperature must be positive and finite")
if not torch.isfinite(logits).all():
raise ValueError("Nonfinite logits")
grouped = {}
for i, (question, criterion) in enumerate(mapping):
grouped.setdefault(question, []).append((criterion, i))
result = {}
for key, options in grouped.items():
probabilities = (
(logits[[i for _, i in options]].double() / temperature).softmax(0).tolist()
)
p = dict(zip([name for name, _ in options], probabilities))
choice = min(p, key=lambda name: (-p[name], name))
result[key] = {
"choice": choice,
"probabilities": p,
"confidence": (p[choice] - 1 / len(p)) / (1 - 1 / len(p)),
}
return result
def prepare_audio(audio, sample_rate):
from scipy.signal import resample_poly
wave = np.asarray(audio, dtype=np.float32)
if wave.ndim == 2:
wave = wave.mean(axis=1)
if wave.ndim != 1 or not len(wave) or not np.isfinite(wave).all():
raise ValueError(
"Audio must be finite mono samples or frames-by-channels samples"
)
if (
not isinstance(sample_rate, int)
or isinstance(sample_rate, bool)
or sample_rate <= 0
):
raise ValueError("sample_rate must be a positive integer")
if sample_rate != 16000:
divisor = math.gcd(sample_rate, 16000)
wave = resample_poly(wave, 16000 // divisor, sample_rate // divisor).astype(
np.float32
)
return wave
|