Conditional pipe execution

packages/cli/src/cli.ts

@@ -19,6 +19,7 @@ import { passthrough } from "./lib/stream/operators/passthrough";
 import { toPromise } from "./lib/stream/operators/to_promise";
 import { completedCounter, exceptionCounter, sourceCounter } from "./operators/counters";
 import { exceptionHandler } from "./operators/exceptions";
+import { filterImages } from "./operators/filter";
 import { saveManifest } from "./operators/manifest";
 import { processImages } from "./operators/process";
 import { saveImages } from "./operators/save";
@@ -87,6 +88,7 @@ function createPipeline(ctx: CliContext, config: Config, manifestFile: string) {
   const paths = typeof config.input === "string" ? [config.input] : config.input;
 
   return searchForImages(paths)
+    .pipe(filterImages(config.inputFilter))
     .pipe(sourceCounter(ctx))
     .pipe(buffer(BUFFER_SIZE))
     .pipe(processImages(config.pipeline, config.concurrency))

packages/cli/src/init/config.ts

@@ -16,8 +16,19 @@ import defaultConfig from "./default_config.json";
 
 const MODULE_NAME = "ipp";
 
+/**
+ * Callback function which must return true to include the image for processing
+ */
+export type InputFilterCallback = (file: string) => boolean;
+
+/**
+ * Filter which can exclude images from processing
+ */
+export type InputFilter = (InputFilterCallback | RegExp)[] | InputFilterCallback | RegExp;
+
 export interface Config {
   input: string | string[];
+  inputFilter?: InputFilter;
   output: string;
   pipeline: Pipeline;
 

packages/cli/src/lib/stream/operators/filter.ts

@@ -0,0 +1,30 @@
+/**
+ * Image Processing Pipeline - Copyright (c) Marcus Cemes
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import { createObjectStream, Operator } from "../object_stream";
+
+export function filter<I, O>(
+  fn: (item: I) => boolean | Promise<boolean>,
+  complete?: () => void | Promise<void>
+): Operator<I, O> {
+  return (source) =>
+    createObjectStream(
+      (async function* () {
+        for await (const item of source) {
+          if (!(await fn(item))) continue;
+
+          if (Array.isArray(item)) {
+            for (const result of item) yield result;
+          } else {
+            yield item;
+          }
+        }
+
+        if (complete) await complete();
+      })()
+    );
+}

packages/cli/src/operators/filter.test.ts

@@ -0,0 +1,118 @@
+/**
+ * Image Processing Pipeline - Copyright (c) Marcus Cemes
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import { Exception } from "../../../common/src/exception";
+import { createObjectStream, ObjectStream } from "../lib/stream/object_stream";
+import { filterImages } from "./filter";
+import { isTaskSource, TaskSource } from "./types";
+
+const imgTask = (iOrFilename: number | string, root = "/root") => ({
+  root,
+  file: typeof iOrFilename === "number" ? `file-${iOrFilename}.png` : iOrFilename,
+});
+
+const awaitAsyncIterable = async (source: AsyncIterable<TaskSource | Exception>) => {
+  const resolved: {
+    tasks: TaskSource[];
+    exceptions: Exception[];
+  } = {
+    tasks: [],
+    exceptions: [],
+  };
+  for await (const tsk of source) {
+    if (isTaskSource(tsk)) resolved.tasks.push(tsk);
+    else if (tsk instanceof Exception) resolved.exceptions.push(tsk);
+  }
+  return resolved;
+};
+
+function imageObjectStreams(
+  amountOrList: number | string[] = 5
+): ObjectStream<TaskSource | Exception> {
+  return createObjectStream(imageAsyncGenerator(amountOrList));
+}
+
+async function* imageAsyncGenerator(
+  amountOrList: number | string[]
+): AsyncGenerator<TaskSource | Exception> {
+  if (typeof amountOrList === "number") {
+    for (let i = 0; i < amountOrList; i++) {
+      yield imgTask(i);
+    }
+  } else {
+    for (const name of amountOrList) {
+      yield name.startsWith("exception-") ? new Exception(name) : imgTask(name);
+    }
+  }
+}
+
+describe("filter.ts", () => {
+  const imageAmount = 10;
+
+  afterEach(() => jest.clearAllMocks());
+
+  test("should pass through images if no filter is supplied", async () => {
+    const stream = imageObjectStreams(imageAmount).pipe(filterImages());
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(imageAmount);
+  });
+
+  test("should filter via RegExp", async () => {
+    const stream = imageObjectStreams(imageAmount).pipe(filterImages(/file-[2-4]\.png$/));
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(3);
+    expect(result.tasks).toStrictEqual([imgTask(2), imgTask(3), imgTask(4)]);
+    expect(result.exceptions).toHaveLength(0);
+  });
+
+  test("should handle exceptions", async () => {
+    const stream = imageObjectStreams([
+      "exception-1",
+      "file-2.png",
+      "file-3.png",
+      "file-4.png",
+    ]).pipe(filterImages(/file-3\.png$/));
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(1);
+    expect(result.tasks).toStrictEqual([imgTask(3)]);
+    expect(result.exceptions).toHaveLength(1);
+  });
+
+  test("should filter via multiple RegExp", async () => {
+    const stream = imageObjectStreams(imageAmount).pipe(
+      filterImages([/file-[2-4]\.png$/, /file-[^3]\.png$/])
+    );
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(2);
+    expect(result.tasks).toStrictEqual([imgTask(2), imgTask(4)]);
+  });
+
+  test("should filter via callback", async () => {
+    const stream = imageObjectStreams(imageAmount).pipe(filterImages((f) => f.endsWith("4.png")));
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(1);
+    expect(result.tasks).toStrictEqual([imgTask(4)]);
+  });
+
+  test("should filter via callback & regexp", async () => {
+    const imgList = ["other.gif"]
+      .concat([1, 2, 3, 4, 5].map((i) => `file-${i}.png`))
+      .concat([1, 2, 3, 4, 5].map((i) => `file-${i}.jpg`));
+    const stream = imageObjectStreams(imgList).pipe(
+      filterImages([(f) => f.endsWith(".png"), /file-[25]\.png$/])
+    );
+
+    const result = await awaitAsyncIterable(stream);
+    expect(result.tasks).toHaveLength(2);
+    expect(result.tasks).toStrictEqual([imgTask(2), imgTask(5)]);
+  });
+});

packages/cli/src/operators/filter.ts

@@ -0,0 +1,33 @@
+/**
+ * Image Processing Pipeline - Copyright (c) Marcus Cemes
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+import { resolve } from "path";
+import { InputFilter } from "../init/config";
+import { Operator } from "../lib/stream/object_stream";
+import { filter } from "../lib/stream/operators/filter";
+import { isTaskSource } from "./types";
+
+export function filterImages<T>(inputFilter?: InputFilter): Operator<T, T> {
+  return filter((item) => {
+    if (!inputFilter || !isTaskSource(item)) {
+      return true;
+    }
+
+    if (!Array.isArray(inputFilter)) {
+      inputFilter = [inputFilter];
+    }
+
+    const file = resolve(item.root, item.file);
+    for (const f of inputFilter) {
+      if (typeof f === "function" ? !f(file) : !f.test(file)) {
+        return false;
+      }
+    }
+
+    return true;
+  });
+}