Compare commits
2 commits
e8bb70feb0
...
0a9b618858
| Author | SHA1 | Date | |
|---|---|---|---|
| 0a9b618858 | |||
| 55239f71a0 |
6 changed files with 128 additions and 15 deletions
|
|
@ -12,12 +12,19 @@ import subprocess
|
|||
import threading
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
|
||||
from faster_whisper import WhisperModel
|
||||
# Windows with a GPU: where the CUDA libraries (cuBLAS, cuDNN) are, separated by ';'.
|
||||
for _d in filter(None, os.environ.get('WHISPER_DLL_DIRS', '').split(';')):
|
||||
os.add_dll_directory(_d)
|
||||
os.environ['PATH'] = _d + os.pathsep + os.environ['PATH']
|
||||
|
||||
import numpy as np # noqa: E402
|
||||
from fastapi import FastAPI, File, Form, HTTPException, UploadFile # noqa: E402
|
||||
from faster_whisper import WhisperModel # noqa: E402
|
||||
|
||||
MODEL = os.environ.get('WHISPER_MODEL', 'large-v3-turbo')
|
||||
THREADS = int(os.environ.get('WHISPER_THREADS', '4'))
|
||||
DEVICE = os.environ.get('WHISPER_DEVICE', 'cpu') # cpu or cuda
|
||||
COMPUTE = os.environ.get('WHISPER_COMPUTE', 'int8') # int8_float16 on a GPU
|
||||
IDLE = int(os.environ.get('WHISPER_IDLE', '600'))
|
||||
CACHE = os.environ.get('WHISPER_CACHE', '/models')
|
||||
|
||||
|
|
@ -28,7 +35,7 @@ state = {'model': None, 'last': 0.0}
|
|||
|
||||
def model():
|
||||
if state['model'] is None:
|
||||
state['model'] = WhisperModel(MODEL, device='cpu', compute_type='int8', cpu_threads=THREADS,
|
||||
state['model'] = WhisperModel(MODEL, device=DEVICE, compute_type=COMPUTE, cpu_threads=THREADS,
|
||||
download_root=CACHE)
|
||||
return state['model']
|
||||
|
||||
|
|
@ -56,7 +63,7 @@ def decode(data: bytes) -> np.ndarray:
|
|||
|
||||
@app.get('/health')
|
||||
def health():
|
||||
return {'ok': True, 'model': MODEL, 'loaded': state['model'] is not None}
|
||||
return {'ok': True, 'model': MODEL, 'device': DEVICE, 'loaded': state['model'] is not None}
|
||||
|
||||
|
||||
@app.post('/v1/audio/transcriptions')
|
||||
|
|
|
|||
|
|
@ -19,8 +19,8 @@ only way to start a transcription is the `POST` below, sent when the user presse
|
|||
The **server** transcribes. It already downloads and decodes episodes for skipping silence
|
||||
(`podcasts/silence.py`), and browsers and the WebView can't fetch podcast audio (CORS). It cuts
|
||||
the audio at pauses and sends the pieces in parallel to a Whisper instance with an
|
||||
OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (a container
|
||||
beside diora-web, `DIORA_WHISPER_URL`), or any instance the user enters. YouTube episodes are
|
||||
OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (`DIORA_WHISPER_URL`),
|
||||
or any instance the user enters. YouTube episodes are
|
||||
not transcribed: the server never fetches their sound (docs/youtube.md).
|
||||
|
||||
The **device** writes the article with the user's own key (docs/ai.md). The key never reaches the
|
||||
|
|
@ -242,3 +242,21 @@ using the metadata, and don't quote garbled passages.
|
|||
```
|
||||
|
||||
Prompt (user turn): the metadata block, then `Transkript:` and the text.
|
||||
|
||||
## The built-in instance (since 2026-11-09)
|
||||
|
||||
Whisper does not fit on the Strato server: with about 2 GB of memory free, large-v3-turbo needed
|
||||
more than 2.2 GB and even `small` broke a 1.2 GB limit, and earlyoom then killed other services.
|
||||
So the built-in instance runs on the user's PC (multi11, GTX 1660 Ti) and diora-prod reaches it over
|
||||
the tailnet: `DIORA_WHISPER_URL=http://100.64.0.3:8766/v1` in `~/docker-compose/diora-prod/.env`.
|
||||
It is `deploy/whisper/server.py` (a small OpenAI-compatible server around faster-whisper,
|
||||
large-v3-turbo, int8_float16 on CUDA; one request at a time; the model is loaded on demand and freed
|
||||
after 10 idle minutes), started at logon from `F:\diora-whisper` (start.cmd via start-hidden.vbs,
|
||||
shortcut in the user's Startup folder; log `F:\diora-whisper\server.log`), listening only on the
|
||||
tailnet address. One hour of speech takes about five minutes.
|
||||
|
||||
When the PC is off, a job does not fail: the server checks the instance before downloading, and a
|
||||
connection error (there or while pieces are out) puts the job back into the queue with
|
||||
`error = "Whisper-Instanz nicht erreichbar, neuer Versuch um HH:MM"`, tried again every ten
|
||||
minutes (`WAIT_RETRY`) for up to two days after the button was pressed (`WAIT_MAX`). Only the
|
||||
built-in instance waits like this; a custom instance's connection errors fail the job as before.
|
||||
|
|
|
|||
|
|
@ -299,6 +299,41 @@ class RunTests(PipelineBase):
|
|||
self.assertIn('Whisper', job.error)
|
||||
self.assertEqual(self.fake.calls, [])
|
||||
|
||||
def test_unreachable_builtin_waits_and_tries_again(self):
|
||||
import socket
|
||||
sock = socket.socket()
|
||||
sock.bind(('127.0.0.1', 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close() # nothing listens there now
|
||||
with override_settings(WHISPER_URL=f'http://127.0.0.1:{port}/v1'):
|
||||
self.configure()
|
||||
job = self.job()
|
||||
transcribe.run(job.pk)
|
||||
job = self.reload(job)
|
||||
self.assertEqual(job.state, 'queued')
|
||||
self.assertIn('nicht erreichbar', job.error)
|
||||
self.assertGreater(transcribe._not_before[job.pk], time.time())
|
||||
self.assertEqual(self.fake.calls, []) # nothing downloaded while the instance is away
|
||||
self.assertEqual(transcribe._next_job(), transcribe.WAIT)
|
||||
transcribe._not_before[job.pk] = 0 # due again: the instance is back
|
||||
with override_settings(WHISPER_URL=self.fake.whisper):
|
||||
transcribe.run(job.pk)
|
||||
job = self.reload(job)
|
||||
self.assertEqual((job.state, job.error), ('done', ''))
|
||||
transcribe._not_before.clear()
|
||||
|
||||
def test_waiting_gives_up_after_two_days(self):
|
||||
from datetime import timedelta
|
||||
from django.utils import timezone
|
||||
with override_settings(WHISPER_URL='http://127.0.0.1:9/v1'):
|
||||
self.configure()
|
||||
job = self.job()
|
||||
TranscriptionJob.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(days=3))
|
||||
transcribe.run(job.pk)
|
||||
job = self.reload(job)
|
||||
self.assertEqual(job.state, 'error')
|
||||
self.assertIn('zwei Tage', job.error)
|
||||
|
||||
@override_settings(WHISPER_URL='')
|
||||
def test_retries_then_succeeds(self):
|
||||
self.fake.fail_first = 2
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ taken oldest first (by id). A job found `downloading` or `transcribing` when the
|
|||
just started (a restart killed it) is queued again. Cancelling deletes the job's row: the
|
||||
worker notices when it next writes progress and stops.
|
||||
"""
|
||||
import datetime
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
|
|
@ -42,10 +43,15 @@ DECODE_TIMEOUT = 3600
|
|||
OPUS_BITRATE = '32k'
|
||||
MAX_PROMPT = 500 # characters
|
||||
ALIVE_CHECK = 5 # seconds between looks whether the job was cancelled
|
||||
WAIT_RETRY = 600 # seconds between tries while the built-in instance can't be reached
|
||||
WAIT_MAX = 48 * 3600 # after this long since the button was pressed a waiting job gives up
|
||||
WAIT_IDLE = 30 # seconds the runner sleeps while every queued job is waiting
|
||||
|
||||
_lock = threading.Lock()
|
||||
_runner = None # the thread working through the queue, or None
|
||||
_resumed = False # whether jobs left over from before a restart were queued again
|
||||
_not_before = {} # job id -> time.time() before which a waiting job isn't tried again
|
||||
WAIT = 'wait' # _next_job: jobs are queued, but none is due yet
|
||||
|
||||
|
||||
class Cancelled(Exception):
|
||||
|
|
@ -56,6 +62,11 @@ class Failed(Exception):
|
|||
"""The job can't succeed; the message is shown to the user."""
|
||||
|
||||
|
||||
class Unreachable(Exception):
|
||||
"""The built-in instance can't be reached (e.g. the machine it runs on is off): the job waits
|
||||
and tries again later instead of failing."""
|
||||
|
||||
|
||||
class Permanent(Exception):
|
||||
"""A request the instance refused for good (a client error): not worth a retry."""
|
||||
|
||||
|
|
@ -85,11 +96,14 @@ def kick():
|
|||
def _next_job():
|
||||
global _runner
|
||||
with _lock:
|
||||
job_id = (TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED)
|
||||
.order_by('id').values_list('id', flat=True).first())
|
||||
if job_id is None:
|
||||
queued = list(TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED)
|
||||
.order_by('id').values_list('id', flat=True))
|
||||
if not queued:
|
||||
_runner = None # decided under the lock, so a job queued right now starts a new runner
|
||||
return job_id
|
||||
return None
|
||||
now = time.time()
|
||||
due = next((i for i in queued if _not_before.get(i, 0) <= now), None)
|
||||
return due if due is not None else WAIT
|
||||
|
||||
|
||||
def _loop():
|
||||
|
|
@ -100,6 +114,9 @@ def _loop():
|
|||
job_id = _next_job()
|
||||
if job_id is None:
|
||||
return
|
||||
if job_id == WAIT:
|
||||
time.sleep(WAIT_IDLE)
|
||||
continue
|
||||
try:
|
||||
run(job_id)
|
||||
except Exception: # run() reports its own failures; this keeps the queue going
|
||||
|
|
@ -124,6 +141,8 @@ def run(job_id):
|
|||
_write(job_id, state=TranscriptionJob.DONE, data=data, error='', finished_at=timezone.now())
|
||||
except Cancelled:
|
||||
log.info('transcription job %s cancelled', job_id)
|
||||
except Unreachable as e:
|
||||
_wait(job_id, str(e))
|
||||
except Failed as e:
|
||||
_fail(job_id, str(e))
|
||||
except Exception as e:
|
||||
|
|
@ -131,6 +150,31 @@ def run(job_id):
|
|||
_fail(job_id, f'{type(e).__name__}: {e}')
|
||||
|
||||
|
||||
def _wait(job_id, reason):
|
||||
"""Back into the queue to try again in WAIT_RETRY seconds, or give up when it waited too long."""
|
||||
job = TranscriptionJob.objects.filter(pk=job_id).only('created_at').first()
|
||||
if job is None:
|
||||
return
|
||||
if (timezone.now() - job.created_at).total_seconds() > WAIT_MAX:
|
||||
_not_before.pop(job_id, None)
|
||||
_fail(job_id, 'Whisper-Instanz zwei Tage lang nicht erreichbar')
|
||||
return
|
||||
_not_before[job_id] = time.time() + WAIT_RETRY
|
||||
again = timezone.localtime(timezone.now() + datetime.timedelta(seconds=WAIT_RETRY)).strftime('%H:%M')
|
||||
log.info('transcription job %s waits for the instance (%s)', job_id, reason)
|
||||
TranscriptionJob.objects.filter(pk=job_id).update(
|
||||
state=TranscriptionJob.QUEUED, chunks_done=0, chunks_total=0, started_at=None,
|
||||
error=f'Whisper-Instanz nicht erreichbar, neuer Versuch um {again}')
|
||||
|
||||
|
||||
def _probe(url):
|
||||
"""Unreachable unless something answers at the built-in instance's address."""
|
||||
try:
|
||||
requests.get(url.rsplit('/audio/transcriptions', 1)[0], timeout=(5, 10)).close()
|
||||
except (requests.ConnectionError, requests.Timeout) as e:
|
||||
raise Unreachable(e.__class__.__name__)
|
||||
|
||||
|
||||
def _fail(job_id, message):
|
||||
TranscriptionJob.objects.filter(pk=job_id).update(
|
||||
state=TranscriptionJob.ERROR, error=(message or 'failed')[:500], finished_at=timezone.now())
|
||||
|
|
@ -173,6 +217,8 @@ def _transcribe(job_id):
|
|||
episode = job.episode
|
||||
conf = for_user(episode.feed.user)
|
||||
url, headers, builtin = _target(conf)
|
||||
if builtin:
|
||||
_probe(url) # before downloading: is the instance there at all?
|
||||
options = {'model': conf.model or DEFAULT_MODEL, 'response_format': 'verbose_json'}
|
||||
if conf.language:
|
||||
options['language'] = conf.language
|
||||
|
|
@ -312,7 +358,10 @@ def _post(url, headers, builtin, name, mime, body, options):
|
|||
try:
|
||||
if builtin:
|
||||
# The operator's own instance (DIORA_WHISPER_URL) may sit on a private network.
|
||||
r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT)
|
||||
try:
|
||||
r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT)
|
||||
except requests.ConnectionError as e:
|
||||
raise Unreachable(e.__class__.__name__)
|
||||
if r.status_code >= 400:
|
||||
if r.status_code < 500 and r.status_code not in (408, 429):
|
||||
raise Permanent(f'HTTP {r.status_code}')
|
||||
|
|
@ -343,6 +392,10 @@ def _send(url, headers, builtin, name, mime, body, options, stop):
|
|||
return _post(url, headers, builtin, name, mime, body, options)
|
||||
except Permanent as e:
|
||||
raise Failed(f'Whisper: {e}')
|
||||
except Unreachable:
|
||||
if attempt >= RETRIES:
|
||||
raise
|
||||
time.sleep(RETRY_PAUSES[min(attempt, len(RETRY_PAUSES) - 1)])
|
||||
except RuntimeError as e:
|
||||
if attempt >= RETRIES:
|
||||
raise Failed(f'Whisper: {e}')
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
* diora service worker — caches the app shell for offline use.
|
||||
*/
|
||||
|
||||
const CACHE = 'diora-v170';
|
||||
const CACHE = 'diora-v171';
|
||||
const SHELL = [
|
||||
'/static/css/app.css',
|
||||
'/static/js/app/01-core.js',
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
// Copied from diora-android 7bb865c 2026-11-09: web-native/static/native-podcast.js
|
||||
// Copied from diora-android ba1e933 2026-11-09: web-native/static/native-podcast.js
|
||||
// by tools/sync_podcast_ui.py. Do not edit here: change it in diora-android
|
||||
// and copy again. The browser host is static/js/podcasts-web.js.
|
||||
|
||||
|
|
@ -4640,7 +4640,7 @@ neuesten Folgen) und die vorhandenen Ordner. Schlage nur Änderungen vor, die wi
|
|||
if (t.writing) return {text: t.writing.text, kind: 'writing'};
|
||||
const j = t.job;
|
||||
if (!trActive(j)) return null;
|
||||
const text = j.state === 'queued' ? 'Wartet …'
|
||||
const text = j.state === 'queued' ? (j.error ? 'Wartet: ' + j.error : 'Wartet …')
|
||||
: j.state === 'downloading' ? 'Lade Folge …'
|
||||
: j.chunks_total ? `Transkribiere ${j.chunks_done || 0}/${j.chunks_total} …` : 'Transkribiere …';
|
||||
return {text, kind: 'job'};
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue