Working analysis pipeline

Dockerfile

@@ -10,5 +10,5 @@ RUN yarn build
 
 COPY . .
 
-EXPOSE 8080
+EXPOSE 8337
 CMD ['node', 'index.js']

bin/patch-deepspeech.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+DS_PATH="./node_modules/deepspeech/lib/binding/v0.10.0-alpha.3/darwin-x64"
+
+if [[ ! $OSTYPE == 'darwin'* ]]; then
+  echo "Patch currently only works on MacOS"
+  exit 0
+fi
+
+if [ -f "$DS_PATH/electron-v15.3" ]; then
+  echo "Deepspeech already fixed"
+else
+  curl -L https://github.com/vanstinator/SubD/files/7532394/electron-v15.3.zip -o "$DS_PATH/electron-v15.3.zip"
+  mkdir "$DS_PATH/electron-v15.3"
+  unzip "$DS_PATH/electron-v15.3.zip" -d "$DS_PATH"
+fi

core/main.ts

@@ -1,13 +1,11 @@
 import { app, BrowserWindow } from 'electron';
 import registerUpdater from 'core/update';
 import createServer from 'core/services/server';
-import ipcListeners from 'core/services/ipc';
+import 'core/services/ipc/listeners';
 
 declare const MAIN_WINDOW_WEBPACK_ENTRY: string;
 declare const MAIN_WINDOW_PRELOAD_WEBPACK_ENTRY: string;
 
