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 <[email protected]>

Author: sourman

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_) 

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'] 

src/rev_ai/models/forced_alignment/forced_alignment_job.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+"""Contains ForcedAlignmentJob dataclass"""
+
+from typing import Dict, Any
+from ..asynchronous.job_status import JobStatus
+
+
+class ForcedAlignmentJob:
+    def __init__(
+            self, id_, created_on, status,
+            completed_on=None,
+            callback_url=None,
+            metadata=None,
+            media_url=None,
+            failure=None,
+            failure_detail=None,
+            processed_duration_seconds=None,
+            delete_after_seconds=None):
+        """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"
+        """
+        self.id = id_
+        self.created_on = created_on
+        self.status = status
+        self.completed_on = completed_on
+        self.callback_url = callback_url
+        self.metadata = metadata
+        self.media_url = media_url
+        self.failure = failure
+        self.failure_detail = failure_detail
+        self.processed_duration_seconds = processed_duration_seconds
+        self.delete_after_seconds = delete_after_seconds
+
+    def __eq__(self, other):
+        """Override default equality operator"""
+        if isinstance(other, self.__class__):
+            return self.__dict__ == other.__dict__
+        return False
+
+    @classmethod
+    def from_json(cls, json: Dict[str, Any]) -> 'ForcedAlignmentJob':
+        """Alternate constructor used for parsing json
+
+        :param json: json dictionary to convert
+        :returns: ForcedAlignmentJob
+        """
+        return cls(
+            id_=json['id'],
+            created_on=json['created_on'],
+            status=JobStatus.from_string(json['status']),
+            completed_on=json.get('completed_on'),
+            callback_url=json.get('callback_url'),
+            metadata=json.get('metadata'),
+            media_url=json.get('media_url'),
+            failure=json.get('failure'),
+            failure_detail=json.get('failure_detail'),
+            processed_duration_seconds=json.get('processed_duration_seconds'),
+            delete_after_seconds=json.get('delete_after_seconds')
+        )

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', [])]
+        )