ImageCollection

class rapidtools.core.ImageCollection(assets: list[ImageAsset] | None = None)

Bases: object

A container class for managing a group of ImageAsset objects.

This class provides utilities for filtering, batch downloading, and exporting metadata for a collection of images.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(id='a', path='a.jpg', allow_missing_file=True),
...         ImageAsset(id='b', path='b.jpg', allow_missing_file=True),
...     ]
... )
>>> len(images), 'a' in images, images.get_ids()
(2, True, ['a', 'b'])

Methods

add	Add one or more assets to the collection.
download_all	Download all images in the collection using multi-threading.
filter	Filter the collection using a custom function.
filter_by_bbox	Filter images that fall within a geographic bounding box.
filter_by_property	Filter assets where a specific property matches a given value.
filter_downloaded	Filter for assets that currently exist on disk.
get_ids	Retrieve a list of IDs for all assets in the collection.
merge	Merge with another ImageCollection.
remove	Remove one or more assets from the collection.
subset	Create a new ImageCollection containing assets at the given indices.
to_dataframe	Convert the collection metadata to a Pandas DataFrame.
to_json	Export the collection metadata to a JSON file.

add(items: ImageAsset | list[ImageAsset], overwrite: bool = False) → None

Add one or more assets to the collection.

This method accepts either a single ImageAsset object or a list of objects. It checks for duplicates based on the asset ID. Assets whose id is None are always appended.

Parameters:

• items (ImageAsset | list[ImageAsset]) – A single ImageAsset object or a list of ImageAsset objects to add.

• overwrite (bool) – If True, existing assets with the same ID will be replaced by the new ones. If False, duplicates are ignored (the existing asset is kept) and an INFO message is logged. Defaults to False.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection()
>>> images.add(ImageAsset(id='a', path='a.jpg', allow_missing_file=True))
>>> images.add(
...     [
...         ImageAsset(id='a', path='a_v2.jpg', allow_missing_file=True),
...         ImageAsset(id='b', path='b.jpg', allow_missing_file=True),
...     ]
... )
INFO: Skipping duplicate asset with ID: a
>>> images.get_ids()
['a', 'b']
>>> images.add(
...     ImageAsset(id='a', path='a_v2.jpg', allow_missing_file=True),
...     overwrite=True,
... )
>>> images[0].filename
'a_v2.jpg'

merge(other: ImageCollection, overwrite_path: bool = False, overwrite_properties: bool = True, add_new: bool = False) → None

Merge with another ImageCollection.

This method iterates through the incoming collection and either merges properties into existing assets (if IDs match) or adds new assets (if add_new is True).

Parameters:

• other (ImageCollection) – The collection to merge with.

• overwrite_path (bool) – If True, existing assets will update their file paths to match the incoming assets. Defaults to False.

• overwrite_properties (bool) – If True, incoming properties (metadata) overwrite existing ones on match. Defaults to True.

• add_new (bool) – If True, assets in other that do not share an ID with any asset in this collection will be appended. If False, unmatched assets are ignored. Defaults to False.

Examples

>>> master_images.merge(new_images, add_new=True)
INFO: Merge complete: 5 merged, 2 added, 0 skipped.

filter(func: Callable[[ImageAsset], bool]) → ImageCollection

Filter the collection using a custom function.

Parameters:

func (Callable[[ImageAsset], bool]) – A callable that takes an ImageAsset as input and returns True if the asset should be kept, or False otherwise.

Returns:

ImageCollection – A new instance containing only the assets for which the function returned True. The original is left untouched.

Example

>>> subset = collection.filter(
...     lambda x: x.properties.get('damage') == 'High'
... )

filter_by_property(key: str, value: Any) → ImageCollection

Filter assets where a specific property matches a given value.

Parameters:

• key (str) – The property key to check (e.g., 'event_name').

• value (Any) – The value the property must equal.

Returns:

ImageCollection – A new collection containing the matching assets.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(
...             id='a',
...             path='a.jpg',
...             properties={'event': 'eaton'},
...             allow_missing_file=True,
...         ),
...         ImageAsset(
...             id='b',
...             path='b.jpg',
...             properties={'event': 'palisades'},
...             allow_missing_file=True,
...         ),
...     ]
... )
>>> images.filter_by_property('event', 'eaton').get_ids()
['a']

filter_downloaded() → ImageCollection

Filter for assets that currently exist on disk.

Returns:

ImageCollection – A new collection containing only assets whose is_downloaded property is True.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> pending = ImageAsset(
...     id='p', path='missing.jpg', allow_missing_file=True
... )
>>> len(ImageCollection([pending]).filter_downloaded())
0

filter_by_bbox(bbox: BoundingBox) → ImageCollection

Filter images that fall within a geographic bounding box.

This method filters the collection to include only images whose latitude / longitude properties fall within the provided BoundingBox. Images without coordinates are excluded.

Parameters:

