PhysicalAssetCollection

class rapidtools.core.PhysicalAssetCollection(assets: InitVar[Iterable[PhysicalAsset]] = None)

Bases: object

A specialized container for collection of PhysicalAsset objects.

This class enforces unique asset IDs and provides fast, O(1), lookups. Internally, it uses a dictionary to store assets, but it behaves like a sequence for iteration and interaction.

Key Features:

• Spatial Querying: Filter assets by BoundingBox or Polygon.

• Data I/O: Import/Export to GeoJSON and convert to Pandas DataFrames.

• Batch Operations: Update attributes across all assets.

• Set Operations: Merge collections with collision handling.

Supported Container Operations:

• Iteration: for asset in collection: ...

• Access:

  • Key: collection['id_str'] (O(1) Fast)

  • Index: collection[0] (O(N) Slower)

  • Slice: collection[:5] (Returns a new PhysicalAssetCollection)

• Assignment: collection['id'] = asset

• Deletion: del collection['id']

• Length: len(collection)

• Membership: if 'pole_01' in collection: ...

Parameters:

assets (Iterable[PhysicalAsset]) – Assets to populate. Can be a list, tuple, or generator. Defaults to empty.

Methods

add	Add one or more assets to the collection.
collect_all_images	Aggregate images from all assets into a single unified collection.
filter_by_attribute	Return a new collection of assets with specified attribute criteria.
filter_by_geometry	Returns a new collection with assets matching a spatial query.
filter_empty	Return a new collection excluding assets with no descriptive attributes.
from_geojson	Creates a new PhysicalAssetCollection from a GeoJSON source.
from_shapefile	Creates a new PhysicalAssetCollection from an ESRI Shapefile.
get	Safely retrieve assets by unique ID(s).
merge	Merge another collection into this instance.
remove	Remove one or more assets from the collection by ID.
set_asset_type	Batch assign a specific asset type to all assets in the collection.
set_attribute	Batch update a specific attribute across all assets in the collection.
summary	Generate a statistical summary of the asset collection.
to_dataframe	Convert the collection to a Pandas DataFrame.
to_geojson	Serialize the collection into a GeoJSON FeatureCollection.
to_shapefile	Exports all assets in the collection to an ESRI Shapefile.

property combined_bounding_box: BoundingBox | None

Calculates the total bounding box that encloses all assets.

Returns:

BoundingBox – A BoundingBox object encompassing all assets, or None if the collection is empty.

Examples

If a PhysicalAssetCollection is empty, it’s combined_bounding_box will be None:

>>> from rapidtools.core import PhysicalAssetCollection
>>> collection = PhysicalAssetCollection()
>>> collection.combined_bounding_box is None
True

Combined bounds reflect the min/max extents across all assets:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset
>>>
>>> collection.add(PhysicalAsset(id="a1", geometry=Point(0, 0)))
>>> collection.add(PhysicalAsset(id="a2", geometry=Point(2, 3)))
>>> collection.add(PhysicalAsset(id="a3", geometry=Point(-1, 5)))
>>> bbox = collection.combined_bounding_box
>>> bbox.bounds
(-1.0, 0.0, 2.0, 5.0)

add(assets: PhysicalAsset | Iterable[PhysicalAsset]) → None

Add one or more assets to the collection.

This method accepts a single PhysicalAsset or an iterable containing multiple PhysicalAsset objects.

If an asset with the same ID already exists, or if an item is not a PhysicalAsset, a warning is logged and that specific item is skipped. It does not stop the addition of other valid assets.

Parameters:

assets (PhysicalAsset or Iterable[PhysicalAsset]) – A single PhysicalAsset or an iterable of PhysicalAsset objects.

Examples

Initialize a PhysicalAsset, a list of two PhysicalAsset objects and a PhysicalAssetCollection:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> asset1 = PhysicalAsset(id='pump_01', geometry=Point(10.0, 20.0))
>>> assets_list = [
...     PhysicalAsset(id='pump_02', geometry=Point(10.1, 20.1)),
...     PhysicalAsset(id='pump_03', geometry=Point(10.2, 20.2))
... ]
>>> collection = PhysicalAssetCollection()

