Merge pull request #57 from nextcloud/feat/trigger-taskprocessing

feat: Add support for taskprocessing trigger event

Author: marcelklehr

.github/workflows/integration_test.yml

@@ -26,7 +26,7 @@ jobs:
       # do not stop on another job's failure
       fail-fast: false
       matrix:
-        php-versions: [ '8.1' ]
+        php-versions: [ '8.2' ]
         databases: [ 'sqlite' ]
         server-versions: [ 'master' ]
 

lib/main.py

@@ -1,5 +1,7 @@
 import threading
+import traceback
 from contextlib import asynccontextmanager
+from threading import Event
 from time import perf_counter, sleep
 import os
 import logging
@@ -40,7 +42,6 @@
 )
 LOGGER = logging.getLogger(os.environ["APP_ID"])
 LOGGER.setLevel(logging.DEBUG)
-ENABLED_FLAG = NextcloudApp().enabled_state
 
 
 def load_models():
@@ -62,18 +63,26 @@ def create_model_loader(file_path):
 
     return lambda: WhisperModel(file_path, device=device)
 
+ENABLED = Event()
 
+TRIGGER = Event()
+WAIT_INTERVAL = 5
+WAIT_INTERVAL_WITH_TRIGGER = 5 * 60
 models = load_models()
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
+    global ENABLED
     setup_nextcloud_logging("stt_whisper2", logging_level=logging.WARNING)
     set_handlers(
         APP,
         enabled_handler,
+        trigger_handler=trigger_handler
     )
-    t = BackgroundProcessTask()
-    t.start()
+    nc = NextcloudApp()
+    if nc.enabled_state:
+        ENABLED.set()
+    start_bg_task()
     yield
 
 
@@ -85,73 +94,76 @@ def get_file(nc, task_id, file_id):
 LAST_MODEL_NAME = None
 LAST_MODEL = None
 
-class BackgroundProcessTask(threading.Thread):
-    def run(self, *args, **kwargs):  # pylint: disable=unused-argument
-        global ENABLED_FLAG
-        global LAST_MODEL_NAME
-        global LAST_MODEL
-
-        nc = NextcloudApp()
-        while True:
-            if not ENABLED_FLAG:
-                sleep(30)
-                ENABLED_FLAG = nc.enabled_state
-                continue
+def start_bg_task():
+    t = threading.Thread(target=background_thread_task)
+    t.start()
 
-            try:
-                next = nc.providers.task_processing.next_task([f'stt_whisper2:{model_name}' for model_name, _ in models.items()], ['core:audio2text'])
-                if not 'task' in next or next is None:
-                    sleep(5)
-                    continue
-                task = next.get('task')
-            except Exception as e:
-                LOGGER.error(str(e))
-                sleep(5)
+def background_thread_task():
+    global ENABLED
+    global LAST_MODEL_NAME
+    global LAST_MODEL
+
+    nc = NextcloudApp()
+    while True:
+        while not ENABLED.is_set():
+            sleep(5)
+
+        try:
+            next = nc.providers.task_processing.next_task([f'stt_whisper2:{model_name}' for model_name, _ in models.items()], ['core:audio2text'])
+            if not 'task' in next or next is None:
+                wait_for_task()
                 continue
+            task = next.get('task')
+        except Exception as e:
+            LOGGER.error(str(e) + "\n" + "".join(traceback.format_exception(e)))
+            wait_for_task(10)
+            continue
+        try:
+            LOGGER.info(f"Next task: {task['id']}")
+            model_name = next.get("provider").get('name').split(':', 2)[1]
+            LOGGER.info( f"model: {model_name}")
+            if LAST_MODEL_NAME == model_name:
+                model = LAST_MODEL
+            else:
+                model_load = models.get(model_name)
+                if model_load is None:
+                    nc.providers.task_processing.report_result(
+                        task["id"], None, "Requested model is not available"
+                    )
+                    continue
+                model = model_load()
+                LAST_MODEL_NAME = model_name
+                LAST_MODEL = model
+
+            LOGGER.info("generating transcription")
+            time_start = perf_counter()
+            file_name = get_file(nc, task["id"], task.get("input").get('input'))
+            segments, info = model.transcribe(file_name)
+            transcript = ''
+            for segment in segments:
+                transcript += segment.text
+                percentage = ( segment.start / info.duration ) * 100
+                nc.providers.task_processing.set_progress(task['id'], percentage)
+            del model
+            LOGGER.info(f"transcription generated: {perf_counter() - time_start}s")
+
+            nc.providers.task_processing.report_result(
+                task["id"],
+                {'output': str(transcript)},
+            )
+        except Exception as e:  # noqa
             try:
