Merge pull request #1 from slashml/first_release

First release

Author: JJneid

.DS_Store

@@ -0,0 +1,2 @@
+.DS_Store
+services.py

.gitignore

@@ -0,0 +1,69 @@
+from django.http import JsonResponse
+import requests
+import os
+from pathlib import Path
+
+def upload_audio(*, audio_data, service_name) -> None:
+    # with open('test.mp3', 'wb+') as mp3:
+    #     for chunk in audio_data.chunks():
+    #         mp3.write(chunk)
+
+    if service_name == 'assembly':
+        Assembly().upload_audio(audio_file=audio_data)
+
+
+
+class Assembly:
+    ASSEMBLY_BASE_URL = 'https://api.assemblyai.com/v2'
+    ASSEMBLY_UPLOAD_URL = ASSEMBLY_BASE_URL+'/upload'
+    ASSEMBLY_TRANSCRIPT_URL = ASSEMBLY_BASE_URL+'/transcript'
+    ASSEMBLY_TRANSCRIPT_STATUS_URL = lambda id: f"{Assembly.ASSEMBLY_BASE_URL}+/transcript/{id}"
+
+    HEADERS:dict = {}
+
+    def __init__(self) -> None:
+        self.HEADERS = {'authorization': os.environ.get('ASSEMBLY_API_KEY')}
+
+    def _read_file(filename, chunk_size=5242880):
+        with open(filename, 'rb') as _file:
+            while True:
+                data = _file.read(chunk_size)
+                if not data:
+                    break
+                yield data
+
+    def upload_audio(self, *, audio_file:Path, headers:dict=None) -> JsonResponse:
+        import pdb
+        pdb.set_trace()
+        response = requests.post(self.ASSEMBLY_UPLOAD_URL,
+                                headers=headers if headers else self.HEADERS,
+                                data=self._read_file(audio_file))
+        return response.json()
+
+    def request_transcript(self, *, upload_url:dict, headers:dict=None):
+
+        transcript_request = {
+            'audio_url': upload_url['upload_url']
+        }
+
+        transcript_response = requests.post(
+            self.ASSEMBLY_TRANSCRIPT_URL,
+            json=transcript_request,
+            headers=headers if headers else self.HEADERS,
+        )
+
+        return transcript_response.json()
+
+    def transcription_status(self, *, transcript_id:str, upload_url:dict, headers:dict=None):
+
+        transcript_request = {
+            'audio_url': upload_url['upload_url']
+        }
+
+        transcript_response = requests.post(
+            self.ASSEMBLY_TRANSCRIPT_STATUS_URL(transcript_id),
+            json=transcript_request,
+            headers=headers if headers else self.HEADERS,
+        )
+
+        return transcript_response.json()

SDK_v1/.DS_Store

@@ -0,0 +1,22 @@
+class SpeechToText:
+    SLASHML_BASE_URL = 'https://api.slashml.com/v1/speech-to-text'
+    SLASHML_UPLOAD_URL = SLASHML_BASE_URL+'/upload'
+    SLASHML_TRANSCRIPT_URL = SLASHML_BASE_URL+'/transcribe'
+    SLASHML_TRANSCRIPT_STATUS_URL = lambda self,id: f"{SpeechToText.SLASHML_TRANSCRIPT_URL}/{id}"
+
+
+    def upload_audio(self, file_location) :
+        # here we can also add the service? assemblyai, aws, gcp?
+        return    self.SLASHML_UPLOAD_URL #response.json()
+
+    def transcribe(self,upload_url, model_params:dict()):
+        # here we can add more model params
+        transcript_request = {'audio_url': upload_url}
+        #transcript_response = requests.post( self.SLASHML_TRANSCRIPT_URL, json=transcript_request)
+        job_id=self.SLASHML_TRANSCRIPT_URL
+        return job_id
+
+    def status(self, job_id:str):
+        #options={"status1"="queue","status2":"completed","status3":"deleted","text":"output_api_result"...}
+        return self.SLASHML_TRANSCRIPT_STATUS_URL(job_id) 
+

SDK_v1/services.py

@@ -0,0 +1,41 @@
+#### unit testing
+import pytest
+
+
+def test_upload():
+    import speechtotext
+    # given: there is a local file
+    speect_to_text = speechtotext.SpeechToText()
+    file_location="local_path"
+    # when
+    result = speect_to_text.upload_audio(file_location)
+
+    # then
+    assert result== 'https://api.slashml.com/v1/speech-to-text/upload'
+
+def test_transcribe():
+    import speechtotext
+    # given: there is a local file
+    speect_to_text = speechtotext.SpeechToText()
+    upload_url="local_path"
+    # when
+    result = speect_to_text.transcribe(upload_url,dict())
+
+    # then
+    assert result== 'https://api.slashml.com/v1/speech-to-text/transcribe'
+
+
+def test_status():
+    import speechtotext
+    # given: there is a local file
+    speect_to_text = speechtotext.SpeechToText()
+    # when
+    job_id = '123'
+    result = speect_to_text.status(job_id)
+
+    # then
+    assert result== 'https://api.slashml.com/v1/speech-to-text/transcribe/{}'.format(job_id)
+
+
+# def test_status():
+#     assert