def extract_features(
model_str: str, image_folder: Path, mkey="pixel_values", batch_size: int = 8
):
- """Extract features from a folder of images
- :param model_str: model name
- :param image_folder: folder with images
- :param mkey: key for the pixel values
- :param batch_size: batch size
- :return: dict with keys "embeds" and "paths"
+ """Extract features from a folder of images.
+
+ Args:
+ model_str (str): Model name.
+ image_folder (Path): Folder with images.
+ mkey (str, optional): Key for the pixel values. Defaults to "pixel_values".
+ batch_size (int, optional): Batch size. Defaults to 8.
+
+ Returns:
+ dict: Dictionary with keys "embeds" and "paths".
"""
out_features = defaultdict(list)
diff --git a/search/search_index.json b/search/search_index.json
index a884966..f766983 100644
--- a/search/search_index.json
+++ b/search/search_index.json
@@ -1 +1 @@
-{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"video-sampler","text":"Video sampler allows you to efficiently sample video frames. Currently, it uses keyframe decoding, frame interval gating and perceptual hashing to reduce duplicated samples.
Use case: for sampling videos for later annotations used in machine learning.
"},{"location":"#table-of-contents","title":"Table of Contents","text":" - video-sampler
- Table of Contents
- Documentation
- Features
- Installation and Usage
- Basic usage
- YT-DLP integration plugin
- API examples
- Advanced usage
- Gating
- CLIP-based gating comparison
- Blur gating
- Benchmarks
- Benchmark videos
- Flit commands
- \ud83d\udee1 License
- \ud83d\udcc3 Citation
"},{"location":"#documentation","title":"Documentation","text":"Documentation is available at https://lemurpwned.github.io/video-sampler/.
"},{"location":"#features","title":"Features","text":" - [x] Direct sampling methods:
- [x]
hash - uses perceptual hashing to reduce duplicated samples - [x]
entropy - uses entropy to reduce duplicated samples (work in progress) - [x]
gzip - uses gzip compressed size to reduce duplicated samples (work in progress) - [x]
buffer - uses sliding buffer to reduce duplicated samples - [x]
grid - uses grid sampling to reduce duplicated samples - [x] Gating methods (modifications on top of direct sampling methods):
- [x]
clip - uses CLIP to filter out frames that do not contain the specified objects - [x]
blur - uses blur detection to filter out frames that are too blurry - [x] Integrations
- [x] YTDLP integration -- streams directly from yt-dlp queries, playlists or single videos
"},{"location":"#installation-and-usage","title":"Installation and Usage","text":"pip install -U video_sampler\n
then you can run
python3 -m video_sampler --help\n
or simply
video_sampler --help\n
"},{"location":"#basic-usage","title":"Basic usage","text":"python3 -m video_sampler hash FatCat.mp4 ./dataset-frames/ --hash-size 3 --buffer-size 20\n
"},{"location":"#yt-dlp-integration-plugin","title":"YT-DLP integration plugin","text":"Before using please consult the ToS of the website you are scraping from -- use responsibly and for research purposes. To use the YT-DLP integration, you need to install yt-dlp first (see yt-dlp). Then, you simply add --yt-dlp to the command, and it changes the meaning of the video_path argument.
video_sampler hash \"ytsearch:cute cats\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
video_sampler hash \"https://www.youtube.com/watch?v=W86cTIoMv2U\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
video_sampler hash \"https://www.youtube.com/watch?v=GbpP3Sxp-1U&list=PLFezMcAw96RGvTTTbdKrqew9seO2ZGRmk\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
The videos are never directly downloaded, only streamed, so you can use it to sample videos from the internet without downloading them first.
"},{"location":"#extra-yt-dlp-options","title":"Extra YT-DLP options","text":"You can pass extra options to yt-dlp by using the -yt-extra-args flag. For example:
this will only sample videos uploaded before 2019-01-01:
... --ytdlp --yt-extra-args '--datebefore 20190101'\n
or this will only sample videos uploaded after 2019-01-01:
... --ytdlp --yt-extra-args '--dateafter 20190101'\n
or this will skip all shorts:
... --ytdlp --yt-extra-args '--match-filter \"original_url!*=/shorts/ & url!*=/shorts/\"\n
"},{"location":"#api-examples","title":"API examples","text":"See examples in https://github.com/LemurPwned/video-sampler/tree/main/scripts.
"},{"location":"#advanced-usage","title":"Advanced usage","text":"There are 3 sampling methods available:
hash - uses perceptual hashing to reduce duplicated samplesentropy - uses entropy to reduce duplicated samples (work in progress)gzip - uses gzip compressed size to reduce duplicated samples (work in progress)
To launch any of them you can run and substitute method-name with one of the above:
video_sampler buffer `method-name` ...other options\n
e.g.
video_sampler buffer entropy --buffer-size 20 ...\n
where buffer-size for entropy and gzip mean the top-k sliding buffer size. Sliding buffer also uses hashing to reduce duplicated samples.
"},{"location":"#gating","title":"Gating","text":"Aside from basic sampling rules, you can also apply gating rules to the sampled frames, further reducing the number of frames. There are 3 gating methods available:
pass - pass all framesclip - use CLIP to filter out frames that do not contain the specified objectsblur - use blur detection to filter out frames that are too blurry
Here's a quick example of how to use clip:
python3 -m video_sampler clip ./videos ./scratch/clip --pos-samples \"a cat\" --neg-samples \"empty background, a lemur\" --hash-size 4\n
"},{"location":"#clip-based-gating-comparison","title":"CLIP-based gating comparison","text":"Here's a brief comparison of the frames sampled with and without CLIP-based gating with the following config:
gate_def = dict(\n type=\"clip\",\n pos_samples=[\"a cat\"],\n neg_samples=[\n \"an empty background\",\n \"text on screen\",\n \"a forest with no animals\",\n ],\n model_name=\"ViT-B-32\",\n batch_size=32,\n pos_margin=0.2,\n neg_margin=0.3,\n )\n
Evidently, CLIP-based gating is able to filter out frames that do not contain a cat and in consequence, reduce the number of frames with plain background. It also thinks that a lemur is a cat, which is not entirely wrong as fluffy creatures go.
Pass gate (no gating) CLIP gate Grid The effects of gating in numbers, for this particular set of examples (see produced vs gated columns). produced represents the number of frames sampled without gating, here after the perceptual hashing, while gated represents the number of frames sampled after gating.
video buffer gate decoded produced gated FatCat.mp4 grid pass 179 31 31 SmolCat.mp4 grid pass 118 24 24 HighLemurs.mp4 grid pass 161 35 35 FatCat.mp4 hash pass 179 101 101 SmolCat.mp4 hash pass 118 61 61 HighLemurs.mp4 hash pass 161 126 126 FatCat.mp4 hash clip 179 101 73 SmolCat.mp4 hash clip 118 61 31 HighLemurs.mp4 hash clip 161 126 66"},{"location":"#blur-gating","title":"Blur gating","text":"Helps a little with blurry videos. Adjust threshold and method (laplacian or fft) for best results. Some results from fft at threshold=20:
video buffer gate decoded produced gated MadLad.mp4 grid pass 120 31 31 MadLad.mp4 hash pass 120 110 110 MadLad.mp4 hash blur 120 110 85"},{"location":"#benchmarks","title":"Benchmarks","text":"Configuration for this benchmark:
SamplerConfig(min_frame_interval_sec=1.0, keyframes_only=True, buffer_size=30, hash_size=X, queue_wait=0.1, debug=True)\n
Video Total frames Hash size Decoded Saved SmolCat 2936 8 118 106 SmolCat - 4 - 61 Fat Cat 4462 8 179 163 Fat Cat - 4 - 101 HighLemurs 4020 8 161 154 HighLemurs - 4 - 126 SamplerConfig(\n min_frame_interval_sec=1.0,\n keyframes_only=True,\n queue_wait=0.1,\n debug=False,\n print_stats=True,\n buffer_config={'type': 'entropy'/'gzip', 'size': 30, 'debug': False, 'hash_size': 8, 'expiry': 50}\n)\n
Video Total frames Type Decoded Saved SmolCat 2936 entropy 118 39 SmolCat - gzip - 39 Fat Cat 4462 entropy 179 64 Fat Cat - gzip - 73 HighLemurs 4020 entropy 161 59 HighLemurs - gzip - 63"},{"location":"#benchmark-videos","title":"Benchmark videos","text":" - SmolCat
- Fat Cat
- HighLemurs
- MadLad
"},{"location":"#flit-commands","title":"Flit commands","text":""},{"location":"#build","title":"Build","text":"flit build\n
"},{"location":"#install","title":"Install","text":"flit install\n
"},{"location":"#publish","title":"Publish","text":"Remember to bump the version in pyproject.toml before publishing.
flit publish\n
"},{"location":"#license","title":"\ud83d\udee1 License","text":"This project is licensed under the terms of the MIT license. See LICENSE for more details.
"},{"location":"#citation","title":"\ud83d\udcc3 Citation","text":"@misc{video-sampler,\n author = {video-sampler},\n title = {Video sampler allows you to efficiently sample video frames},\n year = {2023},\n publisher = {GitHub},\n journal = {GitHub repository},\n howpublished = {\\url{https://github.com/LemurPwned/video-sampler}}\n}\n
"},{"location":"reference/video_sampler/buffer/","title":"Video sampler","text":""},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.EntropyByffer","title":"EntropyByffer","text":" Bases: FrameBuffer
Measure image entropy as a function of the image usability
Source code in video_sampler/buffer.py class EntropyByffer(FrameBuffer):\n \"\"\"Measure image entropy as a function of the image usability\"\"\"\n\n def __init__(\n self, size: int, expiry: int, debug_flag: bool = False, hash_size: int = 8\n ) -> None:\n self.sliding_top_k_buffer = SlidingTopKBuffer(\n size=size, expiry=expiry, debug_flag=debug_flag, hash_size=hash_size\n )\n\n def get_buffer_state(self) -> list[str]:\n return self.sliding_top_k_buffer.get_buffer_state()\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n entropy = item.entropy()\n return self.sliding_top_k_buffer.add(item, {**metadata, \"index\": -entropy})\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n return self.sliding_top_k_buffer.final_flush()\n\n def clear(self):\n self.sliding_top_k_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer","title":"FrameBuffer","text":" Bases: ABC
Source code in video_sampler/buffer.py class FrameBuffer(ABC):\n @abstractmethod\n def add(self, item: Image.Image, metadata: dict[str, Any]) -> None | tuple:\n pass\n\n @abstractmethod\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n \"\"\"Flush the buffer and return the remaining items\"\"\"\n pass\n\n @abstractmethod\n def get_buffer_state(self) -> list[str]:\n \"\"\"Return the current state of the buffer\"\"\"\n pass\n\n @abstractmethod\n def clear(self):\n \"\"\"Clear the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.clear","title":"clear() abstractmethod","text":"Clear the buffer
Source code in video_sampler/buffer.py @abstractmethod\ndef clear(self):\n \"\"\"Clear the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.final_flush","title":"final_flush() abstractmethod","text":"Flush the buffer and return the remaining items
Source code in video_sampler/buffer.py @abstractmethod\ndef final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n \"\"\"Flush the buffer and return the remaining items\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.get_buffer_state","title":"get_buffer_state() abstractmethod","text":"Return the current state of the buffer
Source code in video_sampler/buffer.py @abstractmethod\ndef get_buffer_state(self) -> list[str]:\n \"\"\"Return the current state of the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GridBuffer","title":"GridBuffer","text":" Bases: HashBuffer
A class representing a grid-based buffer for images. Splits the image into a grid and stores the hashes of the grid cells in a mosaic buffer.
Parameters:
Name Type Description Default size int The maximum size of the buffer.
required debug_flag bool A flag indicating whether debug information should be printed.
False hash_size int The size of the hash.
4 grid_x int The number of grid cells in the x-axis.
4 grid_y int The number of grid cells in the y-axis.
4 max_hits int The maximum number of hits allowed for a hash.
1 Attributes:
Name Type Description grid_x int The number of grid cells in the x-axis.
grid_y int The number of grid cells in the y-axis.
max_hits int The maximum number of hits allowed for a hash.
mosaic_buffer dict A dictionary storing the mosaic buffer.
Methods:
Name Description add Adds an image to the buffer along with its metadata.
clear Clears the buffer and the mosaic buffer.
update_ttl_buffer Updates the buffer by expiring images that are not in the grid.
Source code in video_sampler/buffer.py class GridBuffer(HashBuffer):\n \"\"\"\n A class representing a grid-based buffer for images.\n Splits the image into a grid and stores the hashes of the grid cells in a mosaic buffer.\n\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): A flag indicating whether debug information should be printed.\n hash_size (int, optional): The size of the hash.\n grid_x (int, optional): The number of grid cells in the x-axis.\n grid_y (int, optional): The number of grid cells in the y-axis.\n max_hits (int, optional): The maximum number of hits allowed for a hash.\n\n Attributes:\n grid_x (int): The number of grid cells in the x-axis.\n grid_y (int): The number of grid cells in the y-axis.\n max_hits (int): The maximum number of hits allowed for a hash.\n mosaic_buffer (dict): A dictionary storing the mosaic buffer.\n\n Methods:\n add(item, metadata):\n Adds an image to the buffer along with its metadata.\n clear():\n Clears the buffer and the mosaic buffer.\n update_ttl_buffer():\n Updates the buffer by expiring images that are not in the grid.\n\n \"\"\"\n\n def __init__(\n self,\n size: int,\n debug_flag: bool = False,\n hash_size: int = 4,\n grid_x: int = 4,\n grid_y: int = 4,\n max_hits: int = 1,\n ) -> None:\n super().__init__(size, debug_flag, hash_size)\n self.grid_x = grid_x\n self.grid_y = grid_y\n self.max_hits = max_hits\n self.mosaic_buffer = {}\n\n def __get_grid_hash(self, item: Image.Image) -> str:\n \"\"\"Compute grid hashes for a given image\"\"\"\n for x in range(self.grid_x):\n for y in range(self.grid_y):\n yield str(\n phash(\n item.crop(\n (\n x * item.width / self.grid_x,\n y * item.height / self.grid_y,\n (x + 1) * item.width / self.grid_x,\n (y + 1) * item.height / self.grid_y,\n )\n ),\n hash_size=self.hash_size,\n )\n )\n\n def _check_mosaic(self, mosaic_hash: str):\n return mosaic_hash in self.mosaic_buffer\n\n def update_ttl_buffer(self):\n # expire the images that are not in the grid\n if len(self.ordered_buffer) >= self.max_size:\n to_return_hash, return_data = self.ordered_buffer.popitem(last=False)\n if to_return_hash is not None:\n removal_keys = [\n img_hash\n for img_hash, mosaic_hash in self.mosaic_buffer.items()\n if mosaic_hash == to_return_hash\n ]\n for key in removal_keys:\n del self.mosaic_buffer[key]\n return return_data\n return None\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n hash_ = str(phash(item, hash_size=self.hash_size))\n if not self._check_duplicate(hash_):\n # not automatically rejected, check the mosaic buffer\n hash_hits = 0\n hash_sets = []\n for el_hash_ in self.__get_grid_hash(item):\n if el_hash_ in self.mosaic_buffer:\n hash_hits += 1\n hash_sets.append(el_hash_)\n\n if hash_hits < self.max_hits:\n # add image hash to the ttl counter\n self.ordered_buffer[hash_] = (item, metadata)\n # add the image to the mosaic buffer\n # this also automatically overwrites the deleted hashes\n for el_hash in hash_sets:\n self.mosaic_buffer[el_hash] = hash_\n\n if self.debug_flag:\n console.print(\n f\"\\tHash hits: {hash_hits}\"\n f\"\\tHash sets: {len(hash_sets)}\"\n f\"\\tHash buffer: {len(self.get_buffer_state())}\"\n f\"\\tMosaic buffer: {len(self.mosaic_buffer)}\"\n )\n return self.update_ttl_buffer()\n\n def clear(self):\n super().clear()\n self.mosaic_buffer = {}\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GridBuffer.__get_grid_hash","title":"__get_grid_hash(item)","text":"Compute grid hashes for a given image
Source code in video_sampler/buffer.py def __get_grid_hash(self, item: Image.Image) -> str:\n \"\"\"Compute grid hashes for a given image\"\"\"\n for x in range(self.grid_x):\n for y in range(self.grid_y):\n yield str(\n phash(\n item.crop(\n (\n x * item.width / self.grid_x,\n y * item.height / self.grid_y,\n (x + 1) * item.width / self.grid_x,\n (y + 1) * item.height / self.grid_y,\n )\n ),\n hash_size=self.hash_size,\n )\n )\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GzipBuffer","title":"GzipBuffer","text":" Bases: FrameBuffer
Measure compression size as a function of the image usability
Source code in video_sampler/buffer.py class GzipBuffer(FrameBuffer):\n \"\"\"Measure compression size as a function of the image usability\"\"\"\n\n def __init__(\n self, size: int, expiry: int, debug_flag: bool = False, hash_size: int = 8\n ) -> None:\n self.sliding_top_k_buffer = SlidingTopKBuffer(\n size=size, expiry=expiry, debug_flag=debug_flag, hash_size=hash_size\n )\n\n def get_buffer_state(self) -> list[str]:\n return self.sliding_top_k_buffer.get_buffer_state()\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n compressed_l = len(gzip.compress(item.tobytes()))\n return self.sliding_top_k_buffer.add(item, {**metadata, \"index\": -compressed_l})\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n return self.sliding_top_k_buffer.final_flush()\n\n def clear(self):\n self.sliding_top_k_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.HashBuffer","title":"HashBuffer","text":" Bases: FrameBuffer
A buffer that stores frames with their corresponding metadata and checks for duplicates based on image hashes. Args: size (int): The maximum size of the buffer. debug_flag (bool, optional): Flag indicating whether to enable debug mode. Defaults to False. hash_size (int, optional): The size of the image hash. Defaults to 4.
Methods:
Name Description get_buffer_state Returns the current state of the buffer as a list of image hashes.
add Image.Image, metadata: dict[str, Any]) Adds an item to the buffer along with its metadata.
final_flush Yields the stored items and their metadata in the buffer.
Private Methods __add(item: Image.Image, hash_: str, metadata: dict) Adds an item to the buffer with the given hash and metadata.
__check_duplicate(hash_: str) -> bool: Checks if the given hash already exists in the buffer and renews its validity if found.
Source code in video_sampler/buffer.py class HashBuffer(FrameBuffer):\n \"\"\"\n A buffer that stores frames with their corresponding metadata and\n checks for duplicates based on image hashes.\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): Flag indicating whether to enable debug mode. Defaults to False.\n hash_size (int, optional): The size of the image hash. Defaults to 4.\n\n Methods:\n get_buffer_state() -> list[str]:\n Returns the current state of the buffer as a list of image hashes.\n\n add(item: Image.Image, metadata: dict[str, Any])\n Adds an item to the buffer along with its metadata.\n\n final_flush() -> Iterable[tuple[Image.Image | None, dict]]:\n Yields the stored items and their metadata in the buffer.\n\n clear()\n Clears the buffer.\n\n Private Methods:\n __add(item: Image.Image, hash_: str, metadata: dict)\n Adds an item to the buffer with the given hash and metadata.\n\n __check_duplicate(hash_: str) -> bool:\n Checks if the given hash already exists in the buffer and renews its validity if found.\n\n \"\"\"\n\n def __init__(self, size: int, debug_flag: bool = False, hash_size: int = 4) -> None:\n self.ordered_buffer = OrderedDict()\n self.max_size = size\n self.debug_flag = debug_flag\n self.hash_size = hash_size\n\n def get_buffer_state(self) -> list[str]:\n return list(self.ordered_buffer.keys())\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n hash_ = str(phash(item, hash_size=self.hash_size))\n if not self._check_duplicate(hash_):\n return self.__add(hash_, item, metadata)\n return None\n\n def __add(self, hash_: str, item: Image.Image, metadata: dict):\n self.ordered_buffer[hash_] = (item, metadata)\n if len(self.ordered_buffer) >= self.max_size:\n return self.ordered_buffer.popitem(last=False)[1]\n return None\n\n def _check_duplicate(self, hash_: str) -> bool:\n if hash_ in self.ordered_buffer:\n # renew the hash validity\n if self.debug_flag:\n console.print(\n f\"Renewing {hash_}\",\n style=f\"bold {Color.red.value}\",\n )\n self.ordered_buffer.move_to_end(hash_)\n return True\n return False\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n yield from self.ordered_buffer.values()\n\n def clear(self):\n self.ordered_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.SamplerConfig","title":"SamplerConfig dataclass","text":"Configuration options for the video sampler.
Parameters:
Name Type Description Default min_frame_interval_sec float The minimum time interval between sampled frames in seconds. Defaults to 1.
1 keyframes_only bool Flag indicating whether to sample only keyframes. Defaults to True.
True queue_wait float The time to wait between checking the frame queue in seconds. Defaults to 0.1.
0.1 debug bool Flag indicating whether to enable debug mode. Defaults to False.
False print_stats bool Flag indicating whether to print sampling statistics. Defaults to False.
False buffer_config dict[str, Any] Configuration options for the frame buffer. Defaults to {\"type\": \"entropy\", \"size\": 15, \"debug\": True}.
field(default_factory=lambda : {'type': 'hash', 'hash_size': 8, 'size': 15, 'debug': True}) gate_config dict[str, Any] Configuration options for the frame gate. Defaults to {\"type\": \"pass\"}.
field(default_factory=lambda : {'type': 'pass'}) Methods:
Name Description __str__ Returns a string representation of the configuration.
Source code in video_sampler/buffer.py @dataclass\nclass SamplerConfig:\n \"\"\"\n Configuration options for the video sampler.\n\n Args:\n min_frame_interval_sec (float, optional): The minimum time interval\n between sampled frames in seconds. Defaults to 1.\n keyframes_only (bool, optional): Flag indicating whether to\n sample only keyframes. Defaults to True.\n queue_wait (float, optional): The time to wait between checking\n the frame queue in seconds. Defaults to 0.1.\n debug (bool, optional): Flag indicating whether to enable debug mode.\n Defaults to False.\n print_stats (bool, optional): Flag indicating whether to print\n sampling statistics. Defaults to False.\n buffer_config (dict[str, Any], optional): Configuration options for\n the frame buffer. Defaults to {\"type\": \"entropy\", \"size\": 15,\n \"debug\": True}.\n gate_config (dict[str, Any], optional): Configuration options for\n the frame gate. Defaults to {\"type\": \"pass\"}.\n\n Methods:\n __str__() -> str:\n Returns a string representation of the configuration.\n\n \"\"\"\n\n min_frame_interval_sec: float = 1\n keyframes_only: bool = True\n queue_wait: float = 0.1\n debug: bool = False\n print_stats: bool = False\n buffer_config: dict[str, Any] = field(\n default_factory=lambda: {\n \"type\": \"hash\",\n \"hash_size\": 8,\n \"size\": 15,\n \"debug\": True,\n }\n )\n gate_config: dict[str, Any] = field(\n default_factory=lambda: {\n \"type\": \"pass\",\n }\n )\n\n def __str__(self) -> str:\n return str(asdict(self))\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.SlidingTopKBuffer","title":"SlidingTopKBuffer","text":" Bases: FrameBuffer
A class representing a sliding top-k buffer for frames.
Parameters:
Name Type Description Default size int The maximum size of the buffer.
required debug_flag bool A flag indicating whether debug information should be printed.
False expiry int The expiry count for frames.
30 hash_size int The size of the hash.
8 Attributes:
Name Type Description sliding_buffer list The sliding buffer implemented as a min heap.
max_size int The maximum size of the buffer.
debug_flag bool A flag indicating whether debug information should be printed.
expiry_count int The expiry count for frames.
hash_size int The size of the hash.
Methods:
Name Description get_buffer_state Returns the current state of the buffer.
add Adds a frame to the buffer along with its metadata.
final_flush Performs a final flush of the buffer and yields the remaining frames.
clear Clears the buffer.
Source code in video_sampler/buffer.py class SlidingTopKBuffer(FrameBuffer):\n \"\"\"\n A class representing a sliding top-k buffer for frames.\n\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): A flag indicating whether debug information should be printed.\n expiry (int, optional): The expiry count for frames.\n hash_size (int, optional): The size of the hash.\n\n Attributes:\n sliding_buffer (list): The sliding buffer implemented as a min heap.\n max_size (int): The maximum size of the buffer.\n debug_flag (bool): A flag indicating whether debug information should be printed.\n expiry_count (int): The expiry count for frames.\n hash_size (int): The size of the hash.\n\n Methods:\n get_buffer_state() -> list[str]:\n Returns the current state of the buffer.\n add(item, metadata):\n Adds a frame to the buffer along with its metadata.\n final_flush() -> Iterable[tuple[Image.Image | None, dict]]:\n Performs a final flush of the buffer and yields the remaining frames.\n clear():\n Clears the buffer.\n\n \"\"\"\n\n def __init__(\n self, size: int, debug_flag: bool = False, expiry: int = 30, hash_size: int = 8\n ) -> None:\n # it's a min heap with a fixed size\n self.sliding_buffer = []\n self.max_size = size\n self.debug_flag = debug_flag\n self.expiry_count = expiry\n self.hash_size = hash_size\n assert (\n self.expiry_count > self.max_size\n ), \"expiry count must be greater than max size\"\n console.print(\n f\"Creating sliding buffer of size {self.max_size} and expiry {expiry}\",\n style=f\"bold {Color.red.value}\",\n )\n\n def get_buffer_state(self) -> list[str]:\n return [item[:3] for item in self.sliding_buffer]\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n assert \"index\" in metadata, \"metadata must have index key for sliding buffer\"\n average_hash_ = str(average_hash(item, hash_size=self.hash_size))\n to_return = None\n if not self.__check_duplicate(average_hash_):\n heapq.heappush(\n self.sliding_buffer,\n [metadata[\"index\"], 0, average_hash_, item, metadata],\n )\n if len(self.sliding_buffer) >= self.max_size:\n to_return = heapq.heappop(self.sliding_buffer)[-2:]\n # update the expiry count\n expired_indx = -1\n for i in range(len(self.sliding_buffer)):\n self.sliding_buffer[i][1] += 1\n if self.sliding_buffer[i][1] >= self.expiry_count:\n expired_indx = i\n # at any point only one item can be expired\n if expired_indx != -1:\n self.sliding_buffer.pop(expired_indx) # just drop\n return to_return\n\n def __check_duplicate(self, hash_: str) -> bool:\n for item in self.sliding_buffer:\n if item[2] == hash_:\n # renew the hash validity\n if self.debug_flag:\n console.print(\n f\"Renewing {hash_}\",\n style=f\"bold {Color.red.value}\",\n )\n item[1] = 0\n return True\n return False\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n if len(self.sliding_buffer):\n yield heapq.heappop(self.sliding_buffer)[-2:]\n yield None, {}\n\n def clear(self):\n self.sliding_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.create_buffer","title":"create_buffer(buffer_config)","text":"Create a buffer based on the config
Source code in video_sampler/buffer.py def create_buffer(buffer_config: dict[str, Any]):\n \"\"\"Create a buffer based on the config\"\"\"\n console.print(\n f\"Creating buffer of type {buffer_config['type']}\",\n style=f\"bold {Color.red.value}\",\n )\n if buffer_config[\"type\"] == \"hash\":\n return HashBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n )\n elif buffer_config[\"type\"] == \"grid\":\n return GridBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n grid_x=buffer_config[\"grid_x\"],\n grid_y=buffer_config[\"grid_y\"],\n max_hits=buffer_config[\"max_hits\"],\n )\n elif buffer_config[\"type\"] == \"sliding_top_k\":\n return SlidingTopKBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n expiry=buffer_config[\"expiry\"],\n )\n elif buffer_config[\"type\"] == \"passthrough\":\n return PassThroughBuffer()\n elif buffer_config[\"type\"] == \"gzip\":\n return GzipBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n expiry=buffer_config[\"expiry\"],\n )\n elif buffer_config[\"type\"] == \"entropy\":\n return EntropyByffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n expiry=buffer_config[\"expiry\"],\n )\n else:\n raise ValueError(f\"Unknown buffer type {buffer_config['type']}\")\n
"},{"location":"reference/video_sampler/evaluation/","title":"Evaluation","text":""},{"location":"reference/video_sampler/evaluation/#video_sampler.evaluation.compute_total_video_entropy","title":"compute_total_video_entropy()","text":"Compute the total entropy of a video
Source code in video_sampler/evaluation.py def compute_total_video_entropy():\n \"\"\"Compute the total entropy of a video\"\"\"\n pass\n
"},{"location":"reference/video_sampler/gating/","title":"Gating","text":""},{"location":"reference/video_sampler/gating/#video_sampler.gating.BlurGate","title":"BlurGate","text":"Source code in video_sampler/gating.py class BlurGate:\n def __init__(\n self, method: Literal[\"fft\", \"laplacian\"] = \"laplacian\", threshold: float = 100\n ) -> None:\n \"\"\"Gate frames based on bluriness.\n :param method: The method to use for blur detection. Can be \"fft\" or \"laplacian\".\n :param threshold: The threshold for bluriness. The higher the threshold, the less\n blurry the image needs to be to be discarded.\n Those are different depending on the method:\n - 20 is a good start for fft\n - 100 is a good start for laplacian.\n \"\"\"\n self.is_blurry = None\n if method == \"fft\":\n self.is_blurry = self._is_blurry_fft\n elif method == \"laplacian\":\n self.is_blurry = self._is_blurry_laplacian\n else:\n raise ValueError(f\"Unknown blur method {method}\")\n self.threshold = threshold\n\n def __call__(self, frame: Image.Image, meta: dict, last=False) -> GatedObject:\n if self.is_blurry(frame) or last:\n return EMPTY_GATED_OBJECT\n return GatedObject([FrameObject(frame, meta)], 1)\n\n def _is_blurry_laplacian(self, frame: Image.Image) -> bool:\n \"\"\"Check if the image is blurry with laplacian method.\"\"\"\n return (\n cv2.Laplacian(\n cv2.cvtColor(np.array(frame), cv2.COLOR_BGR2GRAY), cv2.CV_64F\n ).var()\n < self.threshold\n )\n\n def _is_blurry_fft(self, frame: Image.Image) -> bool:\n \"\"\"Check if the image is blurry with fft method.\"\"\"\n f = np.fft.fft2(frame)\n fshift = np.fft.fftshift(f)\n magnitude_spectrum = 20 * np.log(np.abs(fshift) + 1e-12)\n return magnitude_spectrum.mean() < self.threshold\n\n def flush(self):\n return EMPTY_GATED_OBJECT\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.BlurGate.__init__","title":"__init__(method='laplacian', threshold=100)","text":"Gate frames based on bluriness. :param method: The method to use for blur detection. Can be \"fft\" or \"laplacian\". :param threshold: The threshold for bluriness. The higher the threshold, the less blurry the image needs to be to be discarded. Those are different depending on the method: - 20 is a good start for fft - 100 is a good start for laplacian.
Source code in video_sampler/gating.py def __init__(\n self, method: Literal[\"fft\", \"laplacian\"] = \"laplacian\", threshold: float = 100\n) -> None:\n \"\"\"Gate frames based on bluriness.\n :param method: The method to use for blur detection. Can be \"fft\" or \"laplacian\".\n :param threshold: The threshold for bluriness. The higher the threshold, the less\n blurry the image needs to be to be discarded.\n Those are different depending on the method:\n - 20 is a good start for fft\n - 100 is a good start for laplacian.\n \"\"\"\n self.is_blurry = None\n if method == \"fft\":\n self.is_blurry = self._is_blurry_fft\n elif method == \"laplacian\":\n self.is_blurry = self._is_blurry_laplacian\n else:\n raise ValueError(f\"Unknown blur method {method}\")\n self.threshold = threshold\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.ClipGate","title":"ClipGate","text":"Source code in video_sampler/gating.py class ClipGate:\n def __init__(\n self,\n pos_samples: list[str] = None,\n neg_samples: list[str] = None,\n model_name: str = \"ViT-B-32\",\n batch_size: int = 32,\n pos_margin: float = 0.2,\n neg_margin: float = 0.3,\n ) -> None:\n self.model, self.preprocess, self.tokenizer = create_model(\n model_name=model_name\n )\n self.pos_margin = pos_margin\n self.neg_margin = neg_margin\n self.batch_size = batch_size\n self.frame_accumulator = []\n self.metadata_accumulator = []\n if pos_samples is None:\n self.pos_samples = torch.zeros((1, 512))\n else:\n self.pos_samples = self._preproc_samples(pos_samples)\n if neg_samples is None:\n self.neg_samples = torch.zeros((1, 512))\n else:\n self.neg_samples = self._preproc_samples(neg_samples)\n\n def __call__(self, frame: Image.Image, meta: dict, last=False) -> Any:\n return self.flush() if last else self.add_frame(frame, meta)\n\n def _preproc_samples(self, sample_texts: list[str]):\n inputs = self.tokenizer(sample_texts)\n embeds = torch.zeros((len(sample_texts), 512))\n with torch.no_grad():\n for i, batch in enumerate(batched(inputs, n=self.batch_size)):\n batch = torch.stack(batch)\n text_embeds = self.model.encode_text(batch.to(DEVICE))\n embeds[i * self.batch_size : (i + 1) * self.batch_size] = (\n text_embeds.cpu()\n )\n embeds /= embeds.norm(dim=-1, keepdim=True)\n return embeds\n\n def _embed_frames(self, frames: list[Image.Image]):\n \"\"\"Compute the embeddings for each frame.\"\"\"\n inputs = torch.stack([self.preprocess(frame) for frame in frames]).to(DEVICE)\n with torch.no_grad():\n image_embeds = self.model.encode_image(inputs).cpu()\n image_embeds /= image_embeds.norm(dim=-1, keepdim=True)\n return image_embeds\n\n def _get_margins(self, frame_embeds: torch.Tensor):\n \"\"\"Compute the margins for each frame.\"\"\"\n org_indx = np.arange(frame_embeds.shape[0])\n neg_distance = frame_embeds @ self.neg_samples.T\n pos_distance = frame_embeds @ self.pos_samples.T\n neg_margin, _ = neg_distance.max(axis=-1)\n pos_margin, _ = pos_distance.max(axis=-1)\n incl_samples = torch.argwhere(\n (neg_margin < self.neg_margin) & (pos_margin >= self.pos_margin)\n )\n return org_indx[incl_samples].ravel()\n\n def add_frame(self, frame: Image.Image, metadata: dict) -> GatedObject:\n self.frame_accumulator.append(frame)\n self.metadata_accumulator.append(metadata)\n if len(self.frame_accumulator) == self.batch_size:\n return self.__process_metadata()\n return EMPTY_GATED_OBJECT\n\n def flush(self):\n return self.__process_metadata()\n\n def __process_metadata(self) -> GatedObject:\n frame_embeds = self._embed_frames(self.frame_accumulator)\n selected_frames = self._get_margins(frame_embeds)\n to_return = [\n FrameObject(self.frame_accumulator[i], self.metadata_accumulator[i])\n for i in range(len(self.frame_accumulator))\n if i in selected_frames\n ]\n self.frame_accumulator.clear()\n self.metadata_accumulator.clear()\n return GatedObject(to_return, len(selected_frames))\n
"},{"location":"reference/video_sampler/iterators/","title":"Iterators","text":""},{"location":"reference/video_sampler/logging/","title":"Logging","text":""},{"location":"reference/video_sampler/sampler/","title":"Sampler","text":""},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.VideoSampler","title":"VideoSampler","text":"The fundamental class for sampling video frames.
Parameters:
Name Type Description Default cfg SamplerConfig The configuration for the video sampler.
required Attributes:
Name Type Description cfg SamplerConfig The configuration for the video sampler.
frame_buffer FrameBuffer The frame buffer used for sampling frames.
gate Gate The gate used for filtering frames.
stats Counter A counter for tracking statistics.
Methods:
Name Description sample Generates sample frames from a video.
write_queue Writes sampled frames to a queue.
Source code in video_sampler/sampler.py class VideoSampler:\n \"\"\"\n The fundamental class for sampling video frames.\n\n Args:\n cfg (SamplerConfig): The configuration for the video sampler.\n\n Attributes:\n cfg (SamplerConfig): The configuration for the video sampler.\n frame_buffer (FrameBuffer): The frame buffer used for sampling frames.\n gate (Gate): The gate used for filtering frames.\n stats (Counter): A counter for tracking statistics.\n\n Methods:\n sample(video_path) -> Iterable[list[FrameObject]]:\n Generates sample frames from a video.\n write_queue(video_path, q):\n Writes sampled frames to a queue.\n\n \"\"\"\n\n def __init__(self, cfg: SamplerConfig) -> None:\n self.cfg = deepcopy(cfg)\n self.frame_buffer = create_buffer(self.cfg.buffer_config)\n self.gate = create_gate(self.cfg.gate_config)\n self.stats = Counter()\n\n def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video\"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n # skip frames if keyframes_only is True\n time_diff = frame.time - prev_time\n self.stats[\"total\"] += 1\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = frame.time\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {frame.time}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": frame.time, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n for res in self.frame_buffer.final_flush():\n if res:\n self.stats[\"produced\"] += 1\n gated_obj = self.gate(*res)\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n gated_obj = self.gate.flush()\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n yield PROCESSING_DONE_ITERABLE\n\n def write_queue(self, video_path: str, q: Queue):\n try:\n item: tuple[FrameObject, int]\n for item in self.sample(video_path=video_path):\n q.put(item)\n except (av.IsADirectoryError, av.InvalidDataError) as e:\n console.print(\n f\"Error while processing {video_path}\",\n f\"\\n\\t{e}\",\n style=f\"bold {Color.red.value}\",\n )\n q.put(PROCESSING_DONE_ITERABLE)\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.VideoSampler.sample","title":"sample(video_path)","text":"Generate sample frames from a video
Source code in video_sampler/sampler.py def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video\"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n # skip frames if keyframes_only is True\n time_diff = frame.time - prev_time\n self.stats[\"total\"] += 1\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = frame.time\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {frame.time}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": frame.time, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n for res in self.frame_buffer.final_flush():\n if res:\n self.stats[\"produced\"] += 1\n gated_obj = self.gate(*res)\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n gated_obj = self.gate.flush()\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n yield PROCESSING_DONE_ITERABLE\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.Worker","title":"Worker","text":"Source code in video_sampler/sampler.py class Worker:\n def __init__(\n self,\n cfg: SamplerConfig,\n devnull: bool = False,\n processor_cls: VideoSampler = VideoSampler,\n extra_processor_args: dict = None,\n ) -> None:\n if extra_processor_args is None:\n extra_processor_args = {}\n self.cfg = cfg\n self.processor = processor_cls(cfg=cfg, **extra_processor_args)\n self.q = Queue()\n self.devnull = devnull\n\n def launch(\n self, video_path: str, output_path: str = \"\", pretty_video_name: str = \"\"\n ) -> None:\n \"\"\"Launch the worker.\n :param video_path: path to the video file\n :param output_path: path to the output folder\n :param pretty_video_name: name of the video file for pretty printing (useful for urls)\n \"\"\"\n if not pretty_video_name:\n pretty_video_name = os.path.basename(video_path)\n if output_path and self.devnull:\n raise ValueError(\"Cannot write to disk when devnull is True\")\n if output_path:\n os.makedirs(output_path, exist_ok=True)\n proc_thread = Thread(\n target=self.processor.write_queue, args=(video_path, self.q)\n )\n proc_thread.start()\n self.queue_reader(output_path, read_interval=self.cfg.queue_wait)\n proc_thread.join()\n if self.cfg.print_stats:\n console.print(\n f\"Stats for: {pretty_video_name}\",\n f\"\\n\\tTotal frames: {self.processor.stats['total']}\",\n f\"\\n\\tDecoded frames: {self.processor.stats['decoded']}\",\n f\"\\n\\tProduced frames: {self.processor.stats['produced']}\",\n f\"\\n\\tGated frames: {self.processor.stats['gated']}\",\n style=f\"bold {Color.magenta.value}\",\n )\n\n def queue_reader(self, output_path, read_interval=0.1) -> None:\n while True:\n if not self.q.empty():\n frame_object: FrameObject\n for frame_object in self.q.get():\n if frame_object.metadata.get(\"end\", False):\n return\n if frame_object.frame is not None and (\n not self.devnull and isinstance(frame_object.frame, Image.Image)\n ):\n frame_object.frame.save(\n os.path.join(\n output_path,\n f\"{frame_object.metadata['frame_time']}.jpg\",\n )\n )\n time.sleep(read_interval)\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.Worker.launch","title":"launch(video_path, output_path='', pretty_video_name='')","text":"Launch the worker. :param video_path: path to the video file :param output_path: path to the output folder :param pretty_video_name: name of the video file for pretty printing (useful for urls)
Source code in video_sampler/sampler.py def launch(\n self, video_path: str, output_path: str = \"\", pretty_video_name: str = \"\"\n) -> None:\n \"\"\"Launch the worker.\n :param video_path: path to the video file\n :param output_path: path to the output folder\n :param pretty_video_name: name of the video file for pretty printing (useful for urls)\n \"\"\"\n if not pretty_video_name:\n pretty_video_name = os.path.basename(video_path)\n if output_path and self.devnull:\n raise ValueError(\"Cannot write to disk when devnull is True\")\n if output_path:\n os.makedirs(output_path, exist_ok=True)\n proc_thread = Thread(\n target=self.processor.write_queue, args=(video_path, self.q)\n )\n proc_thread.start()\n self.queue_reader(output_path, read_interval=self.cfg.queue_wait)\n proc_thread.join()\n if self.cfg.print_stats:\n console.print(\n f\"Stats for: {pretty_video_name}\",\n f\"\\n\\tTotal frames: {self.processor.stats['total']}\",\n f\"\\n\\tDecoded frames: {self.processor.stats['decoded']}\",\n f\"\\n\\tProduced frames: {self.processor.stats['produced']}\",\n f\"\\n\\tGated frames: {self.processor.stats['gated']}\",\n style=f\"bold {Color.magenta.value}\",\n )\n
"},{"location":"reference/video_sampler/schemas/","title":"Schemas","text":""},{"location":"reference/video_sampler/ttl_counter/","title":"Ttl counter","text":""},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter","title":"TTLCounter","text":"TTLCounter is a counter/list that expires items after a TTL period expires.
Source code in video_sampler/ttl_counter.py class TTLCounter:\n \"\"\"TTLCounter is a counter/list that expires items after a TTL period expires.\"\"\"\n\n def __init__(self, max_ttl: int) -> None:\n self.inner_counter = []\n self.max_ttl = max_ttl\n\n def __len__(self):\n \"\"\"Return the number of items in the counter.\"\"\"\n return len(self.inner_counter)\n\n def add_item(self, hash: str):\n \"\"\"Add an item with the max TTL.\"\"\"\n heapq.heappush(self.inner_counter, (self.max_ttl, hash))\n\n def tick(self):\n \"\"\"Decrease the TTL of all items by 1.\"\"\"\n for i, (ttl, hash) in enumerate(self.inner_counter):\n self.inner_counter[i] = (ttl - 1, hash)\n\n def expire_one(self):\n \"\"\"Expire the first item if its TTL is 0. Expires AT MOST one item.\"\"\"\n # peek the first item\n ttl, hash = self.inner_counter[0]\n if ttl <= 0:\n heapq.heappop(self.inner_counter)\n return hash\n return None\n\n def expire_all(self):\n \"\"\"Expire all items.\"\"\"\n for _, hash in self.inner_counter:\n yield hash\n self.inner_counter.clear()\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.__len__","title":"__len__()","text":"Return the number of items in the counter.
Source code in video_sampler/ttl_counter.py def __len__(self):\n \"\"\"Return the number of items in the counter.\"\"\"\n return len(self.inner_counter)\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.add_item","title":"add_item(hash)","text":"Add an item with the max TTL.
Source code in video_sampler/ttl_counter.py def add_item(self, hash: str):\n \"\"\"Add an item with the max TTL.\"\"\"\n heapq.heappush(self.inner_counter, (self.max_ttl, hash))\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.expire_all","title":"expire_all()","text":"Expire all items.
Source code in video_sampler/ttl_counter.py def expire_all(self):\n \"\"\"Expire all items.\"\"\"\n for _, hash in self.inner_counter:\n yield hash\n self.inner_counter.clear()\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.expire_one","title":"expire_one()","text":"Expire the first item if its TTL is 0. Expires AT MOST one item.
Source code in video_sampler/ttl_counter.py def expire_one(self):\n \"\"\"Expire the first item if its TTL is 0. Expires AT MOST one item.\"\"\"\n # peek the first item\n ttl, hash = self.inner_counter[0]\n if ttl <= 0:\n heapq.heappop(self.inner_counter)\n return hash\n return None\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.tick","title":"tick()","text":"Decrease the TTL of all items by 1.
Source code in video_sampler/ttl_counter.py def tick(self):\n \"\"\"Decrease the TTL of all items by 1.\"\"\"\n for i, (ttl, hash) in enumerate(self.inner_counter):\n self.inner_counter[i] = (ttl - 1, hash)\n
"},{"location":"reference/video_sampler/utils/","title":"Utils","text":""},{"location":"reference/video_sampler/utils/#video_sampler.utils.batched","title":"batched(iterable, n)","text":"Batch data into tuples of length n. The last batch may be shorter. from https://docs.python.org/3/library/itertools.html#itertools-recipes
Source code in video_sampler/utils.py def batched(iterable, n):\n \"\"\"\n Batch data into tuples of length n. The last batch may be shorter.\n from https://docs.python.org/3/library/itertools.html#itertools-recipes\n \"\"\"\n if n < 1:\n raise ValueError(\"n must be at least one\")\n it = iter(iterable)\n while batch := tuple(islice(it, n)):\n yield batch\n
"},{"location":"reference/video_sampler/utils/#video_sampler.utils.slugify","title":"slugify(value, allow_unicode=False)","text":"Taken from https://github.com/django/django/blob/master/django/utils/text.py Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated dashes to single dashes. Remove characters that aren't alphanumerics, underscores, or hyphens. Convert to lowercase. Also strip leading and trailing whitespace, dashes, and underscores.
Source code in video_sampler/utils.py def slugify(value, allow_unicode=False):\n \"\"\"\n Taken from https://github.com/django/django/blob/master/django/utils/text.py\n Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated\n dashes to single dashes. Remove characters that aren't alphanumerics,\n underscores, or hyphens. Convert to lowercase. Also strip leading and\n trailing whitespace, dashes, and underscores.\n \"\"\"\n value = str(value)\n if allow_unicode:\n value = unicodedata.normalize(\"NFKC\", value)\n else:\n value = (\n unicodedata.normalize(\"NFKD\", value)\n .encode(\"ascii\", \"ignore\")\n .decode(\"ascii\")\n )\n value = re.sub(r\"[^\\w\\s-]\", \"\", value.lower())\n return re.sub(r\"[-\\s]+\", \"-\", value).strip(\"-_\")\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/","title":"Integrations","text":""},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin","title":"YTDLPPlugin","text":"A plugin for yt-dlp to generate URLs and corresponding titles from the given URL. Methods: generate_urls(url, extra_yt_constr_args=None, extra_info_extract_opts=None) -> Iterable[str]: Generates URLs and corresponding titles from the given URL.
Source code in video_sampler/integrations/yt_dlp_plugin.py class YTDLPPlugin:\n \"\"\"\n A plugin for yt-dlp to generate URLs and corresponding titles from the given URL.\n Methods:\n generate_urls(url, extra_yt_constr_args=None, extra_info_extract_opts=None) -> Iterable[str]:\n Generates URLs and corresponding titles from the given URL.\n\n \"\"\"\n\n def __init__(self, ie_key: str = \"Generic\"):\n \"\"\"\n Initialize the YTDLPPlugin instance.\n :param ie_key (str): The key for the information extractor.\n \"\"\"\n self.ie_key = ie_key\n self.ydl_opts = {\n \"format\": best_video_only,\n }\n\n def generate_urls(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n ) -> Iterable[str]:\n \"\"\"Generate URLs and corresponding titles from the given URL.\n\n :param url (str): The URL to extract information from.\n :param extra_info_extract_opts (dict, optional): Extra options for information extraction.\n\n :return Iterable[str]:\n Tuple[str, str]: A tuple containing the title and URL of each extracted entry.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(params=(self.ydl_opts | extra_info_extract_opts)) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n if \"entries\" not in info:\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"]\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n yield entry[\"title\"], req_format[\"url\"]\n\n def get_subtitles_opts(self, no_download: bool = False) -> dict:\n return {\n \"postprocessors\": [\n {\n \"format\": \"srt\",\n \"key\": \"FFmpegSubtitlesConvertor\",\n \"when\": \"before_dl\",\n }\n ],\n \"format\": best_video_only,\n \"subtitleslangs\": [\"en.*\"],\n \"writeautomaticsub\": True,\n \"writesubtitles\": True,\n }\n\n def generate_urls_by_subs(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n ):\n \"\"\"Download subtitles for a given video URL.\n\n :param video_url (str): The URL of the video to download subtitles for.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(\n params=(self.ydl_opts | extra_info_extract_opts | self.get_subtitles_opts())\n ) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n import json\n\n json.dump(info, open(\"info.json\", \"w\"))\n if \"entries\" not in info:\n req_subs = list(info[\"requested_subtitles\"].values())[0]\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"], download_sub(req_subs[\"url\"])\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n req_subs = list(entry[\"requested_subtitles\"].values())[0]\n yield entry[\"title\"], req_format[\"url\"], download_sub(\n req_subs[\"url\"]\n )\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.__init__","title":"__init__(ie_key='Generic')","text":"Initialize the YTDLPPlugin instance. :param ie_key (str): The key for the information extractor.
Source code in video_sampler/integrations/yt_dlp_plugin.py def __init__(self, ie_key: str = \"Generic\"):\n \"\"\"\n Initialize the YTDLPPlugin instance.\n :param ie_key (str): The key for the information extractor.\n \"\"\"\n self.ie_key = ie_key\n self.ydl_opts = {\n \"format\": best_video_only,\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.generate_urls","title":"generate_urls(url, extra_info_extract_opts=None)","text":"Generate URLs and corresponding titles from the given URL.
:param url (str): The URL to extract information from. :param extra_info_extract_opts (dict, optional): Extra options for information extraction.
:return Iterable[str]: Tuple[str, str]: A tuple containing the title and URL of each extracted entry.
Source code in video_sampler/integrations/yt_dlp_plugin.py def generate_urls(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n) -> Iterable[str]:\n \"\"\"Generate URLs and corresponding titles from the given URL.\n\n :param url (str): The URL to extract information from.\n :param extra_info_extract_opts (dict, optional): Extra options for information extraction.\n\n :return Iterable[str]:\n Tuple[str, str]: A tuple containing the title and URL of each extracted entry.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(params=(self.ydl_opts | extra_info_extract_opts)) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n if \"entries\" not in info:\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"]\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n yield entry[\"title\"], req_format[\"url\"]\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.generate_urls_by_subs","title":"generate_urls_by_subs(url, extra_info_extract_opts=None)","text":"Download subtitles for a given video URL.
:param video_url (str): The URL of the video to download subtitles for.
Source code in video_sampler/integrations/yt_dlp_plugin.py def generate_urls_by_subs(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n):\n \"\"\"Download subtitles for a given video URL.\n\n :param video_url (str): The URL of the video to download subtitles for.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(\n params=(self.ydl_opts | extra_info_extract_opts | self.get_subtitles_opts())\n ) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n import json\n\n json.dump(info, open(\"info.json\", \"w\"))\n if \"entries\" not in info:\n req_subs = list(info[\"requested_subtitles\"].values())[0]\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"], download_sub(req_subs[\"url\"])\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n req_subs = list(entry[\"requested_subtitles\"].values())[0]\n yield entry[\"title\"], req_format[\"url\"], download_sub(\n req_subs[\"url\"]\n )\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.best_video_best_audio","title":"best_video_best_audio(ctx)","text":"Taken from the yt-dlp documentation as-is
Source code in video_sampler/integrations/yt_dlp_plugin.py def best_video_best_audio(ctx):\n \"\"\"Taken from the yt-dlp documentation as-is\"\"\"\n \"\"\"Select the best video and the best audio that won't result in an mkv.\n NOTE: This is just an example and does not handle all cases\"\"\"\n\n # formats are already sorted worst to best\n formats = ctx.get(\"formats\")[::-1]\n\n # acodec='none' means there is no audio\n best_video = next(\n f for f in formats if f[\"vcodec\"] != \"none\" and f[\"acodec\"] == \"none\"\n )\n\n # find compatible audio extension\n audio_ext = {\"mp4\": \"m4a\", \"webm\": \"webm\"}[best_video[\"ext\"]]\n # vcodec='none' means there is no video\n best_audio = next(\n f\n for f in formats\n if (f[\"acodec\"] != \"none\" and f[\"vcodec\"] == \"none\" and f[\"ext\"] == audio_ext)\n )\n\n # These are the minimum required fields for a merged format\n yield {\n \"format_id\": f'{best_video[\"format_id\"]}+{best_audio[\"format_id\"]}',\n \"ext\": best_video[\"ext\"],\n \"requested_formats\": [best_video, best_audio],\n # Must be + separated list of protocols\n \"protocol\": f'{best_video[\"protocol\"]}+{best_audio[\"protocol\"]}',\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.best_video_only","title":"best_video_only(ctx)","text":"Just best video -- save bandwidth
Source code in video_sampler/integrations/yt_dlp_plugin.py def best_video_only(ctx):\n \"\"\"Just best video -- save bandwidth\"\"\"\n # formats are already sorted worst to best\n formats = ctx.get(\"formats\")[::-1]\n\n # acodec='none' means there is no audio\n best_video = next(f for f in formats if f[\"vcodec\"] != \"none\")\n # These are the minimum required fields for a merged format\n yield {\n \"format_id\": f'{best_video[\"format_id\"]}',\n \"ext\": best_video[\"ext\"],\n \"requested_formats\": [best_video],\n # Must be + separated list of protocols\n \"protocol\": f'{best_video[\"protocol\"]}',\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.no_shorts","title":"no_shorts(info, *, incomplete)","text":"Filter out short videos
Source code in video_sampler/integrations/yt_dlp_plugin.py def no_shorts(info, *, incomplete):\n \"\"\"Filter out short videos\"\"\"\n if url := info.get(\"url\", \"\"):\n if \"/shorts\" in url:\n return \"This is a short video\"\n
"},{"location":"reference/video_sampler/language/keyword_capture/","title":"Language","text":""},{"location":"reference/video_sampler/language/keyword_capture/#video_sampler.language.keyword_capture.download_sub","title":"download_sub(sub_url)","text":"Download a VTT subtitle file to a string.
Source code in video_sampler/language/keyword_capture.py def download_sub(sub_url: str):\n \"\"\"Download a VTT subtitle file to a string.\"\"\"\n response = requests.get(url=sub_url)\n return parse_srt_subtitle(response.text)\n
"},{"location":"reference/video_sampler/visualisation/clustering/","title":"Visualisation","text":""},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.build_feature_model","title":"build_feature_model(model_str)","text":"Build a feature extraction model :param model_str: model name :return: tuple of (model, extractor)
Source code in video_sampler/visualisation/clustering.py def build_feature_model(model_str: str):\n \"\"\"Build a feature extraction model\n :param model_str: model name\n :return: tuple of (model, extractor)\n \"\"\"\n extractor = AutoFeatureExtractor.from_pretrained(model_str)\n model = ResNetModel.from_pretrained(model_str)\n return model, extractor\n
"},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.cluster_features","title":"cluster_features(features, max_clusters=50)","text":"Cluster features using t-SNE and KMeans :param features: dict with keys \"embeds\" and \"paths\" :param max_clusters: maximum number of clusters :return: tuple of (X, cluster_labels)
Source code in video_sampler/visualisation/clustering.py def cluster_features(\n features,\n max_clusters=50,\n):\n \"\"\"Cluster features using t-SNE and KMeans\n :param features: dict with keys \"embeds\" and \"paths\"\n :param max_clusters: maximum number of clusters\n :return: tuple of (X, cluster_labels)\n \"\"\"\n proj = TSNE(n_components=2, perplexity=35, metric=\"cosine\")\n Xorg = np.asarray(features[\"embeds\"])\n X = proj.fit_transform(Xorg)\n\n # take about 10% of the frame as the number of clusters\n n_clusters = min(int(0.1 * len(features[\"embeds\"])), max_clusters)\n cluster_model = KMeans(n_clusters=n_clusters, random_state=0).fit(Xorg)\n return X, cluster_model.labels_\n
"},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.extract_features","title":"extract_features(model_str, image_folder, mkey='pixel_values', batch_size=8)","text":"Extract features from a folder of images :param model_str: model name :param image_folder: folder with images :param mkey: key for the pixel values :param batch_size: batch size :return: dict with keys \"embeds\" and \"paths\"
Source code in video_sampler/visualisation/clustering.py def extract_features(\n model_str: str, image_folder: Path, mkey=\"pixel_values\", batch_size: int = 8\n):\n \"\"\"Extract features from a folder of images\n :param model_str: model name\n :param image_folder: folder with images\n :param mkey: key for the pixel values\n :param batch_size: batch size\n :return: dict with keys \"embeds\" and \"paths\"\n \"\"\"\n\n out_features = defaultdict(list)\n model, extractor = build_feature_model(model_str)\n with torch.no_grad():\n all_files = list(image_folder.iterdir())\n for batch in tqdm(\n batched(all_files, batch_size), total=len(all_files) // batch_size\n ):\n # load images\n batch_imgs = [Image.open(img_path).convert(\"RGB\") for img_path in batch]\n # extract features\n batch_imgs = extractor(batch_imgs, return_tensors=\"pt\")[mkey]\n batch_features = model(batch_imgs).pooler_output.squeeze()\n if len(batch) == 1:\n batch_features = batch_features.expand(1, -1)\n batch_features = torch.functional.F.normalize(batch_features, p=2, dim=1)\n out_features[\"embeds\"].extend(batch_features)\n out_features[\"paths\"].extend([img_path.name for img_path in batch])\n return out_features\n
"}]}
\ No newline at end of file
+{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"video-sampler","text":"Video sampler allows you to efficiently sample video frames. Currently, it uses keyframe decoding, frame interval gating and perceptual hashing to reduce duplicated samples.
Use case: for sampling videos for later annotations used in machine learning.
"},{"location":"#table-of-contents","title":"Table of Contents","text":" - video-sampler
- Table of Contents
- Documentation
- Features
- Installation and Usage
- Basic usage
- YT-DLP integration plugin
- API examples
- Advanced usage
- Gating
- CLIP-based gating comparison
- Blur gating
- Benchmarks
- Benchmark videos
- Flit commands
- \ud83d\udee1 License
- \ud83d\udcc3 Citation
"},{"location":"#documentation","title":"Documentation","text":"Documentation is available at https://lemurpwned.github.io/video-sampler/.
"},{"location":"#features","title":"Features","text":" - [x] Direct sampling methods:
- [x]
hash - uses perceptual hashing to reduce duplicated samples - [x]
entropy - uses entropy to reduce duplicated samples (work in progress) - [x]
gzip - uses gzip compressed size to reduce duplicated samples (work in progress) - [x]
buffer - uses sliding buffer to reduce duplicated samples - [x]
grid - uses grid sampling to reduce duplicated samples - [x] Gating methods (modifications on top of direct sampling methods):
- [x]
clip - uses CLIP to filter out frames that do not contain the specified objects - [x]
blur - uses blur detection to filter out frames that are too blurry - [x] Integrations
- [x] YTDLP integration -- streams directly from yt-dlp queries, playlists or single videos
"},{"location":"#installation-and-usage","title":"Installation and Usage","text":"pip install -U video_sampler\n
then you can run
python3 -m video_sampler --help\n
or simply
video_sampler --help\n
"},{"location":"#basic-usage","title":"Basic usage","text":"python3 -m video_sampler hash FatCat.mp4 ./dataset-frames/ --hash-size 3 --buffer-size 20\n
"},{"location":"#yt-dlp-integration-plugin","title":"YT-DLP integration plugin","text":"Before using please consult the ToS of the website you are scraping from -- use responsibly and for research purposes. To use the YT-DLP integration, you need to install yt-dlp first (see yt-dlp). Then, you simply add --yt-dlp to the command, and it changes the meaning of the video_path argument.
video_sampler hash \"ytsearch:cute cats\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
video_sampler hash \"https://www.youtube.com/watch?v=W86cTIoMv2U\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
video_sampler hash \"https://www.youtube.com/watch?v=GbpP3Sxp-1U&list=PLFezMcAw96RGvTTTbdKrqew9seO2ZGRmk\" ./folder-frames/ \\\n --hash-size 3 --buffer-size 20 --ytdlp\n
The videos are never directly downloaded, only streamed, so you can use it to sample videos from the internet without downloading them first.
"},{"location":"#extra-yt-dlp-options","title":"Extra YT-DLP options","text":"You can pass extra options to yt-dlp by using the -yt-extra-args flag. For example:
this will only sample videos uploaded before 2019-01-01:
... --ytdlp --yt-extra-args '--datebefore 20190101'\n
or this will only sample videos uploaded after 2019-01-01:
... --ytdlp --yt-extra-args '--dateafter 20190101'\n
or this will skip all shorts:
... --ytdlp --yt-extra-args '--match-filter \"original_url!*=/shorts/ & url!*=/shorts/\"\n
"},{"location":"#api-examples","title":"API examples","text":"See examples in https://github.com/LemurPwned/video-sampler/tree/main/scripts.
"},{"location":"#advanced-usage","title":"Advanced usage","text":"There are 3 sampling methods available:
hash - uses perceptual hashing to reduce duplicated samplesentropy - uses entropy to reduce duplicated samples (work in progress)gzip - uses gzip compressed size to reduce duplicated samples (work in progress)
To launch any of them you can run and substitute method-name with one of the above:
video_sampler buffer `method-name` ...other options\n
e.g.
video_sampler buffer entropy --buffer-size 20 ...\n
where buffer-size for entropy and gzip mean the top-k sliding buffer size. Sliding buffer also uses hashing to reduce duplicated samples.
"},{"location":"#gating","title":"Gating","text":"Aside from basic sampling rules, you can also apply gating rules to the sampled frames, further reducing the number of frames. There are 3 gating methods available:
pass - pass all framesclip - use CLIP to filter out frames that do not contain the specified objectsblur - use blur detection to filter out frames that are too blurry
Here's a quick example of how to use clip:
python3 -m video_sampler clip ./videos ./scratch/clip --pos-samples \"a cat\" --neg-samples \"empty background, a lemur\" --hash-size 4\n
"},{"location":"#clip-based-gating-comparison","title":"CLIP-based gating comparison","text":"Here's a brief comparison of the frames sampled with and without CLIP-based gating with the following config:
gate_def = dict(\n type=\"clip\",\n pos_samples=[\"a cat\"],\n neg_samples=[\n \"an empty background\",\n \"text on screen\",\n \"a forest with no animals\",\n ],\n model_name=\"ViT-B-32\",\n batch_size=32,\n pos_margin=0.2,\n neg_margin=0.3,\n )\n
Evidently, CLIP-based gating is able to filter out frames that do not contain a cat and in consequence, reduce the number of frames with plain background. It also thinks that a lemur is a cat, which is not entirely wrong as fluffy creatures go.
Pass gate (no gating) CLIP gate Grid The effects of gating in numbers, for this particular set of examples (see produced vs gated columns). produced represents the number of frames sampled without gating, here after the perceptual hashing, while gated represents the number of frames sampled after gating.
video buffer gate decoded produced gated FatCat.mp4 grid pass 179 31 31 SmolCat.mp4 grid pass 118 24 24 HighLemurs.mp4 grid pass 161 35 35 FatCat.mp4 hash pass 179 101 101 SmolCat.mp4 hash pass 118 61 61 HighLemurs.mp4 hash pass 161 126 126 FatCat.mp4 hash clip 179 101 73 SmolCat.mp4 hash clip 118 61 31 HighLemurs.mp4 hash clip 161 126 66"},{"location":"#blur-gating","title":"Blur gating","text":"Helps a little with blurry videos. Adjust threshold and method (laplacian or fft) for best results. Some results from fft at threshold=20:
video buffer gate decoded produced gated MadLad.mp4 grid pass 120 31 31 MadLad.mp4 hash pass 120 110 110 MadLad.mp4 hash blur 120 110 85"},{"location":"#benchmarks","title":"Benchmarks","text":"Configuration for this benchmark:
SamplerConfig(min_frame_interval_sec=1.0, keyframes_only=True, buffer_size=30, hash_size=X, queue_wait=0.1, debug=True)\n
Video Total frames Hash size Decoded Saved SmolCat 2936 8 118 106 SmolCat - 4 - 61 Fat Cat 4462 8 179 163 Fat Cat - 4 - 101 HighLemurs 4020 8 161 154 HighLemurs - 4 - 126 SamplerConfig(\n min_frame_interval_sec=1.0,\n keyframes_only=True,\n queue_wait=0.1,\n debug=False,\n print_stats=True,\n buffer_config={'type': 'entropy'/'gzip', 'size': 30, 'debug': False, 'hash_size': 8, 'expiry': 50}\n)\n
Video Total frames Type Decoded Saved SmolCat 2936 entropy 118 39 SmolCat - gzip - 39 Fat Cat 4462 entropy 179 64 Fat Cat - gzip - 73 HighLemurs 4020 entropy 161 59 HighLemurs - gzip - 63"},{"location":"#benchmark-videos","title":"Benchmark videos","text":" - SmolCat
- Fat Cat
- HighLemurs
- MadLad
"},{"location":"#flit-commands","title":"Flit commands","text":""},{"location":"#build","title":"Build","text":"flit build\n
"},{"location":"#install","title":"Install","text":"flit install\n
"},{"location":"#publish","title":"Publish","text":"Remember to bump the version in pyproject.toml before publishing.
flit publish\n
"},{"location":"#license","title":"\ud83d\udee1 License","text":"This project is licensed under the terms of the MIT license. See LICENSE for more details.
"},{"location":"#citation","title":"\ud83d\udcc3 Citation","text":"@misc{video-sampler,\n author = {video-sampler},\n title = {Video sampler allows you to efficiently sample video frames},\n year = {2023},\n publisher = {GitHub},\n journal = {GitHub repository},\n howpublished = {\\url{https://github.com/LemurPwned/video-sampler}}\n}\n
"},{"location":"reference/video_sampler/buffer/","title":"Video sampler","text":""},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.EntropyByffer","title":"EntropyByffer","text":" Bases: FrameBuffer
Measure image entropy as a function of the image usability
Source code in video_sampler/buffer.py class EntropyByffer(FrameBuffer):\n \"\"\"Measure image entropy as a function of the image usability\"\"\"\n\n def __init__(\n self, size: int, expiry: int, debug_flag: bool = False, hash_size: int = 8\n ) -> None:\n self.sliding_top_k_buffer = SlidingTopKBuffer(\n size=size, expiry=expiry, debug_flag=debug_flag, hash_size=hash_size\n )\n\n def get_buffer_state(self) -> list[str]:\n return self.sliding_top_k_buffer.get_buffer_state()\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n entropy = item.entropy()\n return self.sliding_top_k_buffer.add(item, {**metadata, \"index\": -entropy})\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n return self.sliding_top_k_buffer.final_flush()\n\n def clear(self):\n self.sliding_top_k_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer","title":"FrameBuffer","text":" Bases: ABC
Source code in video_sampler/buffer.py class FrameBuffer(ABC):\n @abstractmethod\n def add(self, item: Image.Image, metadata: dict[str, Any]) -> None | tuple:\n pass\n\n @abstractmethod\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n \"\"\"Flush the buffer and return the remaining items\"\"\"\n pass\n\n @abstractmethod\n def get_buffer_state(self) -> list[str]:\n \"\"\"Return the current state of the buffer\"\"\"\n pass\n\n @abstractmethod\n def clear(self):\n \"\"\"Clear the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.clear","title":"clear() abstractmethod","text":"Clear the buffer
Source code in video_sampler/buffer.py @abstractmethod\ndef clear(self):\n \"\"\"Clear the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.final_flush","title":"final_flush() abstractmethod","text":"Flush the buffer and return the remaining items
Source code in video_sampler/buffer.py @abstractmethod\ndef final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n \"\"\"Flush the buffer and return the remaining items\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.FrameBuffer.get_buffer_state","title":"get_buffer_state() abstractmethod","text":"Return the current state of the buffer
Source code in video_sampler/buffer.py @abstractmethod\ndef get_buffer_state(self) -> list[str]:\n \"\"\"Return the current state of the buffer\"\"\"\n pass\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GridBuffer","title":"GridBuffer","text":" Bases: HashBuffer
A class representing a grid-based buffer for images. Splits the image into a grid and stores the hashes of the grid cells in a mosaic buffer.
Parameters:
Name Type Description Default size int The maximum size of the buffer.
required debug_flag bool A flag indicating whether debug information should be printed.
False hash_size int The size of the hash.
4 grid_x int The number of grid cells in the x-axis.
4 grid_y int The number of grid cells in the y-axis.
4 max_hits int The maximum number of hits allowed for a hash.
1 Attributes:
Name Type Description grid_x int The number of grid cells in the x-axis.
grid_y int The number of grid cells in the y-axis.
max_hits int The maximum number of hits allowed for a hash.
mosaic_buffer dict A dictionary storing the mosaic buffer.
Methods:
Name Description add Adds an image to the buffer along with its metadata.
clear Clears the buffer and the mosaic buffer.
update_ttl_buffer Updates the buffer by expiring images that are not in the grid.
Source code in video_sampler/buffer.py class GridBuffer(HashBuffer):\n \"\"\"\n A class representing a grid-based buffer for images.\n Splits the image into a grid and stores the hashes of the grid cells in a mosaic buffer.\n\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): A flag indicating whether debug information should be printed.\n hash_size (int, optional): The size of the hash.\n grid_x (int, optional): The number of grid cells in the x-axis.\n grid_y (int, optional): The number of grid cells in the y-axis.\n max_hits (int, optional): The maximum number of hits allowed for a hash.\n\n Attributes:\n grid_x (int): The number of grid cells in the x-axis.\n grid_y (int): The number of grid cells in the y-axis.\n max_hits (int): The maximum number of hits allowed for a hash.\n mosaic_buffer (dict): A dictionary storing the mosaic buffer.\n\n Methods:\n add(item, metadata):\n Adds an image to the buffer along with its metadata.\n clear():\n Clears the buffer and the mosaic buffer.\n update_ttl_buffer():\n Updates the buffer by expiring images that are not in the grid.\n\n \"\"\"\n\n def __init__(\n self,\n size: int,\n debug_flag: bool = False,\n hash_size: int = 4,\n grid_x: int = 4,\n grid_y: int = 4,\n max_hits: int = 1,\n ) -> None:\n super().__init__(size, debug_flag, hash_size)\n self.grid_x = grid_x\n self.grid_y = grid_y\n self.max_hits = max_hits\n self.mosaic_buffer = {}\n\n def __get_grid_hash(self, item: Image.Image) -> str:\n \"\"\"Compute grid hashes for a given image\"\"\"\n for x in range(self.grid_x):\n for y in range(self.grid_y):\n yield str(\n phash(\n item.crop(\n (\n x * item.width / self.grid_x,\n y * item.height / self.grid_y,\n (x + 1) * item.width / self.grid_x,\n (y + 1) * item.height / self.grid_y,\n )\n ),\n hash_size=self.hash_size,\n )\n )\n\n def _check_mosaic(self, mosaic_hash: str):\n return mosaic_hash in self.mosaic_buffer\n\n def update_ttl_buffer(self):\n # expire the images that are not in the grid\n if len(self.ordered_buffer) >= self.max_size:\n to_return_hash, return_data = self.ordered_buffer.popitem(last=False)\n if to_return_hash is not None:\n removal_keys = [\n img_hash\n for img_hash, mosaic_hash in self.mosaic_buffer.items()\n if mosaic_hash == to_return_hash\n ]\n for key in removal_keys:\n del self.mosaic_buffer[key]\n return return_data\n return None\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n hash_ = str(phash(item, hash_size=self.hash_size))\n if not self._check_duplicate(hash_):\n # not automatically rejected, check the mosaic buffer\n hash_hits = 0\n hash_sets = []\n for el_hash_ in self.__get_grid_hash(item):\n if el_hash_ in self.mosaic_buffer:\n hash_hits += 1\n hash_sets.append(el_hash_)\n\n if hash_hits < self.max_hits:\n # add image hash to the ttl counter\n self.ordered_buffer[hash_] = (item, metadata)\n # add the image to the mosaic buffer\n # this also automatically overwrites the deleted hashes\n for el_hash in hash_sets:\n self.mosaic_buffer[el_hash] = hash_\n\n if self.debug_flag:\n console.print(\n f\"\\tHash hits: {hash_hits}\"\n f\"\\tHash sets: {len(hash_sets)}\"\n f\"\\tHash buffer: {len(self.get_buffer_state())}\"\n f\"\\tMosaic buffer: {len(self.mosaic_buffer)}\"\n )\n return self.update_ttl_buffer()\n\n def clear(self):\n super().clear()\n self.mosaic_buffer = {}\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GridBuffer.__get_grid_hash","title":"__get_grid_hash(item)","text":"Compute grid hashes for a given image
Source code in video_sampler/buffer.py def __get_grid_hash(self, item: Image.Image) -> str:\n \"\"\"Compute grid hashes for a given image\"\"\"\n for x in range(self.grid_x):\n for y in range(self.grid_y):\n yield str(\n phash(\n item.crop(\n (\n x * item.width / self.grid_x,\n y * item.height / self.grid_y,\n (x + 1) * item.width / self.grid_x,\n (y + 1) * item.height / self.grid_y,\n )\n ),\n hash_size=self.hash_size,\n )\n )\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.GzipBuffer","title":"GzipBuffer","text":" Bases: FrameBuffer
Measure compression size as a function of the image usability
Source code in video_sampler/buffer.py class GzipBuffer(FrameBuffer):\n \"\"\"Measure compression size as a function of the image usability\"\"\"\n\n def __init__(\n self, size: int, expiry: int, debug_flag: bool = False, hash_size: int = 8\n ) -> None:\n self.sliding_top_k_buffer = SlidingTopKBuffer(\n size=size, expiry=expiry, debug_flag=debug_flag, hash_size=hash_size\n )\n\n def get_buffer_state(self) -> list[str]:\n return self.sliding_top_k_buffer.get_buffer_state()\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n compressed_l = len(gzip.compress(item.tobytes()))\n return self.sliding_top_k_buffer.add(item, {**metadata, \"index\": -compressed_l})\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n return self.sliding_top_k_buffer.final_flush()\n\n def clear(self):\n self.sliding_top_k_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.HashBuffer","title":"HashBuffer","text":" Bases: FrameBuffer
A buffer that stores frames with their corresponding metadata and checks for duplicates based on image hashes. Args: size (int): The maximum size of the buffer. debug_flag (bool, optional): Flag indicating whether to enable debug mode. Defaults to False. hash_size (int, optional): The size of the image hash. Defaults to 4.
Methods:
Name Description get_buffer_state Returns the current state of the buffer as a list of image hashes.
add Image.Image, metadata: dict[str, Any]) Adds an item to the buffer along with its metadata.
final_flush Yields the stored items and their metadata in the buffer.
Private Methods __add(item: Image.Image, hash_: str, metadata: dict) Adds an item to the buffer with the given hash and metadata.
__check_duplicate(hash_: str) -> bool: Checks if the given hash already exists in the buffer and renews its validity if found.
Source code in video_sampler/buffer.py class HashBuffer(FrameBuffer):\n \"\"\"\n A buffer that stores frames with their corresponding metadata and\n checks for duplicates based on image hashes.\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): Flag indicating whether to enable debug mode. Defaults to False.\n hash_size (int, optional): The size of the image hash. Defaults to 4.\n\n Methods:\n get_buffer_state() -> list[str]:\n Returns the current state of the buffer as a list of image hashes.\n\n add(item: Image.Image, metadata: dict[str, Any])\n Adds an item to the buffer along with its metadata.\n\n final_flush() -> Iterable[tuple[Image.Image | None, dict]]:\n Yields the stored items and their metadata in the buffer.\n\n clear()\n Clears the buffer.\n\n Private Methods:\n __add(item: Image.Image, hash_: str, metadata: dict)\n Adds an item to the buffer with the given hash and metadata.\n\n __check_duplicate(hash_: str) -> bool:\n Checks if the given hash already exists in the buffer and renews its validity if found.\n\n \"\"\"\n\n def __init__(self, size: int, debug_flag: bool = False, hash_size: int = 4) -> None:\n self.ordered_buffer = OrderedDict()\n self.max_size = size\n self.debug_flag = debug_flag\n self.hash_size = hash_size\n\n def get_buffer_state(self) -> list[str]:\n return list(self.ordered_buffer.keys())\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n hash_ = str(phash(item, hash_size=self.hash_size))\n if not self._check_duplicate(hash_):\n return self.__add(hash_, item, metadata)\n return None\n\n def __add(self, hash_: str, item: Image.Image, metadata: dict):\n self.ordered_buffer[hash_] = (item, metadata)\n if len(self.ordered_buffer) >= self.max_size:\n return self.ordered_buffer.popitem(last=False)[1]\n return None\n\n def _check_duplicate(self, hash_: str) -> bool:\n if hash_ in self.ordered_buffer:\n # renew the hash validity\n if self.debug_flag:\n console.print(\n f\"Renewing {hash_}\",\n style=f\"bold {Color.red.value}\",\n )\n self.ordered_buffer.move_to_end(hash_)\n return True\n return False\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n yield from self.ordered_buffer.values()\n\n def clear(self):\n self.ordered_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.SamplerConfig","title":"SamplerConfig dataclass","text":"Configuration options for the video sampler.
Parameters:
Name Type Description Default min_frame_interval_sec float The minimum time interval between sampled frames in seconds. Defaults to 1.
1 keyframes_only bool Flag indicating whether to sample only keyframes. Defaults to True.
True queue_wait float The time to wait between checking the frame queue in seconds. Defaults to 0.1.
0.1 debug bool Flag indicating whether to enable debug mode. Defaults to False.
False print_stats bool Flag indicating whether to print sampling statistics. Defaults to False.
False buffer_config dict[str, Any] Configuration options for the frame buffer. Defaults to {\"type\": \"entropy\", \"size\": 15, \"debug\": True}.
field(default_factory=lambda : {'type': 'hash', 'hash_size': 8, 'size': 15, 'debug': True}) gate_config dict[str, Any] Configuration options for the frame gate. Defaults to {\"type\": \"pass\"}.
field(default_factory=lambda : {'type': 'pass'}) Methods:
Name Description __str__ Returns a string representation of the configuration.
Source code in video_sampler/buffer.py @dataclass\nclass SamplerConfig:\n \"\"\"\n Configuration options for the video sampler.\n\n Args:\n min_frame_interval_sec (float, optional): The minimum time interval\n between sampled frames in seconds. Defaults to 1.\n keyframes_only (bool, optional): Flag indicating whether to\n sample only keyframes. Defaults to True.\n queue_wait (float, optional): The time to wait between checking\n the frame queue in seconds. Defaults to 0.1.\n debug (bool, optional): Flag indicating whether to enable debug mode.\n Defaults to False.\n print_stats (bool, optional): Flag indicating whether to print\n sampling statistics. Defaults to False.\n buffer_config (dict[str, Any], optional): Configuration options for\n the frame buffer. Defaults to {\"type\": \"entropy\", \"size\": 15,\n \"debug\": True}.\n gate_config (dict[str, Any], optional): Configuration options for\n the frame gate. Defaults to {\"type\": \"pass\"}.\n\n Methods:\n __str__() -> str:\n Returns a string representation of the configuration.\n\n \"\"\"\n\n min_frame_interval_sec: float = 1\n keyframes_only: bool = True\n queue_wait: float = 0.1\n debug: bool = False\n print_stats: bool = False\n buffer_config: dict[str, Any] = field(\n default_factory=lambda: {\n \"type\": \"hash\",\n \"hash_size\": 8,\n \"size\": 15,\n \"debug\": True,\n }\n )\n gate_config: dict[str, Any] = field(\n default_factory=lambda: {\n \"type\": \"pass\",\n }\n )\n\n def __str__(self) -> str:\n return str(asdict(self))\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.SlidingTopKBuffer","title":"SlidingTopKBuffer","text":" Bases: FrameBuffer
A class representing a sliding top-k buffer for frames.
Parameters:
Name Type Description Default size int The maximum size of the buffer.
required debug_flag bool A flag indicating whether debug information should be printed.
False expiry int The expiry count for frames.
30 hash_size int The size of the hash.
8 Attributes:
Name Type Description sliding_buffer list The sliding buffer implemented as a min heap.
max_size int The maximum size of the buffer.
debug_flag bool A flag indicating whether debug information should be printed.
expiry_count int The expiry count for frames.
hash_size int The size of the hash.
Methods:
Name Description get_buffer_state Returns the current state of the buffer.
add Adds a frame to the buffer along with its metadata.
final_flush Performs a final flush of the buffer and yields the remaining frames.
clear Clears the buffer.
Source code in video_sampler/buffer.py class SlidingTopKBuffer(FrameBuffer):\n \"\"\"\n A class representing a sliding top-k buffer for frames.\n\n Args:\n size (int): The maximum size of the buffer.\n debug_flag (bool, optional): A flag indicating whether debug information should be printed.\n expiry (int, optional): The expiry count for frames.\n hash_size (int, optional): The size of the hash.\n\n Attributes:\n sliding_buffer (list): The sliding buffer implemented as a min heap.\n max_size (int): The maximum size of the buffer.\n debug_flag (bool): A flag indicating whether debug information should be printed.\n expiry_count (int): The expiry count for frames.\n hash_size (int): The size of the hash.\n\n Methods:\n get_buffer_state() -> list[str]:\n Returns the current state of the buffer.\n add(item, metadata):\n Adds a frame to the buffer along with its metadata.\n final_flush() -> Iterable[tuple[Image.Image | None, dict]]:\n Performs a final flush of the buffer and yields the remaining frames.\n clear():\n Clears the buffer.\n\n \"\"\"\n\n def __init__(\n self, size: int, debug_flag: bool = False, expiry: int = 30, hash_size: int = 8\n ) -> None:\n # it's a min heap with a fixed size\n self.sliding_buffer = []\n self.max_size = size\n self.debug_flag = debug_flag\n self.expiry_count = expiry\n self.hash_size = hash_size\n assert (\n self.expiry_count > self.max_size\n ), \"expiry count must be greater than max size\"\n console.print(\n f\"Creating sliding buffer of size {self.max_size} and expiry {expiry}\",\n style=f\"bold {Color.red.value}\",\n )\n\n def get_buffer_state(self) -> list[str]:\n return [item[:3] for item in self.sliding_buffer]\n\n def add(self, item: Image.Image, metadata: dict[str, Any]):\n assert \"index\" in metadata, \"metadata must have index key for sliding buffer\"\n average_hash_ = str(average_hash(item, hash_size=self.hash_size))\n to_return = None\n if not self.__check_duplicate(average_hash_):\n heapq.heappush(\n self.sliding_buffer,\n [metadata[\"index\"], 0, average_hash_, item, metadata],\n )\n if len(self.sliding_buffer) >= self.max_size:\n to_return = heapq.heappop(self.sliding_buffer)[-2:]\n # update the expiry count\n expired_indx = -1\n for i in range(len(self.sliding_buffer)):\n self.sliding_buffer[i][1] += 1\n if self.sliding_buffer[i][1] >= self.expiry_count:\n expired_indx = i\n # at any point only one item can be expired\n if expired_indx != -1:\n self.sliding_buffer.pop(expired_indx) # just drop\n return to_return\n\n def __check_duplicate(self, hash_: str) -> bool:\n for item in self.sliding_buffer:\n if item[2] == hash_:\n # renew the hash validity\n if self.debug_flag:\n console.print(\n f\"Renewing {hash_}\",\n style=f\"bold {Color.red.value}\",\n )\n item[1] = 0\n return True\n return False\n\n def final_flush(self) -> Iterable[tuple[Image.Image | None, dict]]:\n if len(self.sliding_buffer):\n yield heapq.heappop(self.sliding_buffer)[-2:]\n yield None, {}\n\n def clear(self):\n self.sliding_buffer.clear()\n
"},{"location":"reference/video_sampler/buffer/#video_sampler.buffer.create_buffer","title":"create_buffer(buffer_config)","text":"Create a buffer based on the config
Source code in video_sampler/buffer.py def create_buffer(buffer_config: dict[str, Any]):\n \"\"\"Create a buffer based on the config\"\"\"\n console.print(\n f\"Creating buffer of type {buffer_config['type']}\",\n style=f\"bold {Color.red.value}\",\n )\n if buffer_config[\"type\"] == \"hash\":\n return HashBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n )\n elif buffer_config[\"type\"] == \"grid\":\n return GridBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n grid_x=buffer_config[\"grid_x\"],\n grid_y=buffer_config[\"grid_y\"],\n max_hits=buffer_config[\"max_hits\"],\n )\n elif buffer_config[\"type\"] == \"sliding_top_k\":\n return SlidingTopKBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n expiry=buffer_config[\"expiry\"],\n )\n elif buffer_config[\"type\"] == \"passthrough\":\n return PassThroughBuffer()\n elif buffer_config[\"type\"] == \"gzip\":\n return GzipBuffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n expiry=buffer_config[\"expiry\"],\n )\n elif buffer_config[\"type\"] == \"entropy\":\n return EntropyByffer(\n size=buffer_config[\"size\"],\n debug_flag=buffer_config[\"debug\"],\n hash_size=buffer_config[\"hash_size\"],\n expiry=buffer_config[\"expiry\"],\n )\n else:\n raise ValueError(f\"Unknown buffer type {buffer_config['type']}\")\n
"},{"location":"reference/video_sampler/evaluation/","title":"Evaluation","text":""},{"location":"reference/video_sampler/evaluation/#video_sampler.evaluation.compute_total_video_entropy","title":"compute_total_video_entropy()","text":"Compute the total entropy of a video
Source code in video_sampler/evaluation.py def compute_total_video_entropy():\n \"\"\"Compute the total entropy of a video\"\"\"\n pass\n
"},{"location":"reference/video_sampler/gating/","title":"Gating","text":""},{"location":"reference/video_sampler/gating/#video_sampler.gating.BlurGate","title":"BlurGate","text":"Source code in video_sampler/gating.py class BlurGate:\n def __init__(\n self, method: Literal[\"fft\", \"laplacian\"] = \"laplacian\", threshold: float = 100\n ) -> None:\n \"\"\"\n Initializes the Gating object.\n\n Args:\n method (str): The method to use for blur detection. Can be \"fft\" or \"laplacian\".\n threshold (float): The threshold for bluriness. The higher the threshold, the less\n blurry the image needs to be to be discarded.\n The default threshold values are:\n - 20 for the \"fft\" method\n - 100 for the \"laplacian\" method.\n\n Raises:\n ValueError: If an unknown blur method is provided.\n \"\"\"\n self.is_blurry = None\n if method == \"fft\":\n self.is_blurry = self._is_blurry_fft\n elif method == \"laplacian\":\n self.is_blurry = self._is_blurry_laplacian\n else:\n raise ValueError(f\"Unknown blur method {method}\")\n self.threshold = threshold\n\n def __call__(self, frame: Image.Image, meta: dict, last=False) -> GatedObject:\n if self.is_blurry(frame) or last:\n return EMPTY_GATED_OBJECT\n return GatedObject([FrameObject(frame, meta)], 1)\n\n def _is_blurry_laplacian(self, frame: Image.Image) -> bool:\n \"\"\"Check if the image is blurry with laplacian method.\"\"\"\n return (\n cv2.Laplacian(\n cv2.cvtColor(np.array(frame), cv2.COLOR_BGR2GRAY), cv2.CV_64F\n ).var()\n < self.threshold\n )\n\n def _is_blurry_fft(self, frame: Image.Image) -> bool:\n \"\"\"Check if the image is blurry with fft method.\"\"\"\n f = np.fft.fft2(frame)\n fshift = np.fft.fftshift(f)\n magnitude_spectrum = 20 * np.log(np.abs(fshift) + 1e-12)\n return magnitude_spectrum.mean() < self.threshold\n\n def flush(self):\n return EMPTY_GATED_OBJECT\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.BlurGate.__init__","title":"__init__(method='laplacian', threshold=100)","text":"Initializes the Gating object.
Parameters:
Name Type Description Default method str The method to use for blur detection. Can be \"fft\" or \"laplacian\".
'laplacian' threshold float The threshold for bluriness. The higher the threshold, the less blurry the image needs to be to be discarded. The default threshold values are: - 20 for the \"fft\" method - 100 for the \"laplacian\" method.
100 Raises:
Type Description ValueError If an unknown blur method is provided.
Source code in video_sampler/gating.py def __init__(\n self, method: Literal[\"fft\", \"laplacian\"] = \"laplacian\", threshold: float = 100\n) -> None:\n \"\"\"\n Initializes the Gating object.\n\n Args:\n method (str): The method to use for blur detection. Can be \"fft\" or \"laplacian\".\n threshold (float): The threshold for bluriness. The higher the threshold, the less\n blurry the image needs to be to be discarded.\n The default threshold values are:\n - 20 for the \"fft\" method\n - 100 for the \"laplacian\" method.\n\n Raises:\n ValueError: If an unknown blur method is provided.\n \"\"\"\n self.is_blurry = None\n if method == \"fft\":\n self.is_blurry = self._is_blurry_fft\n elif method == \"laplacian\":\n self.is_blurry = self._is_blurry_laplacian\n else:\n raise ValueError(f\"Unknown blur method {method}\")\n self.threshold = threshold\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.ClipGate","title":"ClipGate","text":"Source code in video_sampler/gating.py class ClipGate:\n def __init__(\n self,\n pos_samples: list[str] = None,\n neg_samples: list[str] = None,\n model_name: str = \"ViT-B-32\",\n batch_size: int = 32,\n pos_margin: float = 0.2,\n neg_margin: float = 0.3,\n ) -> None:\n \"\"\"\n Initializes the Clip Gating object.\n\n Args:\n pos_samples (list[str], optional): List of positive samples. Defaults to None.\n neg_samples (list[str], optional): List of negative samples. Defaults to None.\n model_name (str, optional): Name of the model. Defaults to \"ViT-B-32\".\n batch_size (int, optional): Batch size. Defaults to 32.\n pos_margin (float, optional): Positive margin. Defaults to 0.2.\n neg_margin (float, optional): Negative margin. Defaults to 0.3.\n \"\"\"\n self.model, self.preprocess, self.tokenizer = create_model(\n model_name=model_name\n )\n self.pos_margin = pos_margin\n self.neg_margin = neg_margin\n self.batch_size = batch_size\n self.frame_accumulator = []\n self.metadata_accumulator = []\n if pos_samples is None:\n self.pos_samples = torch.zeros((1, 512))\n else:\n self.pos_samples = self._preproc_samples(pos_samples)\n if neg_samples is None:\n self.neg_samples = torch.zeros((1, 512))\n else:\n self.neg_samples = self._preproc_samples(neg_samples)\n\n def __call__(self, frame: Image.Image, meta: dict, last=False) -> Any:\n return self.flush() if last else self.add_frame(frame, meta)\n\n def _preproc_samples(self, sample_texts: list[str]):\n inputs = self.tokenizer(sample_texts)\n embeds = torch.zeros((len(sample_texts), 512))\n with torch.no_grad():\n for i, batch in enumerate(batched(inputs, n=self.batch_size)):\n batch = torch.stack(batch)\n text_embeds = self.model.encode_text(batch.to(DEVICE))\n embeds[i * self.batch_size : (i + 1) * self.batch_size] = (\n text_embeds.cpu()\n )\n embeds /= embeds.norm(dim=-1, keepdim=True)\n return embeds\n\n def _embed_frames(self, frames: list[Image.Image]):\n \"\"\"Compute the embeddings for each frame.\"\"\"\n inputs = torch.stack([self.preprocess(frame) for frame in frames]).to(DEVICE)\n with torch.no_grad():\n image_embeds = self.model.encode_image(inputs).cpu()\n image_embeds /= image_embeds.norm(dim=-1, keepdim=True)\n return image_embeds\n\n def _get_margins(self, frame_embeds: torch.Tensor):\n \"\"\"Compute the margins for each frame.\"\"\"\n org_indx = np.arange(frame_embeds.shape[0])\n neg_distance = frame_embeds @ self.neg_samples.T\n pos_distance = frame_embeds @ self.pos_samples.T\n neg_margin, _ = neg_distance.max(axis=-1)\n pos_margin, _ = pos_distance.max(axis=-1)\n incl_samples = torch.argwhere(\n (neg_margin < self.neg_margin) & (pos_margin >= self.pos_margin)\n )\n return org_indx[incl_samples].ravel()\n\n def add_frame(self, frame: Image.Image, metadata: dict) -> GatedObject:\n self.frame_accumulator.append(frame)\n self.metadata_accumulator.append(metadata)\n if len(self.frame_accumulator) == self.batch_size:\n return self.__process_metadata()\n return EMPTY_GATED_OBJECT\n\n def flush(self):\n return self.__process_metadata()\n\n def __process_metadata(self) -> GatedObject:\n frame_embeds = self._embed_frames(self.frame_accumulator)\n selected_frames = self._get_margins(frame_embeds)\n to_return = [\n FrameObject(self.frame_accumulator[i], self.metadata_accumulator[i])\n for i in range(len(self.frame_accumulator))\n if i in selected_frames\n ]\n self.frame_accumulator.clear()\n self.metadata_accumulator.clear()\n return GatedObject(to_return, len(selected_frames))\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.ClipGate.__init__","title":"__init__(pos_samples=None, neg_samples=None, model_name='ViT-B-32', batch_size=32, pos_margin=0.2, neg_margin=0.3)","text":"Initializes the Clip Gating object.
Parameters:
Name Type Description Default pos_samples list[str] List of positive samples. Defaults to None.
None neg_samples list[str] List of negative samples. Defaults to None.
None model_name str Name of the model. Defaults to \"ViT-B-32\".
'ViT-B-32' batch_size int Batch size. Defaults to 32.
32 pos_margin float Positive margin. Defaults to 0.2.
0.2 neg_margin float Negative margin. Defaults to 0.3.
0.3 Source code in video_sampler/gating.py def __init__(\n self,\n pos_samples: list[str] = None,\n neg_samples: list[str] = None,\n model_name: str = \"ViT-B-32\",\n batch_size: int = 32,\n pos_margin: float = 0.2,\n neg_margin: float = 0.3,\n) -> None:\n \"\"\"\n Initializes the Clip Gating object.\n\n Args:\n pos_samples (list[str], optional): List of positive samples. Defaults to None.\n neg_samples (list[str], optional): List of negative samples. Defaults to None.\n model_name (str, optional): Name of the model. Defaults to \"ViT-B-32\".\n batch_size (int, optional): Batch size. Defaults to 32.\n pos_margin (float, optional): Positive margin. Defaults to 0.2.\n neg_margin (float, optional): Negative margin. Defaults to 0.3.\n \"\"\"\n self.model, self.preprocess, self.tokenizer = create_model(\n model_name=model_name\n )\n self.pos_margin = pos_margin\n self.neg_margin = neg_margin\n self.batch_size = batch_size\n self.frame_accumulator = []\n self.metadata_accumulator = []\n if pos_samples is None:\n self.pos_samples = torch.zeros((1, 512))\n else:\n self.pos_samples = self._preproc_samples(pos_samples)\n if neg_samples is None:\n self.neg_samples = torch.zeros((1, 512))\n else:\n self.neg_samples = self._preproc_samples(neg_samples)\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.PassGate","title":"PassGate","text":"Source code in video_sampler/gating.py class PassGate:\n def __call__(self, frame: Image.Image, meta: dict, last=False) -> GatedObject:\n \"\"\"\n Passes the frame through the gating mechanism.\n\n Args:\n frame (Image.Image): The frame to pass through.\n meta (dict): The metadata for the frame.\n last (bool): If this is the last frame in the video.\n\n Returns:\n GatedObject: The gated object containing the processed frame.\n \"\"\"\n return self.flush() if last else GatedObject([FrameObject(frame, meta)], 1)\n\n def flush(self):\n return EMPTY_GATED_OBJECT\n
"},{"location":"reference/video_sampler/gating/#video_sampler.gating.PassGate.__call__","title":"__call__(frame, meta, last=False)","text":"Passes the frame through the gating mechanism.
Parameters:
Name Type Description Default frame Image The frame to pass through.
required meta dict The metadata for the frame.
required last bool If this is the last frame in the video.
False Returns:
Name Type Description GatedObject GatedObject The gated object containing the processed frame.
Source code in video_sampler/gating.py def __call__(self, frame: Image.Image, meta: dict, last=False) -> GatedObject:\n \"\"\"\n Passes the frame through the gating mechanism.\n\n Args:\n frame (Image.Image): The frame to pass through.\n meta (dict): The metadata for the frame.\n last (bool): If this is the last frame in the video.\n\n Returns:\n GatedObject: The gated object containing the processed frame.\n \"\"\"\n return self.flush() if last else GatedObject([FrameObject(frame, meta)], 1)\n
"},{"location":"reference/video_sampler/iterators/","title":"Iterators","text":""},{"location":"reference/video_sampler/logging/","title":"Logging","text":""},{"location":"reference/video_sampler/sampler/","title":"Sampler","text":""},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.SegmentSampler","title":"SegmentSampler","text":" Bases: VideoSampler
Source code in video_sampler/sampler.py class SegmentSampler(VideoSampler):\n def __init__(\n self, cfg: SamplerConfig, segment_generator: Iterable[subtitle_line]\n ) -> None:\n super().__init__(cfg)\n self.segment_generator: Iterable[subtitle_line] = segment_generator\n\n def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video.\n\n Args:\n video_path (str): The path to the video file.\n\n Yields:\n Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.\n \"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n next_segment = next(self.segment_generator)\n segment_boundary_end_sec = next_segment.end_time / 1000\n segment_boundary_start_sec = next_segment.start_time / 1000\n absolute_stop = False\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n ftime = frame.time\n reiters = 0\n # find the next segment that starts after the current frame\n while ftime > segment_boundary_end_sec:\n console.print(\n f\"Seeking to next segment: {segment_boundary_end_sec}/{ftime}\",\n style=f\"bold {Color.yellow.value}\",\n )\n try:\n next_segment = next(self.segment_generator)\n reiters += 1\n segment_boundary_end_sec = next_segment.end_time / 1000\n segment_boundary_start_sec = next_segment.start_time / 1000\n except StopIteration:\n absolute_stop = True\n break\n if reiters > 0:\n console.print(\n f\"Skipped {reiters} segments!\",\n style=f\"bold {Color.red.value}\",\n )\n if absolute_stop:\n break\n # we haven't found the next segment yet\n # the other condition, is where we are after the segment\n # but this is handled by the while loop above\n if ftime <= segment_boundary_start_sec:\n continue\n\n self.stats[\"total\"] += 1\n time_diff = ftime - prev_time\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = ftime\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {ftime}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": ftime, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n yield from self.flush_buffer()\n\n def write_queue(self, video_path: str, q: Queue):\n super().write_queue(video_path, q)\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.SegmentSampler.sample","title":"sample(video_path)","text":"Generate sample frames from a video.
Parameters:
Name Type Description Default video_path str The path to the video file.
required Yields:
Type Description Iterable[list[FrameObject]] Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.
Source code in video_sampler/sampler.py def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video.\n\n Args:\n video_path (str): The path to the video file.\n\n Yields:\n Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.\n \"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n next_segment = next(self.segment_generator)\n segment_boundary_end_sec = next_segment.end_time / 1000\n segment_boundary_start_sec = next_segment.start_time / 1000\n absolute_stop = False\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n ftime = frame.time\n reiters = 0\n # find the next segment that starts after the current frame\n while ftime > segment_boundary_end_sec:\n console.print(\n f\"Seeking to next segment: {segment_boundary_end_sec}/{ftime}\",\n style=f\"bold {Color.yellow.value}\",\n )\n try:\n next_segment = next(self.segment_generator)\n reiters += 1\n segment_boundary_end_sec = next_segment.end_time / 1000\n segment_boundary_start_sec = next_segment.start_time / 1000\n except StopIteration:\n absolute_stop = True\n break\n if reiters > 0:\n console.print(\n f\"Skipped {reiters} segments!\",\n style=f\"bold {Color.red.value}\",\n )\n if absolute_stop:\n break\n # we haven't found the next segment yet\n # the other condition, is where we are after the segment\n # but this is handled by the while loop above\n if ftime <= segment_boundary_start_sec:\n continue\n\n self.stats[\"total\"] += 1\n time_diff = ftime - prev_time\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = ftime\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {ftime}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": ftime, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n yield from self.flush_buffer()\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.VideoSampler","title":"VideoSampler","text":"The fundamental class for sampling video frames.
Parameters:
Name Type Description Default cfg SamplerConfig The configuration for the video sampler.
required Attributes:
Name Type Description cfg SamplerConfig The configuration for the video sampler.
frame_buffer FrameBuffer The frame buffer used for sampling frames.
gate Gate The gate used for filtering frames.
stats Counter A counter for tracking statistics.
Methods:
Name Description sample Generates sample frames from a video.
write_queue Writes sampled frames to a queue.
Source code in video_sampler/sampler.py class VideoSampler:\n \"\"\"\n The fundamental class for sampling video frames.\n\n Args:\n cfg (SamplerConfig): The configuration for the video sampler.\n\n Attributes:\n cfg (SamplerConfig): The configuration for the video sampler.\n frame_buffer (FrameBuffer): The frame buffer used for sampling frames.\n gate (Gate): The gate used for filtering frames.\n stats (Counter): A counter for tracking statistics.\n\n Methods:\n sample(video_path) -> Iterable[list[FrameObject]]:\n Generates sample frames from a video.\n write_queue(video_path, q):\n Writes sampled frames to a queue.\n\n \"\"\"\n\n def __init__(self, cfg: SamplerConfig) -> None:\n self.cfg = deepcopy(cfg)\n self.frame_buffer = create_buffer(self.cfg.buffer_config)\n self.gate = create_gate(self.cfg.gate_config)\n self.stats = Counter()\n\n def flush_buffer(self):\n \"\"\"Flushes the frame buffer and yields gated frames\"\"\"\n for res in self.frame_buffer.final_flush():\n if res:\n self.stats[\"produced\"] += 1\n gated_obj = self.gate(*res)\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n gated_obj = self.gate.flush()\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n yield PROCESSING_DONE_ITERABLE\n\n def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video.\n\n Args:\n video_path (str): The path to the video file.\n\n Yields:\n Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.\n \"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n # skip frames if keyframes_only is True\n time_diff = frame.time - prev_time\n self.stats[\"total\"] += 1\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = frame.time\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {frame.time}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": frame.time, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n yield from self.flush_buffer()\n\n def write_queue(self, video_path: str, q: Queue):\n try:\n item: tuple[FrameObject, int]\n for item in self.sample(video_path=video_path):\n q.put(item)\n except (av.IsADirectoryError, av.InvalidDataError) as e:\n console.print(\n f\"Error while processing {video_path}\",\n f\"\\n\\t{e}\",\n style=f\"bold {Color.red.value}\",\n )\n q.put(PROCESSING_DONE_ITERABLE)\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.VideoSampler.flush_buffer","title":"flush_buffer()","text":"Flushes the frame buffer and yields gated frames
Source code in video_sampler/sampler.py def flush_buffer(self):\n \"\"\"Flushes the frame buffer and yields gated frames\"\"\"\n for res in self.frame_buffer.final_flush():\n if res:\n self.stats[\"produced\"] += 1\n gated_obj = self.gate(*res)\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n gated_obj = self.gate.flush()\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n yield PROCESSING_DONE_ITERABLE\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.VideoSampler.sample","title":"sample(video_path)","text":"Generate sample frames from a video.
Parameters:
Name Type Description Default video_path str The path to the video file.
required Yields:
Type Description Iterable[list[FrameObject]] Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.
Source code in video_sampler/sampler.py def sample(self, video_path: str) -> Iterable[list[FrameObject]]:\n \"\"\"Generate sample frames from a video.\n\n Args:\n video_path (str): The path to the video file.\n\n Yields:\n Iterable[list[FrameObject]]: A generator that yields a list of FrameObjects representing sampled frames.\n \"\"\"\n self.stats.clear()\n self.frame_buffer.clear()\n with av.open(video_path) as container:\n stream = container.streams.video[0]\n if self.cfg.keyframes_only:\n stream.codec_context.skip_frame = \"NONKEY\"\n prev_time = -10\n for frame_indx, frame in enumerate(container.decode(stream)):\n # skip frames if keyframes_only is True\n time_diff = frame.time - prev_time\n self.stats[\"total\"] += 1\n if time_diff < self.cfg.min_frame_interval_sec:\n continue\n prev_time = frame.time\n\n frame_pil: Image = frame.to_image()\n if self.cfg.debug:\n buf = self.frame_buffer.get_buffer_state()\n console.print(\n f\"Frame {frame_indx}\\ttime: {frame.time}\",\n f\"\\t Buffer ({len(buf)}): {buf}\",\n style=f\"bold {Color.green.value}\",\n )\n frame_meta = {\"frame_time\": frame.time, \"frame_indx\": frame_indx}\n self.stats[\"decoded\"] += 1\n if res := self.frame_buffer.add(\n frame_pil,\n metadata=frame_meta,\n ):\n gated_obj = self.gate(*res)\n self.stats[\"produced\"] += 1\n self.stats[\"gated\"] += gated_obj.N\n if gated_obj.frames:\n yield gated_obj.frames\n\n # flush buffer\n yield from self.flush_buffer()\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.Worker","title":"Worker","text":"Source code in video_sampler/sampler.py class Worker:\n def __init__(\n self,\n cfg: SamplerConfig,\n devnull: bool = False,\n processor_cls: VideoSampler = VideoSampler,\n extra_processor_args: dict = None,\n ) -> None:\n if extra_processor_args is None:\n extra_processor_args = {}\n self.cfg = cfg\n self.processor = processor_cls(cfg=cfg, **extra_processor_args)\n self.q = Queue()\n self.devnull = devnull\n\n def launch(\n self, video_path: str, output_path: str = \"\", pretty_video_name: str = \"\"\n ) -> None:\n \"\"\"\n Launch the worker.\n\n Args:\n video_path (str): Path to the video file.\n output_path (str, optional): Path to the output folder. Defaults to \"\".\n pretty_video_name (str, optional): Name of the video file for pretty printing (useful for urls).\n Defaults to \"\".\n \"\"\"\n if not pretty_video_name:\n pretty_video_name = os.path.basename(video_path)\n if output_path and self.devnull:\n raise ValueError(\"Cannot write to disk when devnull is True\")\n if output_path:\n os.makedirs(output_path, exist_ok=True)\n proc_thread = Thread(\n target=self.processor.write_queue, args=(video_path, self.q)\n )\n proc_thread.start()\n self.queue_reader(output_path, read_interval=self.cfg.queue_wait)\n proc_thread.join()\n if self.cfg.print_stats:\n console.print(\n f\"Stats for: {pretty_video_name}\",\n f\"\\n\\tTotal frames: {self.processor.stats['total']}\",\n f\"\\n\\tDecoded frames: {self.processor.stats['decoded']}\",\n f\"\\n\\tProduced frames: {self.processor.stats['produced']}\",\n f\"\\n\\tGated frames: {self.processor.stats['gated']}\",\n style=f\"bold {Color.magenta.value}\",\n )\n\n def queue_reader(self, output_path, read_interval=0.1) -> None:\n \"\"\"\n Reads frames from the queue and saves them as JPEG images.\n\n Args:\n output_path (str): The directory path where the frames will be saved.\n read_interval (float, optional): The time interval between reading frames from the queue.\n Defaults to 0.1 seconds.\n \"\"\"\n while True:\n if not self.q.empty():\n frame_object: FrameObject\n for frame_object in self.q.get():\n if frame_object.metadata.get(\"end\", False):\n return\n if frame_object.frame is not None and (\n not self.devnull and isinstance(frame_object.frame, Image.Image)\n ):\n frame_object.frame.save(\n os.path.join(\n output_path,\n f\"{frame_object.metadata['frame_time']}.jpg\",\n )\n )\n time.sleep(read_interval)\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.Worker.launch","title":"launch(video_path, output_path='', pretty_video_name='')","text":"Launch the worker.
Parameters:
Name Type Description Default video_path str Path to the video file.
required output_path str Path to the output folder. Defaults to \"\".
'' pretty_video_name str Name of the video file for pretty printing (useful for urls). Defaults to \"\".
'' Source code in video_sampler/sampler.py def launch(\n self, video_path: str, output_path: str = \"\", pretty_video_name: str = \"\"\n) -> None:\n \"\"\"\n Launch the worker.\n\n Args:\n video_path (str): Path to the video file.\n output_path (str, optional): Path to the output folder. Defaults to \"\".\n pretty_video_name (str, optional): Name of the video file for pretty printing (useful for urls).\n Defaults to \"\".\n \"\"\"\n if not pretty_video_name:\n pretty_video_name = os.path.basename(video_path)\n if output_path and self.devnull:\n raise ValueError(\"Cannot write to disk when devnull is True\")\n if output_path:\n os.makedirs(output_path, exist_ok=True)\n proc_thread = Thread(\n target=self.processor.write_queue, args=(video_path, self.q)\n )\n proc_thread.start()\n self.queue_reader(output_path, read_interval=self.cfg.queue_wait)\n proc_thread.join()\n if self.cfg.print_stats:\n console.print(\n f\"Stats for: {pretty_video_name}\",\n f\"\\n\\tTotal frames: {self.processor.stats['total']}\",\n f\"\\n\\tDecoded frames: {self.processor.stats['decoded']}\",\n f\"\\n\\tProduced frames: {self.processor.stats['produced']}\",\n f\"\\n\\tGated frames: {self.processor.stats['gated']}\",\n style=f\"bold {Color.magenta.value}\",\n )\n
"},{"location":"reference/video_sampler/sampler/#video_sampler.sampler.Worker.queue_reader","title":"queue_reader(output_path, read_interval=0.1)","text":"Reads frames from the queue and saves them as JPEG images.
Parameters:
Name Type Description Default output_path str The directory path where the frames will be saved.
required read_interval float The time interval between reading frames from the queue. Defaults to 0.1 seconds.
0.1 Source code in video_sampler/sampler.py def queue_reader(self, output_path, read_interval=0.1) -> None:\n \"\"\"\n Reads frames from the queue and saves them as JPEG images.\n\n Args:\n output_path (str): The directory path where the frames will be saved.\n read_interval (float, optional): The time interval between reading frames from the queue.\n Defaults to 0.1 seconds.\n \"\"\"\n while True:\n if not self.q.empty():\n frame_object: FrameObject\n for frame_object in self.q.get():\n if frame_object.metadata.get(\"end\", False):\n return\n if frame_object.frame is not None and (\n not self.devnull and isinstance(frame_object.frame, Image.Image)\n ):\n frame_object.frame.save(\n os.path.join(\n output_path,\n f\"{frame_object.metadata['frame_time']}.jpg\",\n )\n )\n time.sleep(read_interval)\n
"},{"location":"reference/video_sampler/schemas/","title":"Schemas","text":""},{"location":"reference/video_sampler/ttl_counter/","title":"Ttl counter","text":""},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter","title":"TTLCounter","text":"TTLCounter is a counter/list that expires items after a TTL period expires.
Source code in video_sampler/ttl_counter.py class TTLCounter:\n \"\"\"TTLCounter is a counter/list that expires items after a TTL period expires.\"\"\"\n\n def __init__(self, max_ttl: int) -> None:\n self.inner_counter = []\n self.max_ttl = max_ttl\n\n def __len__(self):\n \"\"\"Return the number of items in the counter.\"\"\"\n return len(self.inner_counter)\n\n def add_item(self, hash: str):\n \"\"\"Add an item with the max TTL.\"\"\"\n heapq.heappush(self.inner_counter, (self.max_ttl, hash))\n\n def tick(self):\n \"\"\"Decrease the TTL of all items by 1.\"\"\"\n for i, (ttl, hash) in enumerate(self.inner_counter):\n self.inner_counter[i] = (ttl - 1, hash)\n\n def expire_one(self):\n \"\"\"Expire the first item if its TTL is 0. Expires AT MOST one item.\"\"\"\n # peek the first item\n ttl, hash = self.inner_counter[0]\n if ttl <= 0:\n heapq.heappop(self.inner_counter)\n return hash\n return None\n\n def expire_all(self):\n \"\"\"Expire all items.\"\"\"\n for _, hash in self.inner_counter:\n yield hash\n self.inner_counter.clear()\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.__len__","title":"__len__()","text":"Return the number of items in the counter.
Source code in video_sampler/ttl_counter.py def __len__(self):\n \"\"\"Return the number of items in the counter.\"\"\"\n return len(self.inner_counter)\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.add_item","title":"add_item(hash)","text":"Add an item with the max TTL.
Source code in video_sampler/ttl_counter.py def add_item(self, hash: str):\n \"\"\"Add an item with the max TTL.\"\"\"\n heapq.heappush(self.inner_counter, (self.max_ttl, hash))\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.expire_all","title":"expire_all()","text":"Expire all items.
Source code in video_sampler/ttl_counter.py def expire_all(self):\n \"\"\"Expire all items.\"\"\"\n for _, hash in self.inner_counter:\n yield hash\n self.inner_counter.clear()\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.expire_one","title":"expire_one()","text":"Expire the first item if its TTL is 0. Expires AT MOST one item.
Source code in video_sampler/ttl_counter.py def expire_one(self):\n \"\"\"Expire the first item if its TTL is 0. Expires AT MOST one item.\"\"\"\n # peek the first item\n ttl, hash = self.inner_counter[0]\n if ttl <= 0:\n heapq.heappop(self.inner_counter)\n return hash\n return None\n
"},{"location":"reference/video_sampler/ttl_counter/#video_sampler.ttl_counter.TTLCounter.tick","title":"tick()","text":"Decrease the TTL of all items by 1.
Source code in video_sampler/ttl_counter.py def tick(self):\n \"\"\"Decrease the TTL of all items by 1.\"\"\"\n for i, (ttl, hash) in enumerate(self.inner_counter):\n self.inner_counter[i] = (ttl - 1, hash)\n
"},{"location":"reference/video_sampler/utils/","title":"Utils","text":""},{"location":"reference/video_sampler/utils/#video_sampler.utils.batched","title":"batched(iterable, n)","text":"Batch data into tuples of length n. The last batch may be shorter. from https://docs.python.org/3/library/itertools.html#itertools-recipes
Source code in video_sampler/utils.py def batched(iterable, n):\n \"\"\"\n Batch data into tuples of length n. The last batch may be shorter.\n from https://docs.python.org/3/library/itertools.html#itertools-recipes\n \"\"\"\n if n < 1:\n raise ValueError(\"n must be at least one\")\n it = iter(iterable)\n while batch := tuple(islice(it, n)):\n yield batch\n
"},{"location":"reference/video_sampler/utils/#video_sampler.utils.slugify","title":"slugify(value, allow_unicode=False)","text":"Taken from https://github.com/django/django/blob/master/django/utils/text.py Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated dashes to single dashes. Remove characters that aren't alphanumerics, underscores, or hyphens. Convert to lowercase. Also strip leading and trailing whitespace, dashes, and underscores.
Source code in video_sampler/utils.py def slugify(value, allow_unicode=False):\n \"\"\"\n Taken from https://github.com/django/django/blob/master/django/utils/text.py\n Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated\n dashes to single dashes. Remove characters that aren't alphanumerics,\n underscores, or hyphens. Convert to lowercase. Also strip leading and\n trailing whitespace, dashes, and underscores.\n \"\"\"\n value = str(value)\n if allow_unicode:\n value = unicodedata.normalize(\"NFKC\", value)\n else:\n value = (\n unicodedata.normalize(\"NFKD\", value)\n .encode(\"ascii\", \"ignore\")\n .decode(\"ascii\")\n )\n value = re.sub(r\"[^\\w\\s-]\", \"\", value.lower())\n return re.sub(r\"[-\\s]+\", \"-\", value).strip(\"-_\")\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/","title":"Integrations","text":""},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin","title":"YTDLPPlugin","text":"A plugin for yt-dlp to generate URLs and corresponding titles from the given URL.
Methods:
Name Description generate_urls Generates URLs and corresponding titles from the given URL.
Source code in video_sampler/integrations/yt_dlp_plugin.py class YTDLPPlugin:\n \"\"\"\n A plugin for yt-dlp to generate URLs and corresponding titles from the given URL.\n\n Methods:\n generate_urls(url, extra_yt_constr_args=None, extra_info_extract_opts=None) -> Iterable[str]:\n Generates URLs and corresponding titles from the given URL.\n\n \"\"\"\n\n def __init__(self, ie_key: str = \"Generic\"):\n \"\"\"\n Initialize the YTDLPPlugin instance.\n \"\"\"\n self.ie_key = ie_key\n self.ydl_opts = {\n \"format\": best_video_only,\n }\n\n def generate_urls(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n ) -> Iterable[str]:\n \"\"\"Generate URLs and download subtitles for a given video URL.\n\n Args:\n url (str): The URL of the video to download subtitles for.\n extra_info_extract_opts (dict, optional): Additional options for extracting video information.\n\n Yields:\n tuple: A tuple containing the video title, video format URL, and downloaded subtitles.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(params=(self.ydl_opts | extra_info_extract_opts)) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n if \"entries\" not in info:\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"]\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n yield entry[\"title\"], req_format[\"url\"]\n\n def get_subtitles_opts(self, no_download: bool = False) -> dict:\n return {\n \"postprocessors\": [\n {\n \"format\": \"srt\",\n \"key\": \"FFmpegSubtitlesConvertor\",\n \"when\": \"before_dl\",\n }\n ],\n \"format\": best_video_only,\n \"subtitleslangs\": [\"en.*\"],\n \"writeautomaticsub\": True,\n \"writesubtitles\": True,\n }\n\n def generate_urls_by_subs(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n ):\n \"\"\"Generate URLs and download subtitles for a given video URL.\n\n Args:\n url (str): The URL of the video to download subtitles for.\n extra_info_extract_opts (dict, optional): Additional options for extracting video information.\n\n Yields:\n tuple: A tuple containing the video title, video format URL, and downloaded subtitles.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(\n params=(self.ydl_opts | extra_info_extract_opts | self.get_subtitles_opts())\n ) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n import json\n\n json.dump(info, open(\"info.json\", \"w\"))\n if \"entries\" not in info:\n req_subs = list(info[\"requested_subtitles\"].values())[0]\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"], download_sub(req_subs[\"url\"])\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n req_subs = list(entry[\"requested_subtitles\"].values())[0]\n yield entry[\"title\"], req_format[\"url\"], download_sub(\n req_subs[\"url\"]\n )\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.__init__","title":"__init__(ie_key='Generic')","text":"Initialize the YTDLPPlugin instance.
Source code in video_sampler/integrations/yt_dlp_plugin.py def __init__(self, ie_key: str = \"Generic\"):\n \"\"\"\n Initialize the YTDLPPlugin instance.\n \"\"\"\n self.ie_key = ie_key\n self.ydl_opts = {\n \"format\": best_video_only,\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.generate_urls","title":"generate_urls(url, extra_info_extract_opts=None)","text":"Generate URLs and download subtitles for a given video URL.
Parameters:
Name Type Description Default url str The URL of the video to download subtitles for.
required extra_info_extract_opts dict Additional options for extracting video information.
None Yields:
Name Type Description tuple Iterable[str] A tuple containing the video title, video format URL, and downloaded subtitles.
Source code in video_sampler/integrations/yt_dlp_plugin.py def generate_urls(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n) -> Iterable[str]:\n \"\"\"Generate URLs and download subtitles for a given video URL.\n\n Args:\n url (str): The URL of the video to download subtitles for.\n extra_info_extract_opts (dict, optional): Additional options for extracting video information.\n\n Yields:\n tuple: A tuple containing the video title, video format URL, and downloaded subtitles.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(params=(self.ydl_opts | extra_info_extract_opts)) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n if \"entries\" not in info:\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"]\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n yield entry[\"title\"], req_format[\"url\"]\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.YTDLPPlugin.generate_urls_by_subs","title":"generate_urls_by_subs(url, extra_info_extract_opts=None)","text":"Generate URLs and download subtitles for a given video URL.
Parameters:
Name Type Description Default url str The URL of the video to download subtitles for.
required extra_info_extract_opts dict Additional options for extracting video information.
None Yields:
Name Type Description tuple A tuple containing the video title, video format URL, and downloaded subtitles.
Source code in video_sampler/integrations/yt_dlp_plugin.py def generate_urls_by_subs(\n self,\n url: str,\n extra_info_extract_opts: dict = None,\n):\n \"\"\"Generate URLs and download subtitles for a given video URL.\n\n Args:\n url (str): The URL of the video to download subtitles for.\n extra_info_extract_opts (dict, optional): Additional options for extracting video information.\n\n Yields:\n tuple: A tuple containing the video title, video format URL, and downloaded subtitles.\n \"\"\"\n if extra_info_extract_opts is None:\n extra_info_extract_opts = {}\n extr_args = {\"ie_key\": self.ie_key} if \"ytsearch\" not in url else {}\n with YoutubeDL(\n params=(self.ydl_opts | extra_info_extract_opts | self.get_subtitles_opts())\n ) as ydl:\n info = ydl.extract_info(url, download=False, **extr_args)\n import json\n\n json.dump(info, open(\"info.json\", \"w\"))\n if \"entries\" not in info:\n req_subs = list(info[\"requested_subtitles\"].values())[0]\n req_format = info[\"requested_formats\"][0]\n yield info[\"title\"], req_format[\"url\"], download_sub(req_subs[\"url\"])\n else:\n for entry in info.get(\"entries\", []):\n req_format = entry[\"requested_formats\"][0]\n req_subs = list(entry[\"requested_subtitles\"].values())[0]\n yield entry[\"title\"], req_format[\"url\"], download_sub(\n req_subs[\"url\"]\n )\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.best_video_best_audio","title":"best_video_best_audio(ctx)","text":"Taken from the yt-dlp documentation as-is
Source code in video_sampler/integrations/yt_dlp_plugin.py def best_video_best_audio(ctx):\n \"\"\"Taken from the yt-dlp documentation as-is\"\"\"\n \"\"\"Select the best video and the best audio that won't result in an mkv.\n NOTE: This is just an example and does not handle all cases\"\"\"\n\n # formats are already sorted worst to best\n formats = ctx.get(\"formats\")[::-1]\n\n # acodec='none' means there is no audio\n best_video = next(\n f for f in formats if f[\"vcodec\"] != \"none\" and f[\"acodec\"] == \"none\"\n )\n\n # find compatible audio extension\n audio_ext = {\"mp4\": \"m4a\", \"webm\": \"webm\"}[best_video[\"ext\"]]\n # vcodec='none' means there is no video\n best_audio = next(\n f\n for f in formats\n if (f[\"acodec\"] != \"none\" and f[\"vcodec\"] == \"none\" and f[\"ext\"] == audio_ext)\n )\n\n # These are the minimum required fields for a merged format\n yield {\n \"format_id\": f'{best_video[\"format_id\"]}+{best_audio[\"format_id\"]}',\n \"ext\": best_video[\"ext\"],\n \"requested_formats\": [best_video, best_audio],\n # Must be + separated list of protocols\n \"protocol\": f'{best_video[\"protocol\"]}+{best_audio[\"protocol\"]}',\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.best_video_only","title":"best_video_only(ctx)","text":"Just best video -- save bandwidth
Source code in video_sampler/integrations/yt_dlp_plugin.py def best_video_only(ctx):\n \"\"\"Just best video -- save bandwidth\"\"\"\n # formats are already sorted worst to best\n formats = ctx.get(\"formats\")[::-1]\n\n # acodec='none' means there is no audio\n best_video = next(f for f in formats if f[\"vcodec\"] != \"none\")\n # These are the minimum required fields for a merged format\n yield {\n \"format_id\": f'{best_video[\"format_id\"]}',\n \"ext\": best_video[\"ext\"],\n \"requested_formats\": [best_video],\n # Must be + separated list of protocols\n \"protocol\": f'{best_video[\"protocol\"]}',\n }\n
"},{"location":"reference/video_sampler/integrations/yt_dlp_plugin/#video_sampler.integrations.yt_dlp_plugin.no_shorts","title":"no_shorts(info, *, incomplete)","text":"Filter out short videos
Source code in video_sampler/integrations/yt_dlp_plugin.py def no_shorts(info, *, incomplete):\n \"\"\"Filter out short videos\"\"\"\n if url := info.get(\"url\", \"\"):\n if \"/shorts\" in url:\n return \"This is a short video\"\n
"},{"location":"reference/video_sampler/language/keyword_capture/","title":"Language","text":""},{"location":"reference/video_sampler/language/keyword_capture/#video_sampler.language.keyword_capture.download_sub","title":"download_sub(sub_url)","text":"Download a VTT subtitle file to a string.
Source code in video_sampler/language/keyword_capture.py def download_sub(sub_url: str):\n \"\"\"Download a VTT subtitle file to a string.\"\"\"\n response = requests.get(url=sub_url)\n return parse_srt_subtitle(response.text)\n
"},{"location":"reference/video_sampler/visualisation/clustering/","title":"Visualisation","text":""},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.build_feature_model","title":"build_feature_model(model_str)","text":"Build a feature extraction model.
Parameters:
Name Type Description Default model_str str Model name.
required Returns:
Name Type Description tuple Tuple of (model, extractor).
Source code in video_sampler/visualisation/clustering.py def build_feature_model(model_str: str):\n \"\"\"Build a feature extraction model.\n\n Args:\n model_str (str): Model name.\n\n Returns:\n tuple: Tuple of (model, extractor).\n \"\"\"\n extractor = AutoFeatureExtractor.from_pretrained(model_str)\n model = ResNetModel.from_pretrained(model_str)\n return model, extractor\n
"},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.cluster_features","title":"cluster_features(features, max_clusters=50)","text":"Cluster features using t-SNE and KMeans
Parameters:
Name Type Description Default features ndarray dict with keys \"embeds\" and \"paths\"
required max_clusters int maximum number of clusters
50 Retruns tuple: of (X, cluster_labels)
Source code in video_sampler/visualisation/clustering.py def cluster_features(\n features,\n max_clusters=50,\n):\n \"\"\"Cluster features using t-SNE and KMeans\n\n Args:\n features (np.ndarray): dict with keys \"embeds\" and \"paths\"\n max_clusters (int): maximum number of clusters\n\n Retruns:\n tuple: of (X, cluster_labels)\n \"\"\"\n proj = TSNE(n_components=2, perplexity=35, metric=\"cosine\")\n Xorg = np.asarray(features[\"embeds\"])\n X = proj.fit_transform(Xorg)\n\n # take about 10% of the frame as the number of clusters\n n_clusters = min(int(0.1 * len(features[\"embeds\"])), max_clusters)\n cluster_model = KMeans(n_clusters=n_clusters, random_state=0).fit(Xorg)\n return X, cluster_model.labels_\n
"},{"location":"reference/video_sampler/visualisation/clustering/#video_sampler.visualisation.clustering.extract_features","title":"extract_features(model_str, image_folder, mkey='pixel_values', batch_size=8)","text":"Extract features from a folder of images.
Parameters:
Name Type Description Default model_str str Model name.
required image_folder Path Folder with images.
required mkey str Key for the pixel values. Defaults to \"pixel_values\".
'pixel_values' batch_size int Batch size. Defaults to 8.
8 Returns:
Name Type Description dict Dictionary with keys \"embeds\" and \"paths\".
Source code in video_sampler/visualisation/clustering.py def extract_features(\n model_str: str, image_folder: Path, mkey=\"pixel_values\", batch_size: int = 8\n):\n \"\"\"Extract features from a folder of images.\n\n Args:\n model_str (str): Model name.\n image_folder (Path): Folder with images.\n mkey (str, optional): Key for the pixel values. Defaults to \"pixel_values\".\n batch_size (int, optional): Batch size. Defaults to 8.\n\n Returns:\n dict: Dictionary with keys \"embeds\" and \"paths\".\n \"\"\"\n\n out_features = defaultdict(list)\n model, extractor = build_feature_model(model_str)\n with torch.no_grad():\n all_files = list(image_folder.iterdir())\n for batch in tqdm(\n batched(all_files, batch_size), total=len(all_files) // batch_size\n ):\n # load images\n batch_imgs = [Image.open(img_path).convert(\"RGB\") for img_path in batch]\n # extract features\n batch_imgs = extractor(batch_imgs, return_tensors=\"pt\")[mkey]\n batch_features = model(batch_imgs).pooler_output.squeeze()\n if len(batch) == 1:\n batch_features = batch_features.expand(1, -1)\n batch_features = torch.functional.F.normalize(batch_features, p=2, dim=1)\n out_features[\"embeds\"].extend(batch_features)\n out_features[\"paths\"].extend([img_path.name for img_path in batch])\n return out_features\n
"}]}
\ No newline at end of file
diff --git a/sitemap.xml.gz b/sitemap.xml.gz
index 921c045..c64e4af 100644
Binary files a/sitemap.xml.gz and b/sitemap.xml.gz differ