Analyze large fingerprint collection

cli/cli/handlers/finder.py

@@ -16,8 +16,8 @@ def local_matches(self):
         from winnow.pipeline.generate_local_matches import generate_local_matches
         from winnow.utils.files import scan_videos
 
-        configure_logging_cli()
         config = self._pipeline.config
+        configure_logging_cli(config.logging)
 
         videos = scan_videos(config.sources.root, "**", extensions=config.sources.extensions)
         generate_local_matches(files=videos, pipeline=self._pipeline)
@@ -26,7 +26,8 @@ def remote_matches(self, repo: Optional[str] = None, contributor: Optional[str]
         """Find matches between local files and remote fingerprints."""
         from winnow.pipeline.generate_remote_matches import generate_remote_matches
 
-        configure_logging_cli()
+        config = self._pipeline.config
+        configure_logging_cli(config.logging)
 
         if repo is not None:
             repo = str(repo)

cli/cli/handlers/pipeline.py

@@ -16,7 +16,7 @@ def all(self):
         from winnow.pipeline.extract_exif import extract_exif
         from winnow.pipeline.pipeline_context import PipelineContext
 
-        configure_logging_cli()
+        configure_logging_cli(self._config.logging)
 
         # Resolve list of video files from the directory
         absolute_root = os.path.abspath(self._config.sources.root)
@@ -25,4 +25,4 @@ def all(self):
         pipeline_context = PipelineContext(self._config)
         generate_local_matches(files=videos, pipeline=pipeline_context)
         detect_scenes(files=videos, pipeline=pipeline_context)
-        extract_exif(None, pipeline=pipeline_context)
+        extract_exif(videos, pipeline=pipeline_context)

db/access/files.py

@@ -517,13 +517,6 @@ def query_local_files(session: Session, path_hash_pairs) -> Query:
         query = query.filter(tuple_(Files.file_path, Files.sha256).in_(tuple(path_hash_pairs)))
         return query
 
-    @staticmethod
-    def query_local_file_ids(session: Session, path_hash_pairs) -> Query:
-        """Query local files by (path, hash) pairs."""
-        query = session.query(Files.id).filter(Files.contributor == None)  # noqa: E711
-        query = query.filter(tuple_(Files.file_path, Files.sha256).in_(tuple(path_hash_pairs)))
-        return query
-
     @staticmethod
     def query_remote_files(session: Session, repository_name: str = None, contributor_name: str = None) -> Query:
         """Query remote signatures from database."""

db/schema.py

@@ -318,3 +318,23 @@ class FileFilterPreset(Base):
     name = Column(String(100), nullable=False, unique=True)
     # Any filter data as JSON blob
     filters = Column(JSON, nullable=False)
+
+
+class TaskLogRecord(Base):
+    """Task execution log.
+
+    Motivation
+    ----------
+    Sometimes there is no way to determine whether the task is already completed just by looking
+    at the results alone. For example if template-matching is performed and no matches was found
+    there will be zero ``TemplateMatches`` in the database. So the results before and after the
+    run will be identical. Thus, some indication that the task was successfully executed is
+    needed. ``TaskLogRecord`` fills this gap.
+    """
+
+    __tablename__ = "task_logs"
+
+    id = Column(Integer, primary_key=True)
+    task_name = Column(String(100), nullable=False, unique=False)
+    timestamp = Column(DateTime, nullable=False)
+    details = Column(JSON)

default.config.yaml

@@ -1,5 +1,7 @@
 sources:
   root: data/
+  hash_mode: file
+  hash_cache: data/representations/hashes
 
   extensions:
     - mp4
@@ -12,7 +14,6 @@ sources:
 repr:
   directory: data/representations
   storage_type: detect
-  hash_mode: file
 
 processing:
   frame_sampling: 1
@@ -32,7 +33,7 @@ database:
   uri: postgresql://postgres:admin@postgres:5432/videodeduplicationdb
 
 templates:
-  source_path: data/templates/test-group/CCSI Object Recognition External/
+  source_path: data/templates/
   distance: 0.07
   distance_min: 0.05
 

docker-compose.yml

@@ -84,6 +84,7 @@ services:
       CELERY_BROKER: "redis://redis:6379/0"
       CELERY_RESULT_BACKEND: "redis://redis:6379/0"
       TASK_LOG_DIRECTORY: "/project/pipeline-logs"
+      LUIGI_CONFIG_PATH: "/project/config/luigi.cfg"
     volumes:
       # Set the BENETECH_DATA_LOCATION environment variable to the path
       # on your host machine where you placed the source data

environment-gpu.yaml

@@ -19,6 +19,12 @@ dependencies:
   - pysoundfile
   - h5py==2.9.0
   - ffmpeg
+  - numba
+  - pynndescent
+  - luigi
+  - tabulate
+  - umap-learn
+  - networkit
   - pip
   - pip:
       - lmdb
@@ -42,8 +48,9 @@ dependencies:
       - yt-dlp
       - dacite
       - deprecation
+      - trimap
+      - pacmap
       - torch
       - grpcio==1.43.0
       - grpcio-tools==1.43.0
 
-prefix: C:\ProgramData\Anaconda3\envs\winnow

environment.yaml

@@ -20,6 +20,12 @@ dependencies:
   - pysoundfile
   - h5py==2.9.0
   - ffmpeg
+  - numba
+  - pynndescent
+  - luigi
+  - tabulate
+  - umap-learn
+  - networkit
   - pip
   - pip:
       - lmdb
@@ -43,6 +49,8 @@ dependencies:
       - yt-dlp
       - dacite
       - deprecation
+      - trimap
+      - pacmap
       - torch
       - grpcio==1.43.0
       - grpcio-tools==1.43.0

extract_exif.py

@@ -1,19 +1,18 @@
+import logging.config
+
 import click
+import luigi
 
-from winnow.pipeline.extract_exif import extract_exif
-from winnow.pipeline.pipeline_context import PipelineContext
+from winnow.pipeline.luigi.exif import ExifTask
 from winnow.utils.config import resolve_config
-from winnow.utils.logging import configure_logging_cli
 
 
 @click.command()
 @click.option("--config", "-cp", help="path to the project config file", default=None)
 def main(config):
-    logger = configure_logging_cli()
-    logger.info("Loading config file")
     config = resolve_config(config_path=config)
-
-    extract_exif(videos=None, pipeline=PipelineContext(config))
+    logging.config.fileConfig("./logging.conf")
+    luigi.build([ExifTask(config=config)], local_scheduler=True, workers=1)
 
 
 if __name__ == "__main__":

extract_features.py

@@ -1,13 +1,17 @@
+import logging.config
 import os
 
 import click
+import luigi
 
-from winnow.pipeline.extract_video_signatures import extract_video_signatures
-from winnow.pipeline.pipeline_context import PipelineContext
-from winnow.pipeline.store_database_signatures import store_database_signatures
+from winnow.pipeline.luigi.exif import ExifTask, ExifFileListFileTask
+from winnow.pipeline.luigi.signatures import (
+    SignaturesTask,
+    DBSignaturesTask,
+    SignaturesByPathListFileTask,
+    DBSignaturesByPathListFileTask,
+)
 from winnow.utils.config import resolve_config
-from winnow.utils.files import scan_videos, scan_videos_from_txt
-from winnow.utils.logging import configure_logging_cli
 
 
 @click.command()
@@ -36,19 +40,29 @@
     is_flag=True,
 )
 def main(config, list_of_files, frame_sampling, save_frames):
