Add metadata helper to generate metadata and keyframes preview image

.github/workflows/PR-test.yml

@@ -11,13 +11,13 @@ jobs:
     if: contains(github.event.pull_request.labels.*.name, 'Ready To Test')
     strategy:
       matrix:
-        node-version: [16.x]
+        node-version: [18.x]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
       - name: Use Node.js ${{ matrix.node-version }}
-        uses: actions/setup-node@v2
+        uses: actions/setup-node@v3
         with:
           node-version: ${{ matrix.node-version }}
       - name: Add FFmpeg

.github/workflows/checkin.yml

@@ -12,13 +12,13 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        node-version: [16.x]
+        node-version: [18.x]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Use Node.js ${{ matrix.node-version }}
-      uses: actions/setup-node@v2
+      uses: actions/setup-node@v3
       with:
         node-version: ${{ matrix.node-version }}
     - name: Add FFmpeg

ava.config.js

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2023 IROHA LAB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+const {familySync, GLIBC}  = require('detect-libc');
+
+module.exports = {
+    extensions: [
+        "ts"
+    ],
+    files: [
+        "src/utils/*.spec.ts",
+        "src/services/*.spec.ts",
+        "src/processors/*.spec.ts",
+        "src/api-service/controller/*.spec.ts",
+        "src/JobManager/*.spec.ts",
+        "src/domains/*.spec.ts"
+    ],
+    require: [
+        "ts-node/register"
+    ],
+    verbose: true,
+    workerThreads: familySync() !== GLIBC
+};

package.json

@@ -4,7 +4,7 @@
    "description": "Video Process for mira project",
    "main": "index.js",
    "scripts": {
-      "test": "$(npm bin)/ava",
+      "test": "ava",
       "start:jobscheduler": "cross-env START_AS=JOB_SCHEDULER ts-node src/main.ts",
       "start:jobexecutor": "cross-env START_AS=JOB_EXECUTOR ts-node src/main.ts",
       "start:jobexecutor:meta": "cross-env START_AS=JOB_EXECUTOR EXEC_MODE=META_MODE ts-node src/main.ts",
@@ -28,7 +28,7 @@
    "license": "Apache-2.0",
    "dependencies": {
       "@cloudamqp/amqp-client": "^2.1.0",
-      "@irohalab/mira-shared": "^3.5.0",
+      "@irohalab/mira-shared": "^3.7.0",
       "@mikro-orm/core": "^5.5.3",
       "@mikro-orm/migrations": "^5.5.3",
       "@mikro-orm/postgresql": "^5.5.3",
@@ -39,6 +39,7 @@
       "cors": "^2.8.5",
       "esprima": "^4.0.1",
       "express": "^4.17.3",
+      "fast-average-color-node": "^2.6.0",
       "fontkit": "^2.0.2",
       "inversify": "^6.0.1",
       "inversify-binding-decorators": "^4.0.0",
@@ -66,6 +67,7 @@
       "@types/tail": "^2.2.1",
       "ava": "^4.2.0",
       "cross-env": "^7.0.3",
+      "detect-libc": "^2.0.2",
       "rimraf": "^3.0.2",
       "supertest": "^6.1.4",
       "ts-node": "^10.7.0",
@@ -74,22 +76,5 @@
    },
    "engines": {
       "node": ">=16.0.0"
-   },
-   "ava": {
-      "extensions": [
-         "ts"
-      ],
-      "files": [
-         "src/utils/*.spec.ts",
-         "src/services/*.spec.ts",
-         "src/processors/*.spec.ts",
-         "src/api-service/controller/*.spec.ts",
-         "src/JobManager/*.spec.ts",
-         "src/domains/*.spec.ts"
-      ],
-      "require": [
-         "ts-node/register"
-      ],
-      "verbose": true
    }
 }

src/JobExecutor.ts

