Spaces:
Running on Zero
Running on Zero
File size: 3,401 Bytes
f94ee13 | 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 | import csv
import hashlib
import json
import math
import struct
import wave
from autoace_audio.batch import ManifestEntry, parse_manifest_for_inference, process_batch
from autoace_audio.pipeline import prepare_audio
def write_wav(path, frequency=440, duration=0.15, sample_rate=16_000):
with wave.open(str(path), "wb") as output:
output.setnchannels(1)
output.setsampwidth(2)
output.setframerate(sample_rate)
frames = bytearray()
for index in range(int(duration * sample_rate)):
sample = int(8_000 * math.sin(2 * math.pi * frequency * index / sample_rate))
frames.extend(struct.pack("<h", sample))
output.writeframes(frames)
def write_manifest(path, label_value):
with path.open("w", encoding="utf-8", newline="") as stream:
writer = csv.DictWriter(stream, fieldnames=("name", "result_json"))
writer.writeheader()
writer.writerow({"name": "call.wav", "result_json": label_value})
def test_batch_isolates_file_failures_and_continues(tmp_path):
write_wav(tmp_path / "first.wav", 440)
(tmp_path / "broken.mp3").write_bytes(b"not audio")
write_wav(tmp_path / "third.wav", 660)
entries = (
ManifestEntry("first.wav"),
ManifestEntry("broken.mp3"),
ManifestEntry("third.wav"),
)
report = process_batch(entries, tmp_path, prepare_audio)
assert [item.status for item in report.items] == ["ok", "error", "ok"]
assert report.success_count == 2
assert report.failure_count == 1
assert report.items[1].failure is not None
assert report.items[1].failure.code == "audio_decode_failed"
assert str(tmp_path) not in report.items[1].failure.message
def test_missing_file_is_an_isolated_error(tmp_path):
write_wav(tmp_path / "present.wav")
entries = (ManifestEntry("missing.wav"), ManifestEntry("present.wav"))
report = process_batch(entries, tmp_path, lambda path: path.name)
assert [item.status for item in report.items] == ["error", "ok"]
assert report.items[0].failure.code == "missing_file"
def test_inference_results_do_not_change_when_labels_are_removed_or_altered(tmp_path):
audio = tmp_path / "call.wav"
write_wav(audio)
empty_manifest = tmp_path / "empty.csv"
altered_manifest = tmp_path / "altered.csv"
write_manifest(empty_manifest, "")
write_manifest(
altered_manifest,
json.dumps(
{
"emotional_tone": "distressed",
"emotional_intensity": "high",
"background_noise_present": True,
"background_noise_type": "television",
"background_noise_severity": "high",
"audio_quality": "severely_impaired",
"speaker_overlap_present": True,
"long_silence_present": True,
"confidence": 1.0,
}
),
)
def inference_stub(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
without_labels = process_batch(
parse_manifest_for_inference(empty_manifest), tmp_path, inference_stub
)
with_altered_labels = process_batch(
parse_manifest_for_inference(altered_manifest), tmp_path, inference_stub
)
assert without_labels.items == with_altered_labels.items
assert not hasattr(parse_manifest_for_inference(altered_manifest)[0], "result_json")
|