-                LOGGER.info(f"Next task: {task['id']}")
-                model_name = next.get("provider").get('name').split(':', 2)[1]
-                LOGGER.info( f"model: {model_name}")
-                if LAST_MODEL_NAME == model_name:
-                    model = LAST_MODEL
-                else:
-                    model_load = models.get(model_name)
-                    if model_load is None:
-                        NextcloudApp().providers.task_processing.report_result(
-                            task["id"], None, "Requested model is not available"
-                        )
-                        continue
-                    model = model_load()
-                    LAST_MODEL_NAME = model_name
-                    LAST_MODEL = model
-
-                LOGGER.info("generating transcription")
-                time_start = perf_counter()
-                file_name = get_file(nc, task["id"], task.get("input").get('input'))
-                segments, _ = model.transcribe(file_name)
-                del model
-                LOGGER.info(f"transcription generated: {perf_counter() - time_start}s")
-
-                transcript = ''
-                for segment in segments:
-                    transcript += segment.text
-                NextcloudApp().providers.task_processing.report_result(
-                    task["id"],
-                    {'output': str(transcript)},
-                )
-            except Exception as e:  # noqa
-                try:
-                    LOGGER.error(str(e))
-                    nc.providers.task_processing.report_result(task["id"], None, str(e))
-                except:
-                    pass
+                LOGGER.error(str(e) + "\n" + "".join(traceback.format_exception(e)))
+                nc.providers.task_processing.report_result(task["id"], None, str(e))
+            except:
+                pass
 
 
 async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str:
-    global ENABLED_FLAG
+    global ENABLED
 
-    ENABLED_FLAG = enabled
     if enabled is True:
+        ENABLED.set()
         LOGGER.info("Hello from %s", nc.app_cfg.app_name)
         for model_name, _ in models.items():
             await nc.providers.task_processing.register(TaskProcessingProvider(
@@ -161,11 +173,27 @@ async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str:
                 expected_runtime=120,
             ))
     else:
+        ENABLED.clear()
         LOGGER.info("Bye bye from %s", nc.app_cfg.app_name)
         for model_name, _ in models.items():
             await nc.providers.task_processing.unregister(f'stt_whisper2:{model_name}', True)
     return ""
 
 
+def trigger_handler(providerId: str):
+    global TRIGGER
+    TRIGGER.set()
+
+# Waits for `interval` seconds or WAIT_INTERVAL
+# In case a TRIGGER event comes in, WAIT_INTERVAL is set (increased) to WAIT_INTERVAL_WITH_TRIGGER
+def wait_for_task(interval = None):
+    global WAIT_INTERVAL
+    global WAIT_INTERVAL_WITH_TRIGGER
+    if interval is None:
+        interval = WAIT_INTERVAL
+    if TRIGGER.wait(timeout=interval):
+        WAIT_INTERVAL = WAIT_INTERVAL_WITH_TRIGGER
+    TRIGGER.clear()
+
 if __name__ == "__main__":
     run_app("main:APP", log_level="trace")

lib/ocs.py

@@ -2,7 +2,7 @@
 import typing
 from json import loads
 
-from httpx import Response, codes, ReadTimeout
+from niquests import Response, codes, RequestException
 from nc_py_api import NextcloudException
 
 
@@ -24,7 +24,7 @@ def check_error(response: Response, info: str = ""):
 		else:
 			phrase = "Unknown error"
 		raise NextcloudException(status_code, reason=phrase, info=info)
-	if not codes.is_error(status_code):
+	if status_code < 400:
 		return
 	raise NextcloudException(status_code, reason=codes(status_code).phrase, info=info)
 
@@ -42,12 +42,9 @@ def ocs(
 	ncSession.init_adapter()
 	info = f"request: {method} {path}"
 	nested_req = kwargs.pop("nested_req", False)
-	try:
-		response: Response = ncSession.adapter.request(
-			method, path, content=content, json=json, params=params, files=files, **kwargs
-		)
-	except ReadTimeout:
-		raise NextcloudException(408, info=info) from None
+	response: Response = ncSession.adapter.request(
+		method, path, data=content, json=json, params=params, files=files, **kwargs
+	)
 	if response.status_code >= 400:
 		print(loads(response.text))
 	check_error(response, info)

requirements.txt

@@ -1,5 +1,7 @@
-nc_py_api[app]==0.20.2
+nc_py_api[app]==0.22.0
 pydantic
 faster-whisper
 python-multipart
 ctranslate2==4.4.0
+niquests
+requests