@@ -45,6 +45,7 @@ import { randomUUID } from 'crypto';
 import { getStdLogger } from './utils/Logger';
 import { JobCleaner } from './JobManager/JobCleaner';
 import { JobType } from './domains/JobType';
+import { VideoOutputMetadata } from './domains/VideoOutputMetadata';
 
 const logger = getStdLogger();
 
@@ -218,15 +219,12 @@ export class JobExecutor implements JobApplication {
         this.currentJM.events.on(JobManager.EVENT_JOB_FINISHED, async (finishedJobId: string) => {
             // find all output path
             try {
-                if (job.jobMessage.jobType === JobType.META_JOB) {
-                    await this.sendNoNeedToProcessMessage(job);
-                } else {
-                    const outputVertices = await this._databaseService.getVertexRepository().getOutputVertices(finishedJobId);
-                    const outputPathList = outputVertices.map(vx => {
-                        return vx.outputPath;
-                    });
-                    await this.notifyFinished(job, outputPathList);
-                }
+                job = await this._databaseService.getJobRepository().findOne({id:finishedJobId});
+                const outputVertices = await this._databaseService.getVertexRepository().getOutputVertices(finishedJobId);
+                const outputPathList = outputVertices.map(vx => {
+                    return vx.outputPath;
+                });
+                await this.notifyFinished(job, outputPathList);
                 await this.finalizeJM();
             } catch (err) {
                 logger.error(err);
@@ -263,27 +261,44 @@ export class JobExecutor implements JobApplication {
     //     return normalizedOutputPath;
     // }
 
-    private async notifyFinished(job: Job, outputFilePathList: string[]): Promise<void> {
+    private async notifyFinished(job: Job, outputPathList: string[]): Promise<void> {
         const msg = new VideoManagerMessage();
         msg.id = randomUUID();
-        msg.processedFiles = outputFilePathList.map((outputFilePath) => {
+        msg.processedFiles = outputPathList.map((outputPath) => {
             const remoteFile = new RemoteFile();
-            remoteFile.filename = basename(outputFilePath);
-            remoteFile.fileLocalPath = outputFilePath;
+            remoteFile.filename = basename(outputPath);
+            remoteFile.fileLocalPath = outputPath;
             remoteFile.fileUri = this._configManager.getFileUrl(remoteFile.filename, job.jobMessageId);
             return remoteFile;
         });
+        const thumbnailPath = new RemoteFile();
+        thumbnailPath.filename = basename(job.metadata.thumbnailPath);
+        thumbnailPath.fileLocalPath = job.metadata.thumbnailPath;
+        thumbnailPath.fileUri = this._configManager.getFileUrl(thumbnailPath.filename, job.jobMessageId);
+        const keyframeImagePathList = job.metadata.keyframeImagePathList.map((p) => {
+            const keyframeImagePath = new RemoteFile();
+            keyframeImagePath.filename = basename(p);
+            keyframeImagePath.fileLocalPath = p;
+            keyframeImagePath.fileUri = this._configManager.getFileUrl(keyframeImagePath.filename, job.jobMessageId);
+            return keyframeImagePath;
+        });
+        msg.metadata = Object.assign({}, job.metadata, {thumbnailPath, keyframeImagePathList});
         msg.jobExecutorId = this.id;
         msg.bangumiId = job.jobMessage.bangumiId;
         msg.videoId = job.jobMessage.videoId;
         msg.downloadTaskId = job.jobMessage.downloadTaskId;
-        msg.isProcessed = true;
+        msg.isProcessed = job.jobMessage.jobType === JobType.NORMAL_JOB;
         if (await this._rabbitmqService.publish(VIDEO_MANAGER_EXCHANGE, VIDEO_MANAGER_GENERAL, msg)) {
             // TODO: do something
             logger.info('TODO: after published to VIDEO_MANAGER_EXCHANGE');
         }
     }
 
+    /**
+     * @deprecated
+     * @param job
+     * @private
+     */
     private async sendNoNeedToProcessMessage(job: Job) {
         const vmMsg = new VideoManagerMessage();
         vmMsg.id = randomUUID();

src/JobManager/JobManager.spec.ts

@@ -33,6 +33,8 @@ import { JobManager } from './JobManager';
 import { FakeJobRepository } from '../test-helpers/FakeJobRepository';
 import { FakeVertexRepository } from '../test-helpers/FakeVertexRepository';
 import { FakeSentry } from '@irohalab/mira-shared/test-helpers/FakeSentry';
+import { FakeJobMetadataHelper } from '../test-helpers/FakeJobMetadataHelper';
+import { JobMetadataHelper } from './JobMetadataHelper';
 
 type Cxt = { container: Container };
 
@@ -45,6 +47,7 @@ test.beforeEach((t) => {
     container.bind<VertexManager>(TYPES_VM.VertexManager).to(FakeVertexManager);
     container.bind<JobManager>(JobManager).toSelf();
     container.bind<interfaces.Factory<VertexManager>>(TYPES_VM.VertexManagerFactory).toAutoFactory<VertexManager>(FakeVertexManager);
+    container.bind<JobMetadataHelper>(TYPES_VM.JobMetadataHelper).to(FakeJobMetadataHelper).inSingletonScope();
     container.bind<Sentry>(TYPES.Sentry).to(FakeSentry);
     const configManager = container.get<ConfigManager>(TYPES.ConfigManager);
     (configManager as FakeConfigManager).profilePath = join(projectRoot, 'temp/job-manager');
@@ -83,7 +86,7 @@ test.serial('Should run the job and manage lifecycle of a job', async (t) => {
             resolve(true);
         });
         jm.events.on(JobManager.EVENT_JOB_FAILED, (jobId: string) => {
-            reject(false);
+            reject('Job Failed');
         });
         vx.completeAllVertices();
     });

src/JobManager/JobManager.ts

@@ -29,6 +29,7 @@ import { EventEmitter } from 'events';
 import { EVENT_VERTEX_FAIL, TERMINAL_VERTEX_FINISHED, VERTEX_MANAGER_LOG, VertexManager } from './VertexManager';
 import { FileManageService } from '../services/FileManageService';
 import { JobMessage } from '../domains/JobMessage';
+import { JobMetadataHelper } from './JobMetadataHelper';
 
 @injectable()
 export class JobManager {
@@ -44,6 +45,7 @@ export class JobManager {
                 @inject(TYPES.ConfigManager) private _configManager: ConfigManager,
                 @inject(TYPES_VM.VertexManagerFactory) private _vmFactory: interfaces.AutoFactory<VertexManager>,
                 private _fileManager: FileManageService,
+                @inject(TYPES_VM.JobMetadataHelper) private _metaDataHelper: JobMetadataHelper,
                 @inject(TYPES.Sentry) private _sentry: Sentry) {
     }
 
@@ -100,6 +102,7 @@ export class JobManager {
                 this._jobLogger.error(err);
                 this._jobLogger.info(LOG_END_FLAG);
                 this._sentry.capture(err);
+                this.events.emit(JobManager.EVENT_JOB_FAILED, this._job.id);
             }
         });
 
@@ -111,6 +114,9 @@ export class JobManager {
                     return vertexMap[vertexId].status === VertexStatus.Finished;
                 });
                 if (allVerticesFinished) {
+                    this._job.status = JobStatus.MetaData;
+                    this._job = await jobRepo.save(this._job) as Job;
+                    this._job.metadata = await this._metaDataHelper.processMetaData(vertexMap, this._jobLogger);
                     this._job.status = JobStatus.Finished;
                     this._job.finishedTime = new Date();
                     this._job = await jobRepo.save(this._job) as Job;
@@ -122,6 +128,7 @@ export class JobManager {
                 this._jobLogger.error(error);
                 this._jobLogger.info(LOG_END_FLAG);
                 this._sentry.capture(error);
+                this.events.emit(JobManager.EVENT_JOB_FAILED, this._job.id);
             }
         });
 

src/JobManager/JobMetaDataHelperImpl.spec.ts

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2023 IROHA LAB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import 'reflect-metadata';
+import { Container } from 'inversify';
+import test from 'ava';
+import { Sentry, TYPES } from '@irohalab/mira-shared';
+import { FakeSentry } from '@irohalab/mira-shared/test-helpers/FakeSentry';
+import { join } from 'path';
+import { projectRoot } from '../test-helpers/helpers';
+import { JobMetadataHelperImpl } from './JobMetadataHelperImpl';
+import { Vertex } from '../entity/Vertex';
+import { copyFile, mkdir, readdir, stat, unlink } from 'fs/promises';
+import { getStdLogger } from '../utils/Logger';
+import { JobMetadataHelper } from './JobMetadataHelper';
+import { TYPES_VM } from '../TYPES';
+
+type Cxt = { container: Container };
+const testVideoDir = join(projectRoot, 'tests');
+const videoTemp = join(projectRoot, 'temp/metadata');
+
+test.before(async (t) => {
+    try {
+        await stat(videoTemp);
+    } catch (err) {
+        if (err.code === 'ENOENT') {
+            await mkdir(videoTemp);
+        } else {
+            throw err;
+        }
+    }
+});
+
+test.beforeEach((t) => {
+    const context = t.context as Cxt;
+    const container = new Container({ autoBindInjectable: true });
+    context.container = container;
+    container.bind<Sentry>(TYPES.Sentry).to(FakeSentry).inSingletonScope();
+    container.bind<JobMetadataHelper>(TYPES_VM.JobMetadataHelper).to(JobMetadataHelperImpl).inSingletonScope();
+});
+
+test.afterEach(async (t) => {
+    const files = await readdir(videoTemp);
+    for (const file of files) {
+        await unlink(join(videoTemp, file));
+    }
+});
+
+test('test generate metadata', async (t) => {
+    const context = t.context as Cxt;
+    const container = context.container;
+    const jobMetaDataHelper = container.get(JobMetadataHelperImpl);
+    const vertex = new Vertex();
+    const testVideoFilename = 'test-video-1.mp4';
+    vertex.outputPath = join(videoTemp, testVideoFilename);
+    await copyFile(join(testVideoDir, testVideoFilename), vertex.outputPath);
+    const verticesMap = {[vertex.id]: vertex};
+    const logger = getStdLogger();
+    const metadata = await jobMetaDataHelper.processMetaData(verticesMap, logger);
+    t.true(!!metadata, 'metadata should not be null');
+    t.true(/^#[0-9a-f]{6}/i.test(metadata.dominantColorOfThumbnail), 'dominant color should be a string');
+    t.true(Number.isInteger(metadata.duration), `duration should be integer, ${metadata.duration}`);
+    t.true(metadata.frameWidth > 0, `frameSize should be greater than 0, ${metadata.frameWidth}`);
+    t.true(metadata.height > 0, `height should be greater than 0 ${metadata.height}`);
+    const f = await stat(metadata.keyframeImagePathList[0]);
+    t.true(f.isFile() && f.size > 0, 'keyframeImage should exists');
+});

src/JobManager/JobMetadataHelper.ts

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2023 IROHA LAB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { VideoOutputMetadata } from '../domains/VideoOutputMetadata';
+import { VertexMap } from '../domains/VertexMap';
+import pino from 'pino';
+
+export interface JobMetadataHelper {
+    processMetaData(vertexMap: VertexMap, jobLogger: pino.Logger): Promise<VideoOutputMetadata>;
+    generatePreviewImage(videoPath: string, metaData: VideoOutputMetadata, jobLogger: pino.Logger): Promise<void>;
+}