-ipcListeners();
-
 // modify your existing createWindow() function
 function createWindow() {
   const win = new BrowserWindow({

core/services/ipc/broadcast.ts

@@ -0,0 +1,7 @@
+import { webContents } from 'electron';
+
+export default function broadcastMessage(eventName: string, ...args: any) {
+  webContents.getAllWebContents().forEach(wc => {
+    wc.send(eventName, ...args);
+  });
+}

core/services/ipc/listeners/analyze.ts

@@ -0,0 +1,10 @@
+import { ipcMain } from 'electron';
+import logger from 'electron-log';
+import queue from 'core/services/queue';
+
+const log = logger.scope('ipc_analyze');
+
+ipcMain.on('analyze.start', (event, files: { movie: BasicFile; subtitle: BasicFile }) => {
+  log.info('analyze.start', files);
+  queue.queueMedia(files.movie, files.subtitle);
+});

core/services/ipc/listeners/index.ts

@@ -0,0 +1,3 @@
+import './analyze';
+import './queue';
+import './setup';

core/services/ipc/listeners/queue.ts

@@ -0,0 +1,6 @@
+import { ipcMain } from 'electron';
+import queue from 'core/services/queue';
+
+ipcMain.on('queue.get', event => {
+  event.reply('queue.update', queue.queue);
+});

core/services/ipc/listeners/setup.ts

@@ -0,0 +1,34 @@
+import { ipcMain } from 'electron';
+import logger from 'electron-log';
+import { verifyModelFiles, downloadModelFiles, Downloader } from 'core/services/speech/download';
+
+const log = logger.scope('ipc_analyze');
+
+let downloader: Downloader;
+
+ipcMain.on('setup.startupCheck', event => {
+  const hasModelFiles = verifyModelFiles();
+  event.reply('setup.startupResult', { hasModelFiles });
+});
+
+ipcMain.on('setup.download.start', event => {
+  // start download
+  downloader = downloadModelFiles();
+  downloader.on('progress', progress => {
+    event.reply('setup.download.progress', progress);
+  });
+  downloader.on('finish', () => {
+    log.info('Download finished');
+    event.reply('setup.download.finished');
+  });
+  downloader.on('error', error => {
+    log.info('Download error', error);
+    event.reply('setup.download.error', error);
+  });
+  downloader.start();
+});
+
+ipcMain.on('setup.download.cancel', event => {
+  log.info('Cancel download');
+  downloader.cancel();
+});

core/services/queue/jobs/analyze.ts

@@ -0,0 +1,18 @@
+import detectSpeech from 'core/services/speech/detect';
+import groupByWord from 'core/services/speech/groupByWord';
+import compareToSubtitle from 'core/services/subtitles/compare';
+
+export default async function analyzeJob(
+  mediaPath: string,
+  subtitlePath: string,
+  onProgress: (percent: number) => void
+) {
+  const speech = await detectSpeech(mediaPath, onProgress);
+  const words = groupByWord(speech.text);
+  if (words) {
+    compareToSubtitle(subtitlePath, words);
+  }
+  return {
+    score: 100
+  };
+}

core/services/queue/queue.ts

@@ -1,6 +1,11 @@
 import { v4 as uuid } from 'uuid';
+import logger from 'electron-log';
+import broadcastMessage from 'core/services/ipc/broadcast';
+import analyzeJob from './jobs/analyze';
 
-const MAX_SIZE = 100;
+const log = logger.scope('queue');
+
+const MAX_SIZE = 10;
 
 enum QUEUE_STATUS {
   COMPLETE = 'complete',
@@ -10,17 +15,21 @@ enum QUEUE_STATUS {
 }
 
 interface QueueItem {
+  data?: any;
   id: string;
   media: BasicFile;
   progress: number;
   status: QUEUE_STATUS;
+  error?: string;
   subtitle: BasicFile;
 }
 
 class JobQueue {
+  activeIndex: number;
   queue: QueueItem[];
 
   constructor() {
+    this.activeIndex = -1;
     this.queue = [];
   }
 
@@ -35,6 +44,50 @@ class JobQueue {
       progress: 0,
       status: QUEUE_STATUS.QUEUED
     });
+
+    this.postUpdate();
+
+    if (this.activeIndex < 0) {
+      this.start();
+    }
+  }
+
+  postUpdate() {
+    broadcastMessage('queue.update', this.queue);
+  }
+
+  async start() {
+    this.activeIndex = this.queue.findIndex(item => item.status === QUEUE_STATUS.QUEUED);
+    log.debug('this.activeIndex', this.activeIndex);
+    if (this.activeIndex < 0) {
+      log.warn('No jobs in queue');
+      return;
+    }
+    const activeItem = this.queue[this.activeIndex];
+    activeItem.status = QUEUE_STATUS.PROCESSING;
+    this.postUpdate();
+
+    try {
+      await this.runJob();
+      activeItem.status = QUEUE_STATUS.COMPLETE;
+      activeItem.progress = 1;
+      this.postUpdate();
+    } catch (error) {
+      activeItem.status = QUEUE_STATUS.ERROR;
+    }
+    this.postUpdate();
+  }
+
+  async runJob() {
+    // TODO: Edit this to allow for other types of jobs?
+    this.queue[this.activeIndex].data = await analyzeJob(
+      this.queue[this.activeIndex].media.path,
+      this.queue[this.activeIndex].subtitle.path,
+      percent => {
+        this.queue[this.activeIndex].progress = percent;
+        this.postUpdate();
+      }
+    );
   }
 }
 

core/services/server/index.ts

@@ -1,7 +1,7 @@
 import express from 'express';
 import logger from 'electron-log';
 
-const PORT = process.env.PORT || 8080;
+const PORT = process.env.PORT || 8337;
 const log = logger.scope('express');
 const app = express();
 

core/services/speech/audio.ts

@@ -0,0 +1,51 @@
+import { exec, spawn } from 'child_process';
+import pathToFfmpeg from 'ffmpeg-static';
+import { Readable } from 'stream';
+
+const DURATION_REGEX = /Duration:\s(\d{2}):(\d{2}):([\d.]+)/;
+
+console.log(pathToFfmpeg);
+
+export function getDuration(inputFile: string): Promise<number> {
+  return new Promise((resolve, reject) => {
+    exec(`"${pathToFfmpeg}" -hide_banner -i "${inputFile}"`, (error, stdout, stderr) => {
+      const matches = stderr.match(DURATION_REGEX);
+      if (matches?.length === 4) {
+        const [, hours, minutes, seconds] = matches;
+        const duration = parseInt(hours, 10) * 3600 + parseInt(minutes, 10) * 60 + parseFloat(seconds);
+        return resolve(duration);
+      }
+      reject('Could not determine media duration');
+    });
+  });
+}
+
+export default function extractAudio(inputFile: string, sampleRate: number): Readable {
+  const child = spawn(pathToFfmpeg, [
+    '-i',
+    inputFile,
+    '-hide_banner',
+    '-loglevel',
+    'error',
+    '-ar',
+    sampleRate.toString(),
+    '-ac',
+    '1',
+    '-f',
+    's16le',
+    '-acodec',
+    'pcm_s16le',
+    'pipe:1'
+  ]);
+  let ffmpegOutput = '';
+  child.stderr.on('data', (data: Buffer) => {
+    ffmpegOutput += data.toString('utf-8');
+  });
+  child.on('exit', code => {
+    if (code !== 0) {
+      console.error(ffmpegOutput);
+      throw new Error(`ffmpeg exited with code: ${code}`);
+    }
+  });
+  return child.stdout;
+}