-    logger = configure_logging_cli()
-    logger.info("Loading config file")
     config = resolve_config(config_path=config, frame_sampling=frame_sampling, save_frames=save_frames)
+    logging.config.fileConfig("./logging.conf")
 
-    logger.info("Searching for Dataset Video Files")
     if list_of_files is None:
-        videos = scan_videos(config.sources.root, "**", extensions=config.sources.extensions)
+        luigi.build(
+            [
+                ExifTask(config=config),
+                SignaturesTask(config=config),
+                DBSignaturesTask(config=config),
+            ],
+            local_scheduler=True,
+            workers=1,
+        )
     else:
-        videos = scan_videos_from_txt(list_of_files, extensions=config.sources.extensions)
-
-    pipeline = PipelineContext(config)
-    extract_video_signatures(files=videos, pipeline=pipeline)
-    store_database_signatures(files=videos, pipeline=pipeline)
+        luigi.build(
+            [
+                ExifFileListFileTask(config=config, path_list_file=list_of_files),
+                SignaturesByPathListFileTask(config=config, path_list_file=list_of_files),
+                DBSignaturesByPathListFileTask(config=config, path_list_file=list_of_files),
+            ],
+            local_scheduler=True,
+            workers=1,
+        )
 
 
 if __name__ == "__main__":