bbox (BoundingBox) – The BoundingBox object defining the area.

Returns:

ImageCollection – A new collection containing only the assets within the bounds.

Example

>>> from rapidtools.core import BoundingBox, ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(
...             id='inside',
...             path='a.jpg',
...             properties={'latitude': 34.15, 'longitude': -118.15},
...             allow_missing_file=True,
...         ),
...         ImageAsset(
...             id='outside',
...             path='b.jpg',
...             properties={'latitude': 40.0, 'longitude': -74.0},
...             allow_missing_file=True,
...         ),
...         ImageAsset(
...             id='no_coords', path='c.jpg', allow_missing_file=True
...         ),
...     ]
... )
>>> bbox = BoundingBox(-118.2, 34.1, -118.1, 34.2)
>>> images.filter_by_bbox(bbox).get_ids()
['inside']

get_ids(ignore_none: bool = True) → list[str | int | None]

Retrieve a list of IDs for all assets in the collection.

Parameters:

ignore_none (bool) – If True, assets with no ID (None) are excluded from the list. Defaults to True.

Returns:

list[str | int | None] – The IDs in collection order.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(id='a', path='a.jpg', allow_missing_file=True),
...         ImageAsset(id='b', path='b.jpg', allow_missing_file=True),
...     ]
... )
>>> images.get_ids()
['a', 'b']

remove(items: ImageAsset | str | list[ImageAsset | str]) → None

Remove one or more assets from the collection.

This method accepts a single ImageAsset, a single ID string, or a list containing a mix of both.

Parameters:

items – An ImageAsset object, a string ID, or a list of these to remove.

Examples

Initialize an ImageCollection:

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> # Initialize a collection with some assets
>>> img1 = ImageAsset(id='img_001', path='path/to/1.jpg',
... allow_missing_file=True)
>>> img2 = ImageAsset(id='img_002', path='path/to/2.jpg',
... allow_missing_file=True)
>>> img3 = ImageAsset(id='img_003', path='path/to/3.jpg',
... allow_missing_file=True)
>>> collection = ImageCollection([img1, img2, img3])

Remove by ID: >>> collection.remove(‘img_001’) Removed 1 assets from the collection. >>> len(collection) 2

Remove by Object: >>> collection.remove(img2) Removed 1 assets from the collection. >>> len(collection) 1

Remove a list of mixed types: >>> collection.add([img1, img2]) >>> collection.remove([‘img_001’, img2]) Removed 2 assets from the collection. >>> len(collection) 1

subset(indices: list[int]) → ImageCollection

Create a new ImageCollection containing assets at the given indices.

Parameters:

indices (list[int]) – A list of integer indices (e.g., [0, 5, 10]).

Returns:

ImageCollection – A new instance holding the selected assets.

Raises:

IndexError – If any index is out of bounds.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(id='a', path='a.jpg', allow_missing_file=True),
...         ImageAsset(id='b', path='b.jpg', allow_missing_file=True),
...         ImageAsset(id='c', path='c.jpg', allow_missing_file=True),
...     ]
... )
>>> images.subset([0, 2]).get_ids()
['a', 'c']

download_all(max_workers: int = 5, overwrite: bool = False) → None

Download all images in the collection using multi-threading.

Each asset’s ImageAsset.download() is called with a shared requests.Session. Failures are logged per asset and do not stop the remaining downloads. Progress is logged every ten images.

Parameters:

• max_workers (int) – The maximum number of parallel download threads. Defaults to 5.

• overwrite (bool) – If True, existing files will be re-downloaded and overwritten. Defaults to False.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(
...             id='a',
...             path='downloads/a.jpg',
...             properties={'url': 'https://example.com/a.jpg'},
...             allow_missing_file=True,
...         )
...     ]
... )
>>> images.download_all(max_workers=2)
INFO: Starting batch download for 1 images with 2 workers.

to_dataframe() → Any

Convert the collection metadata to a Pandas DataFrame.

Returns:

pd.DataFrame – A DataFrame where each row represents an asset. Columns include id, path, is_downloaded, and all keys found in the asset properties.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [
...         ImageAsset(
...             id='a',
...             path='a.jpg',
...             properties={'event': 'eaton'},
...             allow_missing_file=True,
...         )
...     ]
... )
>>> df = images.to_dataframe()
>>> list(df.columns)
['id', 'path', 'is_downloaded', 'event']

to_json(filepath: str | Path) → None

Export the collection metadata to a JSON file.

The output is a list of {'id', 'path', 'properties'} objects. Values that are not natively JSON-serializable are written via str().

Parameters:

filepath (str | Path) – The path where the JSON file will be saved.

Example

>>> from rapidtools.core import ImageAsset, ImageCollection
>>>
>>> images = ImageCollection(
...     [ImageAsset(id='a', path='a.jpg', allow_missing_file=True)]
... )
>>> images.to_json('images.json')