Add a single asset to the PhysicalAssetCollection:

>>> collection.add(asset1)
>>> collection.summary()
{'total_assets': 1,
 'total_images': 0,
 'asset_types': {'Unknown': 1},
 'bounds': BoundingBox(minx=10.0, miny=20.0, maxx=10.0, maxy=20.0)}
Add multiple assets at once:

>>> collection.add(assets_list)
>>> len(collection)
3

Try to add a list of assets including duplicates and incorrect types. In the batch below, 'pump_01' is a duplicate, 'NotAnAsset' is the wrong type, so only 'pump_04' is added:

>>> mixed_batch = [
...     PhysicalAsset(id='pump_04', geometry=Point(10.3, 20.3)),
...     PhysicalAsset(id='pump_01', geometry=Point(10.0, 20.0)),
...     'NotAnAsset'
... ]
>>> collection.add(mixed_batch)
WARNING: Skipping duplicate asset with ID
'pump_01' (already exists).
WARNING: Skipping invalid item: Expected PhysicalAsset, got 'str'.
>>> len(collection)
4

collect_all_images() → ImageCollection

Aggregate images from all assets into a single unified collection.

This method iterates through every physical asset, extracts its associated images, and merges them into a new ImageCollection.

Returns:

ImageCollection – A new collection containing every image found across all assets in this PhysicalAssetCollection.

Examples

Setup a collection where 'pole_1' has 2 images, 'pole_2' has 0 images, and 'pole_3' has 1 image:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset,                 ...     PhysicalAssetCollection, ImageAsset
>>>
>>> a1 = PhysicalAsset(id='pole_1', geometry=Point(0, 0))
>>> a1.image_assets.add([
...     ImageAsset(id='img_A',path='',allow_missing_file=True),
...     ImageAsset(id='img_B',path='',allow_missing_file=True)
... ])
>>>
>>> a2 = PhysicalAsset(id='pole_2', geometry=Point(1, 1))
>>>
>>> a3 = PhysicalAsset(id='pole_3', geometry=Point(2, 2))
>>> a3.image_assets.add(
...     ImageAsset(id='img_C',
...     path='',
...     allow_missing_file=True
... ))
>>>
>>> collection = PhysicalAssetCollection([a1, a2, a3])

Run the aggregation. Note that ‘pole_2’ is skipped efficiently:

>>> all_images = collection.collect_all_images()
INFO: Collected 3 images from 2 assets.
>>> len(all_images)
3
>>> 'img_A' in all_images
True

classmethod from_geojson(source: str | Path | dict[str, Any], asset_type: str | None = None, overwrite_asset_type: bool = False) → PhysicalAssetCollection

Creates a new PhysicalAssetCollection from a GeoJSON source.

This factory method accepts a file path (string or Path object) pointing to a GeoJSON file, or a dictionary object representing valid GeoJSON data. It parses the ‘features’ list and initializes a new collection containing the assets found.

Mixed-Type Datasets:

This method automatically extracts all properties from the GeoJSON. If the file contains a mix of asset types under standard keys (like ‘asset_type’, ‘type’, or ‘category’), they are natively preserved and instantly accessible via the asset.asset_type property.

Parameters:

• source (str or Path or dict[str, Any]) –

  The GeoJSON data source. Can be one of:

  • A string path to a .geojson file.

  • A pathlib.Path object pointing to a file.

  • A Python dictionary structure representing a GeoJSON FeatureCollection.

• asset_type (str or None, optional) – A default asset type to apply to the imported assets (e.g., ‘building’). If the dataset already contains mixed types, leave this as None to preserve them, or use it alongside overwrite_asset_type=False to automatically fill in any unlabeled/blank assets. Defaults to None.

• overwrite_asset_type (bool, optional) – If True, overrides any existing type labels in the GeoJSON. If False, only applies the asset_type to assets that do not already have one (perfect for data with missing labels). Defaults to False.

Returns:

PhysicalAssetCollection – A new instance of the collection populated with assets parsed from the GeoJSON source.

Raises:

• TypeError – If source is not a string, Path, or dictionary.

• FileNotFoundError – If source is a path to a file that does not exist.

• ValueError – If the loaded data is not a valid GeoJSON ‘FeatureCollection’.

