Fluid Vision

Directory Structure

fluid_vision/
├── .coverage
├── poetry.lock
├── pyproject.toml
├── usage.txt
├── notebooks/
├── src/
│   └── fluid_vision/
│       ├── __init__.py
│       ├── data/
│       │   ├── __init__.py
│       │   └── frame_extractor/
│       │       ├── __init__.py
│       │       └── frame_extractor.py
│       ├── embeddings/
│       │   └── frame_embedder.py
│       ├── encoder/
│       │   └── encoder.py
│       ├── models/
│       │   ├── __init__.py
│       │   ├── embedding_models_enum.py
│       │   └── model_loader/
│       │       └── model_loader.py
│       └── utils/
│           ├── cropping.py
│           └── load_files.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_cropping.py
│   ├── test_dino_embedding.py
│   ├── test_embedding_large.py
│   ├── test_embeddings_base.py
│   ├── test_encoding.py
│   ├── test_frame_extraction_phash.py
│   └── test_frame_fps.py
└── .pytest_cache/
    ├── .gitignore
    ├── CACHEDIR.TAG
    ├── README.md
    └── v/
        └── cache/
            └── nodeids

Module Usage

1. Frame Extraction

Extract frames at a target FPS:

from fluid_vision.data.frame_extractor.frame_extractor import FrameExtractor

extractor = FrameExtractor(target_fps=5)
result = extractor.extract_frames_fps("path/to/video.mp4")
frames_dict = result["frames_dict"]  # {frame_number: PIL.Image}

Extract frames by perceptual hash (skip duplicates):

extractor = FrameExtractor(frame_skip=10)
result = extractor.extract_frames_by_hash("path/to/video.mp4")
frames_dict = result["frames_dict"]

Extract frames with cropping:

extractor = FrameExtractor(target_fps=5, cropping={"quadrant": 2}, skip_similar=0.5)
result = extractor.extract_frames_fps("path/to/video.mp4")
cropped_frames = result["frames_dict"]

# Optionally filter frames based on average hash function
# By default all duplicate frames are skipped
filtered_frames_dict = extractor.filter_frames_by_hash(frames_dict)["filtered_frames_dict"]

2. Loading Images and Pickle Files

Load a dictionary from a pickle file:

from fluid_vision.utils.load_files import load_dict_from_pickle

data = load_dict_from_pickle("path/to/file.pkl")

Load all images from a folder:

from fluid_vision.utils.load_files import load_image_list

images = load_image_list("path/to/folder")

3. Embedding Models

Load a model and processor:

from fluid_vision.models.model_loader.model_loader import ModelLoader

model_loader = ModelLoader(model_name="openai/clip-vit-base-patch16")
model, processor = model_loader.load_model()

4. Frame Embedding

Get embedding for a single frame:

from fluid_vision.embeddings.frame_embedder import FrameEmbedder

embedder = FrameEmbedder(model_name="openai/clip-vit-base-patch16", model=model, processor=processor, device="cpu")
embedding = embedder.get_frame_embedding(frame)  # frame is a PIL.Image

Get embeddings for a batch of frames:

result = embedder.get_batch_embeddings([frame1, frame2, ...])
embeddings = result["embeddings"]  # numpy.ndarray

5. Encoding Embeddings as Images

Create an encoding matrix image from embeddings:

from fluid_vision.encoder.encoder import EmbeddingEncoder

encoder = EmbeddingEncoder()
result = encoder.create_encoding_matrix(embeddings)  # embeddings: np.ndarray
encoding_image = result["encoding_image"]  # numpy.ndarray (image)

Build and Install with Poetry

1. Install Poetry (if not already installed):

   pip install poetry
   

2. Install dependencies:

   poetry install
   

3. Build the package:

   poetry build
   

4. Run tests:

   poetry run pytest
   

5. Activate the virtual environment:

   poetry shell
   

Remote Installation via devpi server:

Add this to your pip config to enable installation from devpi server

global.extra-index-url='https://pypi.org/simple'
global.index-url='https://devpi.fluidanalytics.ai/fluid_vision/dev/+simple/'
global.trusted-host='devpi.fluidanalytics.ai'