generate_matches.py

@@ -1,13 +1,11 @@
+import logging.config
 import os
 
 import click
+import luigi
 
-from winnow.pipeline.detect_scenes import detect_scenes
-from winnow.pipeline.generate_local_matches import generate_local_matches
-from winnow.pipeline.pipeline_context import PipelineContext
+from winnow.pipeline.luigi.matches import MatchesReportTask, MatchesByFileListTask
 from winnow.utils.config import resolve_config
-from winnow.utils.files import scan_videos, scan_videos_from_txt
-from winnow.utils.logging import configure_logging_cli
 
 
 @click.command()
@@ -36,19 +34,15 @@
     is_flag=True,
 )
 def main(config, list_of_files, frame_sampling, save_frames):
-    logger = configure_logging_cli()
-    logger.info("Loading config file")
     config = resolve_config(config_path=config, frame_sampling=frame_sampling, save_frames=save_frames)
+    logging.config.fileConfig("./logging.conf")
 
-    logger.info("Searching for Dataset Video Files")
     if list_of_files is None:
-        videos = scan_videos(config.sources.root, "**", extensions=config.sources.extensions)
+        luigi.build([MatchesReportTask(config=config)], local_scheduler=True, workers=1)
     else:
-        videos = scan_videos_from_txt(list_of_files, extensions=config.sources.extensions)
-
-    pipeline = PipelineContext(config)
-    generate_local_matches(files=videos, pipeline=pipeline)
-    detect_scenes(files=videos, pipeline=pipeline)
+        luigi.build(
+            [MatchesByFileListTask(config=config, path_list_file=list_of_files)], local_scheduler=True, workers=1
+        )
 
 
 if __name__ == "__main__":

generate_remote_matches.py

@@ -1,11 +1,11 @@
+import logging.config
 import os
 
 import click
+import luigi
 
-from winnow.pipeline.generate_remote_matches import generate_remote_matches
-from winnow.pipeline.pipeline_context import PipelineContext
+from winnow.pipeline.luigi.matches import RemoteMatchesTask
 from winnow.utils.config import resolve_config
-from winnow.utils.logging import configure_logging_cli
 
 
 @click.command()
@@ -16,11 +16,6 @@
     help="remote repository name",
     default=None,
 )
-@click.option(
-    "--contributor",
-    help="remote contributor name",
-    default=None,
-)
 @click.option(
     "--frame-sampling",
     "-fs",
@@ -38,11 +33,10 @@
     default=None,
     is_flag=True,
 )
-def main(repo, contributor, config, frame_sampling, save_frames):
-    logger = configure_logging_cli()
-    logger.info("Loading config file")
+def main(repo, config, frame_sampling, save_frames):
     config = resolve_config(config_path=config, frame_sampling=frame_sampling, save_frames=save_frames)
-    generate_remote_matches(repository_name=repo, contributor_name=contributor, pipeline=PipelineContext(config))
+    logging.config.fileConfig("./logging.conf")
+    luigi.build([RemoteMatchesTask(config=config, repository_name=repo)], local_scheduler=True, workers=1)
 
 
 if __name__ == "__main__":