mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 13:40:27 +02:00
JB
338ff104e8
- Added `ConversionService` class to handle job queuing, processing, and cancellation. - Introduced `Job`, `JobLog`, and `JobResult` data classes to manage job details and results. - Implemented job status tracking with enums for better state management. - Created a web interface with HTML templates for job submission and monitoring. - Developed CSS styles for a modern UI layout and responsive design. - Added functionality for voice profile management in the voice mixer. - Implemented a Docker Compose configuration for GPU support. - Wrote unit tests for the conversion service to ensure job processing works as expected.
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from abogen.web.service import JobStatus, build_service
|
|
|
|
|
|
def test_service_processes_job(tmp_path):
|
|
uploads = tmp_path / "uploads"
|
|
outputs = tmp_path / "outputs"
|
|
uploads.mkdir()
|
|
outputs.mkdir()
|
|
|
|
source = uploads / "sample.txt"
|
|
payload = "hello world"
|
|
source.write_text(payload, encoding="utf-8")
|
|
|
|
runner_invocations: list[str] = []
|
|
|
|
def runner(job):
|
|
runner_invocations.append(job.id)
|
|
job.add_log("processing")
|
|
job.progress = 1.0
|
|
job.processed_characters = job.total_characters or len(payload)
|
|
job.result.audio_path = outputs / f"{job.id}.wav"
|
|
|
|
service = build_service(
|
|
runner=runner,
|
|
output_root=outputs,
|
|
uploads_root=uploads,
|
|
)
|
|
|
|
job = service.enqueue(
|
|
original_filename="sample.txt",
|
|
stored_path=source,
|
|
language="a",
|
|
voice="af_alloy",
|
|
speed=1.0,
|
|
use_gpu=False,
|
|
subtitle_mode="Sentence",
|
|
output_format="wav",
|
|
save_mode="Save next to input file",
|
|
output_folder=outputs,
|
|
replace_single_newlines=False,
|
|
subtitle_format="srt",
|
|
total_characters=len(payload),
|
|
)
|
|
|
|
deadline = time.time() + 5
|
|
while time.time() < deadline and job.status not in {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED}:
|
|
time.sleep(0.05)
|
|
|
|
service.shutdown()
|
|
|
|
assert runner_invocations, "conversion runner was never called"
|
|
assert job.status is JobStatus.COMPLETED
|
|
assert job.progress == 1.0
|
|
assert job.result.audio_path == outputs / f"{job.id}.wav" |