• JSONDecodeError – If the file exists but contains invalid JSON.

Examples

Import PhysicalAssetCollection:

>>> from rapidtools.core import PhysicalAssetCollection

Loading from a file path and assigning a default asset type:

>>> path = 'data/assets.geojson'
>>> collection = PhysicalAssetCollection.from_geojson(
...     path, asset_type='building'
... )
>>> print(len(collection))
5

Loading from a dictionary:

>>> data = {
...     "type": "FeatureCollection",
...     "features": [
...         {
...             "type": "Feature",
...             "properties": {"id": "A1", "name": "Pump"},
...             "geometry": {"type": "Point", "coordinates": [0, 0]}
...         }
...     ]
... }
>>> collection = PhysicalAssetCollection.from_geojson(
...     data, asset_type='pump'
... )
>>> asset = collection.get("A1")
>>> asset.summary()
--- PhysicalAsset Summary ---
ID:          A1
Asset Type:  pump
Geometry:    POINT (0 0)
Attributes (3):
{
    "id": "A1",
    "name": "Pump",
    "asset_type": "pump"
}
Image Assets (0):
  (No image assets)
-----------------------------

classmethod from_shapefile(file: str | Path, asset_type: str | None = None, overwrite_asset_type: bool = False) → PhysicalAssetCollection

Creates a new PhysicalAssetCollection from an ESRI Shapefile.

This factory method reads a shapefile, extracts the geometries and attribute tables, and populates a new collection. It attempts to reconstruct the original id if it was saved in the shapefile’s attribute table.

Mixed-Type Datasets:

This method automatically maps the Shapefile’s DBF attribute table to the asset properties. If the Shapefile contains mixed asset types under standard columns (like ‘asset_type’, ‘type’, or ‘category’), they are natively preserved.

Parameters:

• file (str or Path) – The path to the shapefile (.shp). The accompanying core files (.shx and .dbf) must be located in the same directory. It is also highly recommended that the projection file (.prj) and character encoding file (.cpg) are present to ensure spatial and text data are interpreted correctly.

• asset_type (str or None, optional) – A default asset type to apply to the imported assets (e.g., ‘road’). If the shapefile already contains mixed types, leave this as None to preserve them, or use it alongside overwrite_asset_type=False to automatically fill in any unlabeled/blank rows. Defaults to None.

• overwrite_asset_type (bool, optional) – If True, overrides any existing type labels in the shapefile’s attribute table. If False, only applies the asset_type to assets that don’t already have one. Defaults to False.

Returns:

PhysicalAssetCollection – A new instance populated with the assets from the shapefile.

Raises:

shapefile.ShapefileException – If the file cannot be read.

Examples

Import PhysicalAssetCollection class:

>>> from rapidtools.core import PhysicalAssetCollection

Load a shapefile from path, safely filling in missing asset types (ensuring accompanying .shx and .dbf files are in the same directory):

>>> path = 'data/assets.shp'
>>> collection = PhysicalAssetCollection.from_shapefile(
...     path, asset_type='building'
... )
INFO: Loaded 5 assets from Shapefile.
>>> print(len(collection))
5

Inspect an imported asset to view its reconstructed geometry and attributes:

>>> asset = collection.get('0')
>>> asset.summary()
--- PhysicalAsset Summary ---
ID:          0
Asset Type:  building
Geometry:    POLYGON ((-118.54801743326061 34.050774380694754,
                       -118.54799871557931 34.0507...
Attributes (1):
{
    "asset_type": "building"
}
Image Assets (0):
  (No image assets)
-----------------------------

get(asset_ids: str | Iterable[str], default: Any = None) → PhysicalAsset | Any | list[PhysicalAsset | Any]

Safely retrieve assets by unique ID(s).

This method is polymorphic:

1. If asset_ids is a str, it returns a single PhysicalAsset (or default).

2. If asset_ids is an Iterable (list, tuple, etc.), it returns a list of results in the same order as the requested IDs.

Parameters:

• asset_ids – A single asset ID string OR an iterable of ID strings.

• default – Value to return if an ID is not found. Defaults to None.

Returns:

• PhysicalAsset (or default) if input is a single string.

• list[PhysicalAsset or Any] if input is a list/iterable.

Raises:

TypeError – If asset_ids is not a str or Iterable.

Examples

Create a PhysicalAssetCollection consisting of two PhysicalAsset objects:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> collection = PhysicalAssetCollection()
>>> collection.add(PhysicalAsset(
>>>     id='pump_01',
>>>     geometry=Point(0,0)
>>> ))
>>> collection.add(PhysicalAsset(
>>>     id='pump_02',
>>>     geometry=Point(1,1)
>>> ))

Single lookup (returns PhysicalAsset or None):

>>> asset = collection.get('pump_01')
>>> asset.id
'pump_01'

Batch lookup (returns a list of PhysicalAsset objects):

>>> results = collection.get(['pump_01', 'pump_02', 'missing_id'])
>>> print(results)
[<PhysicalAsset id='pump_01' asset_type='N/A' geometry='Point'>,
<PhysicalAsset id='pump_02' asset_type='N/A' geometry='Point'>,
None]

# Set a custom default value:

>>> collection.get('missing_id', default='Not Found')
'Not Found'

filter_by_attribute(key: str, value: Any, operator: str = '==') → PhysicalAssetCollection

Return a new collection of assets with specified attribute criteria.

This method filters the collection based on the key existing in an asset’s attributes dictionary and comparing it against value using the specified operator.

Parameters:

• key (str) – The dictionary key to look up in asset.attributes.

• value (Any) – The value to compare against.

• operator (str, optional) –

  The comparison operator to use. Supported options:

  • '==': Exact match (default).

  • '!=': Not equal.

  • '>', '>=', '<', '<=': Numeric/Lexical comparison.

  • 'in': Checks if asset attribute is in the provided value (list/set).

  • 'contains': Checks if asset attribute CONTAINS the

    provided value.

  • 'exists': Checks if the key exists (value argument is ignored).

Returns:

PhysicalAssetCollection – A new collection containing only the matching assets.

Raises:

ValueError – If the provided operator string is not supported.

Examples

Create a collection consisting of three physical assets:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> collection = PhysicalAssetCollection()
>>> collection.add([
...     PhysicalAsset(
...         id='a1',
...         geometry=Point(0,0),
...         attributes={'status': 'active', 'age': 5}
...     ),
...     PhysicalAsset(
...         id='a2',
...         geometry=Point(1,0),
...         attributes={'status': 'inactive', 'age': 10}
...     ),
...     PhysicalAsset(
...         id='a3',
...         geometry=Point(0,1),
...         attributes={'status': 'active', 'age': 15}
...     )
... ])

Filter for exact match (default behavior). Get assets that have an active status (i.e., status=active):

>>> active = collection.filter_by_attribute('status', 'active')
>>> len(active)
2

Filter using inequalities. Get assets older than 8:

>>> old = collection.filter_by_attribute('age', 8, operator='>')
>>> len(old)
2

Filter using 'in' to find assets matching ANY value in a list. In this example, we keep the asset if status is 'active' or 'repair'. If no assets are 'repair', it simply returns the 'active' ones:

>>> subset = collection.filter_by_attribute(
...     'status',
...     ['active', 'repair'],
...     operator='in'
... )
>>> subset.summary()
{'total_assets': 2,
 'total_images': 0,
 'asset_types': {'Unknown': 2},
 'bounds': BoundingBox(minx=0.0, miny=0.0, maxx=0.0, maxy=1.0)}
>>> subset[0].attributes
{'status': 'active', 'age': 5}

filter_by_geometry(geometry: BoundingBox | PolygonRegion | BaseGeometry, predicate: str = 'intersects') → PhysicalAssetCollection

Returns a new collection with assets matching a spatial query.

This method accepts BoundingBox, PolygonRegion, or raw Shapely geometries. It leverages the underlying geometry for efficient filtering.

Parameters:

• geometry –

  The search area can be:

  • An instance of BoundingBox or PolygonRegion.

  • A raw Shapely BaseGeometry.

• predicate (str) –

  The spatial relationship to check. Defaults to 'intersects'.

  • 'intersects': Returns assets that touch or overlap (Default).

  • 'within': Returns assets strictly inside the geometry.

  • 'contains': Returns assets that strictly contain the geometry.

Returns:

PhysicalAssetCollection – A subset of matching assets.

Examples

Create an asset collection consisting of five physical assets:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset,                 ... PhysicalAssetCollection, BoundingBox, PolygonRegion
>>>
>>> assets = [
...     PhysicalAsset(
...         id='pole_1',
...         geometry=Point(1, 1),
...         attributes={'status': 'active', 'voltage': 110},
...     ),
...     PhysicalAsset(
...         id='pole_2',
...         geometry=Point(2, 2),
...         attributes={'status': 'repair', 'voltage': 220},
...     ),
...     PhysicalAsset(
...         id='pole_3',
...         geometry=Point(5, 5),
...         attributes={'status': 'inactive', 'voltage': 110},
...     ),
...     PhysicalAsset(
...         id='pump_1',
...         geometry=Point(8, 8),
...         attributes={'status': 'active', 'voltage': 'N/A'},
...     ),
...     PhysicalAsset(
...         id='pump_2',
...         geometry=Point(12, 12),
...         attributes={'status': 'active', 'voltage': 220},
...     ),
... ]
>>> collection = PhysicalAssetCollection(assets)

Filter using BoundingBox (0,0 to 6,6):

>>> bbox = BoundingBox(0, 0, 6, 6)
>>> in_box = collection.filter_by_geometry(bbox)
>>> print(
... f'Assets in BoundingBox(0,0,6,6): {[a.id for a in in_box]}'
... )
Assets in BoundingBox(0,0,6,6): ['pole_1', 'pole_2', 'pole_3']

Filter using a triangular PolygonRegion:

>>> triangle = PolygonRegion([(0,0), (0,3), (3,0)])
>>> in_tri = collection.filter_by_geometry(triangle)
>>> print(f'Assets in triangle: {[a.id for a in in_tri]}')
Assets in triangle: ['pole_1']

A comparison of a strict 'within' check vs ‘intersects'. pole_3 is at (5,5). Following BoundingBox barely touches pole_3. Please note that in checking for a spatial relationship, intersects (i.e., the default predicate) includes edges, while within excludes them:

>>> touching_box = BoundingBox(5, 5, 10, 10)
>>> touching = collection.filter_by_geometry(touching_box)
>>> print(f'Touching bounding box: {[a.id for a in touching]}')
Touching bounding box: ['pole_3', 'pump_1']
>>>
>>> inside = collection.filter_by_geometry(
...    touching_box,
...    predicate='within'
... )
>>> print(f'Strictly inside: {[a.id for a in inside]}')
Strictly inside: ['pump_1']

filter_empty() → PhysicalAssetCollection

Return a new collection excluding assets with no descriptive attributes.

This method acts as a data-cleaning tool. It identifies assets where the attributes dictionary is completely empty and excludes them from the resulting collection.

Returns:

PhysicalAssetCollection – A new collection containing only assets that possess at least one metadata attribute.

Example

>>> # If you have 500 assets but only 360 were successfully analyzed:
>>> print(len(collection))
500
>>> clean_collection = collection.filter_empty()
>>> print(len(clean_collection))
360

merge(other: PhysicalAssetCollection, strategy: Literal['skip', 'overwrite', 'raise'] = 'skip') → None

Merge another collection into this instance.

Merges another PhysicalAssetCollection into this instance.

This method modifies the current collection in-place by adding assets from the other collection. If an asset ID already exists in the current collection, the conflict is resolved according to the strategy parameter.

Parameters:

• other (PhysicalAssetCollection) – The collection to merge into this instance.

• strategy (str, optional) –

  The rule for handling duplicate asset IDs. Defaults to 'skip'. Available options are:

  • 'skip': Keep the existing asset in this collection; ignore the incoming asset.

  • 'overwrite': Replace the existing asset in this collection with the incoming asset.

  • 'raise': Abort the merge and raise a ValueError if a duplicate ID is found.

Returns:

None – This method modifies the collection in place.

Raises:

• TypeError – If other is not an instance of PhysicalAssetCollection.

• ValueError – If strategy is not one of 'skip', 'overwrite', or 'raise'.

• ValueError – If strategy is 'raise' and a duplicate asset ID is encountered.

Examples

Create two collections, one containing a single asset, and another containing two assets (one of which shares the same ID as the asset in the first collection):

>>> # Assume we have a collection with one asset
>>> main_col = PhysicalAssetCollection()
>>> main_col.add(PhysicalAsset(id='1', name='Original Server'))
>>>
>>> # Assume we have a second collection to merge in
>>> new_col = PhysicalAssetCollection()
>>> new_col.add(PhysicalAsset(id='1', name='New Server')) # Duplicate ID
>>> new_col.add(PhysicalAsset(id='2', name='Switch'))     # New ID

Merge using the 'skip' strategy (default behavior). The existing 'Original Server' is preserved. The 'Switch' is added:

>>> main_col.merge(new_col, strategy='skip')
>>> print(main_col.get('1').name)
Original Server
>>> print(main_col.get('2').name)
Switch

Merge using the 'overwrite' strategy. The existing asset is updated to 'New Server'. The 'Switch' is added:

>>> main_col.merge(new_col, strategy='overwrite')
>>> print(main_col.get('1').name)
New Server

Merge using the 'raise' strategy. The operation fails immediately upon finding the duplicate '1'.

>>> try:
...     main_col.merge(new_col, strategy='raise')
... except ValueError as e:
...     print(e)
Duplicate asset ID '1' found during merge.

remove(asset_ids: str | Iterable[str]) → None

Remove one or more assets from the collection by ID.

This method is polymorphic:

1. If asset_ids is a str, it removes that single asset.

2. If asset_ids is an Iterable, it iterates and removes all IDs found.

If an ID is not found in the collection, the operation is skipped for that specific ID and a warning is logged.

Parameters:

asset_ids – A single asset ID string OR an iterable of ID strings.

Raises:

TypeError – If asset_ids is not a str or an Iterable.

Examples

Create a PhysicalAssetCollection consisting of two PhysicalAsset objects:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> collection = PhysicalAssetCollection()
>>> collection.add([
...     PhysicalAsset(id='p1', geometry=Point(0,0)),
...     PhysicalAsset(id='p2', geometry=Point(1,1))
... ])

Remove the PhysicalAsset with ID:'p1' from the collection:

>>> collection.remove('p1')
>>> print(collection)
PhysicalAssetCollection(_data={'p2': <PhysicalAsset id='p2'
asset_type='N/A' geometry='Point'>})

Try to remove a non-matching item:

>>> collection.remove('p99')
WARNING: Asset ID 'p99' not found. Skipping removal.

Batch remove multiple assets:

>>> collection.remove(['p2', 'p99'])
WARNING: Asset ID 'p99' not found. Skipping removal.
>>> len(collection)
0

set_attribute(key: str, value: Any, overwrite: bool = False) → None

Batch update a specific attribute across all assets in the collection.

This method assigns value to asset.attributes[key]. It supports both static values and dynamic value generation using functions.

Parameters:

• key (str) – The attribute name to set.

• value (Any) –

  The value to assign.

  • If a static value (int, str), it is assigned directly.

  • If a callable (function/lambda), it is executed for each asset as value(asset), and the result is assigned. This is useful for derived properties or avoiding shared mutable references (see examples).

• overwrite (bool) – If False, existing attributes will be preserved and not updated. If True, the new value overwrites any existing value. Defaults to False.

Examples

Setup a collection with two assets for the examples:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> assets = [
...     PhysicalAsset(
...     id='a1',
...     geometry=Point(0,0),
...     attributes={'status': 'old'}
... ),
...     PhysicalAsset(id='a2', geometry=Point(1,1)) # No attributes
... ]
>>> collection = PhysicalAssetCollection(assets)

Add a new tag to every asset (static assigment):

>>> collection.set_attribute('category', 'utility_pole')
INFO: Updated attribute 'category' for 2 assets.
>>> collection['a1'].attributes['category']
'utility_pole'

Overwrite existing attribute values. In the example below, 'a1' already has status='old'. By default, set_attribute does not modify an existing value. To force an update, set the overwrite argument to True:

>>> collection.set_attribute('status', 'new')
INFO: Updated attribute 'status' for 1 asset.
>>> collection['a1'].attributes['status']
'old'
>>> collection.set_attribute('status', 'new', overwrite=True)
INFO: Updated attribute 'status' for 2 assets.
>>> collection['a1'].attributes['status']
'new'

Derive values for each asset (dynamic assignment). This example uses a lambda function to get the x coordinates of assets using their geometry information:

>>> collection.set_attribute(
...     'x_coord',
...     value=lambda asset: asset.geometry.x,
...     overwrite=True
... )
INFO: Updated attribute 'x_coord' for 2 assets.
>>> collection['a2'].attributes['x_coord']
1.0

How to initialize mutable objects safely (and not fall for the list Trap). This example uses a lambda that returns a new list generate a unique list for each asset, avoiding the common bug where all assets share the same list reference:

>>> collection.set_attribute('logs', lambda _: list())
INFO: Updated attribute 'logs' for 2 assets.
>>> collection['a1'].attributes['logs'].append('Log Entry 1')
>>> collection['a2'].attributes['logs']
[]

set_asset_type(asset_type: str, overwrite: bool = True) → None

Batch assign a specific asset type to all assets in the collection.

This is a convenience wrapper around set_attribute that ensures the type is saved using the standardized primary key (‘asset_type’).

Parameters:

• asset_type (str) – The type label to assign (e.g., ‘building’, ‘utility_pole’).

• overwrite (bool) – If True, overwrites any existing asset types. If False, only assigns the type to assets that do not currently have one. Defaults to True.

Examples

Create a collection of assets:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> collection = PhysicalAssetCollection([
...     PhysicalAsset(id='1', geometry=Point(0,0)),
...     PhysicalAsset(id='2', geometry=Point(1,1))
... ])

Assign the ‘building’ type to all assets in the collection:

>>> collection.set_asset_type('building')
>>> print(collection['1'].asset_type)
building
>>> print(collection['2'].asset_type)
building

summary() → dict[str, Any]

Generate a statistical summary of the asset collection.

This method aggregates data across all assets to provide a high-level overview. It determines an asset’s type by checking the priority keys defined in PhysicalAsset._ASSET_TYPE_KEYS against the asset’s attributes.

Returns:

dict[str, Any] –

A dictionary containing:

• 'total_assets' (int): Total number of physical assets.

• 'total_images' (int): Sum of images across all assets.

• 'asset_types' (dict): A frequency count of assets by type.

• 'bounds' (tuple): The (minx, miny, maxx, maxy) bounds of the entire collection.

Examples

Create a collection with a diverse set of assets. Please note that assets 1 and 2 are defined using the standard 'type' key. The type for asset 3 is define using 'category' key (a fallback key in _ASSET_TYPE_KEYS). Lastly, asset 4 has no known type keys:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> col = PhysicalAssetCollection()
>>> col.add(PhysicalAsset(
...     id='A1',
...     geometry=Point(0,0),
...     attributes={'type': 'Pole'}
... ))
>>> col.add(PhysicalAsset(
...     id='A2',
...     geometry=Point(1,1),
...     attributes={'type': 'Pole'}
... ))
>>> col.add(PhysicalAsset(
...     id='A3',
...     geometry=Point(2,2),
...     attributes={'category': 'Valve'}
... ))
>>> col.add(PhysicalAsset(
...     id='A4',
...     geometry=Point(3,3),
...     attributes={'material': 'wood'}
... ))

Create a summary:

>>> stats = col.summary()
>>> print(stats['total_assets'])
4
>>> print(stats['asset_types'])
{'Pole': 2, 'Valve': 1, 'Unknown': 1}

to_dataframe() → DataFrame

Convert the collection to a Pandas DataFrame.

Returns a tabular representation of the assets. The resulting DataFrame includes standard columns ('id', 'asset_type', 'geometry_wkt', 'image_count') and expands the 'attributes' dictionary so that each attribute key becomes its own column.

Returns:

pd.DataFrame – A DataFrame where each row is a PhysicalAsset.

Example

Create a sample collection:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> col = PhysicalAssetCollection()
>>> col.add(
...     PhysicalAsset(
...         id='A1',
...         geometry=Point(0,0),
...         attributes={'material': 'wood'}
... ))

Convert to DataFrame:

>>> df = col.to_dataframe()
>>> print(df.iloc[0]['material'])
wood

to_geojson(file: str | Path | None = None, indent: int | None = 2, ignore_properties: list[str] | None = None) → dict[str, Any]

Serialize the collection into a GeoJSON FeatureCollection.

This method converts all assets in the collection into a standard GeoJSON structure. It can either return the dictionary directly or write it to a file if a path is provided.

It supports an optional filter to exclude specific metadata fields (such as the heavy ‘image_assets’ list) from the output to reduce file size.

Parameters:

• file (str or Path or None, optional) – If provided, the GeoJSON data will be written to this file path. Defaults to None.

• indent (int | None, optional) – The indentation level for JSON pretty-printing. Set to None for a compact representation. Defaults to 2.

• ignore_properties (list[str] or None, optional) – A list of property keys to exclude from each feature in the output. This is useful for removing large nested objects like ‘image_assets’ when only metadata is needed. Defaults to None.

Returns:

dict[str, Any] – A dictionary representing the GeoJSON FeatureCollection.

Examples

Write the collection to a GeoJSON dictionary:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> col = PhysicalAssetCollection()
>>> col.add(PhysicalAsset(id='A1', geometry=Point(0,0)))
>>> data = col.to_geojson()
>>> print(data['type'])
FeatureCollection
>>> print(data['features'][0]['id'])
A1

Write the collection to a GeoJSON file name 'output.json' with custom indentation:

>>> _ = col.to_geojson(file='output.json', indent=4)

Export only metadata by ignoring the ‘image_assets’ property:

>>> data = col.to_geojson(ignore_properties=['image_assets'])
>>> print('image_assets' in data['features'][0]['properties'])
False

to_shapefile(file: str | Path, crs: str = 'EPSG:4326', ignore_properties: list[str] | None = None) → None

Exports all assets in the collection to an ESRI Shapefile.

Because a shapefile requires a fixed tabular schema, this method performs a first pass over all assets to unify the attributes. If an attribute exists in some assets but not others, the missing values will be written as empty/null.

It supports an optional filter to exclude specific metadata fields (such as the heavy ‘image_assets’ list) from the output to reduce file size and prevent cluttering the DBF attribute table.

Note on Shapefile limitations:

• Column names (attribute keys) are strictly truncated to 10 chars.

• String values are strictly truncated to 254 characters.

• Nested data structures (lists, dicts) are serialized to JSON strings.

Because of these limits, data that does not fit the DBF table is written to a JSON sidecar next to the shapefile (<name>.rapidtools.json): the image assets of every asset, keyed by asset ID, and the mapping from truncated column names back to the original attribute keys. The DBF itself only records the number of images per asset (n_images). from_shapefile() reads the sidecar when it is present, so the round trip preserves image assets and attribute names.

Parameters:

• file (str or Path) – The destination path for the shapefile. The string should end in ‘.shp’. The method will generate the accompanying .shx, .dbf, and .prj files alongside it.

• crs (str, optional) – The Coordinate Reference System (CRS). Currently, this method auto-generates a .prj file for ‘EPSG:4326’ (WGS84). Defaults to ‘EPSG:4326’.

• ignore_properties (list[str] or None, optional) – A list of property keys to exclude from the attribute table. Defaults to None.

Raises:

• ImportError – If the ‘pyshp’ library is not installed.

• ValueError – If the collection is empty.

Examples

Create a collection:

>>> from shapely.geometry import Point
>>> from rapidtools.core import PhysicalAsset, PhysicalAssetCollection
>>>
>>> col = PhysicalAssetCollection()
>>> col.add(PhysicalAsset(
...     id='A1',
...     geometry=Point(0,0),
...     attributes={'status': 'active'}
... ))
>>> col.to_shapefile('data/assets.shp')

Write the collection to a Shapefile, exporting only basic metadata (ignoring the ‘image_assets’ and ‘notes’ properties):

>>> col.to_shapefile(
...     'data/assets_light.shp',
...     ignore_properties=['image_assets', 'notes']
... )