From 9550707997e53ec774aa8d22ce4614bcc147dbbc Mon Sep 17 00:00:00 2001 From: Ahmed Mansour Date: Sun, 12 Jan 2025 09:32:44 -0500 Subject: [PATCH] Add Forced Alignment Client and Models (#120) * Introduced `ForcedAlignmentClient` for interacting with the Rev AI forced alignment API. * Added data models: `ForcedAlignmentJob`, `ForcedAlignmentResult`, `ElementAlignment`, and `Monologue` to handle alignment jobs. * Implemented methods for processing jobs in forced_alignment_client. * Created a module for forced alignment models to organize classes. Signed-off-by: Ahmed Mansour --- src/rev_ai/forced_alignment_client.py | 85 +++++++++++++++++++ .../models/forced_alignment/__init__.py | 6 ++ .../forced_alignment/forced_alignment_job.py | 41 +++++++++ .../forced_alignment_result.py | 78 +++++++++++++++++ 4 files changed, 210 insertions(+) create mode 100644 src/rev_ai/forced_alignment_client.py create mode 100644 src/rev_ai/models/forced_alignment/__init__.py create mode 100644 src/rev_ai/models/forced_alignment/forced_alignment_job.py create mode 100644 src/rev_ai/models/forced_alignment/forced_alignment_result.py diff --git a/src/rev_ai/forced_alignment_client.py b/src/rev_ai/forced_alignment_client.py new file mode 100644 index 0000000..1d6331c --- /dev/null +++ b/src/rev_ai/forced_alignment_client.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +"""Client used for interacting with our forced alignment api""" + +import json +from .generic_api_client import GenericApiClient +from .models.forced_alignment import ForcedAlignmentJob, ForcedAlignmentResult + + +class ForcedAlignmentClient(GenericApiClient): + """Client for interacting with the Rev AI forced alignment api""" + + # Default version of Rev AI forced alignment api + api_version = 'v1' + + # Default api name of Rev AI forced alignment api + api_name = 'alignment' + + def __init__(self, access_token): + """Constructor + + :param access_token: access token which authorizes all requests and links them to your + account. Generated on the settings page of your account dashboard + on Rev AI. + """ + GenericApiClient.__init__(self, access_token, self.api_name, self.api_version, + ForcedAlignmentJob.from_json, ForcedAlignmentResult.from_json) + + def submit_job_url( + self, + source_config=None, + source_transcript_config=None, + transcript_text=None, + metadata=None, + delete_after_seconds=None, + notification_config=None, + language=None): + """Submit a job to the Rev AI forced alignment api. + + :param source_config: CustomerUrlData object containing url of the source media and + optional authentication headers to use when accessing the source url + :param source_transcript_config: CustomerUrlData object containing url of the transcript file and + optional authentication headers to use when accessing the transcript url + :param transcript_text: The text of the transcript to be aligned (no punctuation, just words) + :param metadata: info to associate with the alignment job + :param delete_after_seconds: number of seconds after job completion when job is auto-deleted + :param notification_config: CustomerUrlData object containing the callback url to + invoke on job completion as a webhook and optional authentication headers to use when + calling the callback url + :param language: Language code for the audio and transcript. One of: "en", "es", "fr" + :returns: ForcedAlignmentJob object + :raises: HTTPError + """ + if not source_config: + raise ValueError('source_config must be provided') + if not (source_transcript_config or transcript_text): + raise ValueError('Either source_transcript_config or transcript_text must be provided') + if source_transcript_config and transcript_text: + raise ValueError('Only one of source_transcript_config or transcript_text may be provided') + + payload = self._enhance_payload({ + 'source_config': source_config.to_dict() if source_config else None, + 'source_transcript_config': source_transcript_config.to_dict() if source_transcript_config else None, + 'transcript_text': transcript_text, + 'language': language + }, metadata, None, delete_after_seconds, notification_config) + + return self._submit_job(payload) + + def get_result_json(self, id_): + """Get result of a forced alignment job as json. + + :param id_: id of job to be requested + :returns: job result data as raw json + :raises: HTTPError + """ + return self._get_result_json(id_) + + def get_result_object(self, id_): + """Get result of a forced alignment job as ForcedAlignmentResult object. + + :param id_: id of job to be requested + :returns: job result data as ForcedAlignmentResult object + :raises: HTTPError + """ + return self._get_result_object(id_) \ No newline at end of file diff --git a/src/rev_ai/models/forced_alignment/__init__.py b/src/rev_ai/models/forced_alignment/__init__.py new file mode 100644 index 0000000..174bedd --- /dev/null +++ b/src/rev_ai/models/forced_alignment/__init__.py @@ -0,0 +1,6 @@ +"""Module containing models for Rev AI forced alignment""" + +from .forced_alignment_job import ForcedAlignmentJob +from .forced_alignment_result import ForcedAlignmentResult, Monologue, ElementAlignment + +__all__ = ['ForcedAlignmentJob', 'ForcedAlignmentResult', 'Monologue', 'ElementAlignment'] \ No newline at end of file diff --git a/src/rev_ai/models/forced_alignment/forced_alignment_job.py b/src/rev_ai/models/forced_alignment/forced_alignment_job.py new file mode 100644 index 0000000..80cb77b --- /dev/null +++ b/src/rev_ai/models/forced_alignment/forced_alignment_job.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +"""Contains ForcedAlignmentJob dataclass""" + +from dataclasses import dataclass +from typing import Optional, Dict, Any + +from ..apijob import ApiJob + + +@dataclass +class ForcedAlignmentJob(ApiJob): + """Dataclass containing information about a Rev AI forced alignment job + + :param id: unique identifier for this job + :param status: current job status + :param created_on: date and time at which this job was created + :param completed_on: date and time at which this job was completed + :param metadata: customer-provided metadata + :param type: type of job (always "alignment") + :param media_url: URL of the media to be aligned + :param failure: details about job failure if status is "failed" + """ + media_url: Optional[str] = None + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'ForcedAlignmentJob': + """Creates a ForcedAlignmentJob from the given json dictionary + + :param json: json dictionary to convert + :returns: ForcedAlignmentJob + """ + return ForcedAlignmentJob( + id=json.get('id'), + status=json.get('status'), + created_on=json.get('created_on'), + completed_on=json.get('completed_on'), + metadata=json.get('metadata'), + type=json.get('type'), + media_url=json.get('media_url'), + failure=json.get('failure') + ) \ No newline at end of file diff --git a/src/rev_ai/models/forced_alignment/forced_alignment_result.py b/src/rev_ai/models/forced_alignment/forced_alignment_result.py new file mode 100644 index 0000000..1d5daf3 --- /dev/null +++ b/src/rev_ai/models/forced_alignment/forced_alignment_result.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""Contains ForcedAlignmentResult dataclass""" + +from dataclasses import dataclass +from typing import List, Dict, Any + + +@dataclass +class ElementAlignment: + """Dataclass containing information about an aligned word + + :param value: the word that was aligned + :param ts: start time of the word in seconds + :param end_ts: end time of the word in seconds + :param type: type of element (always "text") + """ + value: str + ts: float + end_ts: float + type: str = "text" + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'ElementAlignment': + """Creates an ElementAlignment from the given json dictionary + + :param json: json dictionary to convert + :returns: ElementAlignment + """ + return ElementAlignment( + value=json.get('value'), + ts=json.get('ts'), + end_ts=json.get('end_ts'), + type=json.get('type', 'text') + ) + + +@dataclass +class Monologue: + """Dataclass containing information about a monologue section + + :param speaker: speaker identifier + :param elements: list of words in this monologue with timing information + """ + speaker: int + elements: List[ElementAlignment] + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'Monologue': + """Creates a Monologue from the given json dictionary + + :param json: json dictionary to convert + :returns: Monologue + """ + return Monologue( + speaker=json.get('speaker', 0), + elements=[ElementAlignment.from_json(element) for element in json.get('elements', [])] + ) + + +@dataclass +class ForcedAlignmentResult: + """Dataclass containing the result of a forced alignment job + + :param monologues: A Monologue object per speaker containing the words + they spoke with timing information + """ + monologues: List[Monologue] + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'ForcedAlignmentResult': + """Creates a ForcedAlignmentResult from the given json dictionary + + :param json: json dictionary to convert + :returns: ForcedAlignmentResult + """ + return ForcedAlignmentResult( + monologues=[Monologue.from_json(monologue) for monologue in json.get('monologues', [])] + ) \ No newline at end of file