From 51dc258b1c5a9d0168bfda7659827fad92d577b0 Mon Sep 17 00:00:00 2001 From: Daithi Date: Tue, 16 Dec 2025 18:20:39 -0700 Subject: [PATCH 1/3] commit before generating files --- docs/tutorials/point_cloud_example.md | 104 ++++++++ fastfuels_sdk/inventories.py | 35 +++ fastfuels_sdk/pointclouds.py | 368 ++++++++++++++++++++++++++ tests/data/threedep_domain.geojson | 9 + tests/test_inventories.py | 85 +++++- tests/test_pointclouds.py | 278 +++++++++++++++++++ tests/utils.py | 20 +- 7 files changed, 897 insertions(+), 2 deletions(-) create mode 100644 docs/tutorials/point_cloud_example.md create mode 100644 fastfuels_sdk/pointclouds.py create mode 100644 tests/data/threedep_domain.geojson create mode 100644 tests/test_pointclouds.py diff --git a/docs/tutorials/point_cloud_example.md b/docs/tutorials/point_cloud_example.md new file mode 100644 index 0000000..4e849ff --- /dev/null +++ b/docs/tutorials/point_cloud_example.md @@ -0,0 +1,104 @@ + +# Tutorial: Incorporate ALS Point Cloud Data into a FastFuels Domain +This tutorial demonstrates how to incorporate airborne laser scanning (ALS / LiDAR) point cloud data into an existing FastFuels domain. It utilizes a point cloud–derived canopy structure to inform the tree inventory and canopy fuel grids, enabling higher‐fidelity fuels representation. + +## Prerequisites + +Before starting this tutorial, make sure you have: +- FastFuels SDK installed (`pip install fastfuels-sdk`) +- A valid FastFuels API key +- Basic familiarity with Python and GeoPandas + +## What You'll Learn + +By the end of this tutorial, you'll know how to: +- Create an ALS point cloud within a FastFuels domain +- Generate a tree inventory informed by the ALS data + +## Step 1: Create a FastFuels Domain + +For the polygon created by the coordinates: +```python +import geopandas as gpd +from shapely.geometry import Polygon + +# Define the polygon coordinates for Lubrecht area +coordinates = [ + [-113.43635167116199,46.89738742250387], + [-113.44842074255764,46.8894785976307], + [-113.44763362920567,46.88740657162339], + [-113.44993666456837,46.8858524995899], + [-113.44923700825561,46.88429838253265], + [-113.44273603501621,46.88389988372731], + [-113.43538964373204,46.882365635689325], + [-113.42005550954369,46.88523484307359], + [-113.42329141999039,46.88919967571519], + [-113.42361209580038,46.892566564773205], + [-113.4263524163587,46.89527586047467], + [-113.42696461563226,46.895973083547176], + [-113.42853884233625,46.89916027357725], + [-113.42711037736427,46.90210825569156], + [-113.42798494775504,46.90336309078498], + [-113.42932595568779,46.903004569469715], + [-113.43241610440292,46.90099282206219], + [-113.43445676864847,46.8980248588519], + [-113.43635167116199,46.89738742250387] +] + +# Create a GeoDataFrame +polygon = Polygon(coordinates) +roi = gpd.GeoDataFrame( + geometry=[polygon], + crs="EPSG:4326" # WGS 84 coordinate system +) + +from fastfuels_sdk.domains import Domain + +domain = Domain.from_geodataframe( + geodataframe=roi, + name="Blue Mountain ROI", + description="Test area in Lubrecht Experimental Forest", + horizontal_resolution=2.0, # 2-meter horizontal resolution + vertical_resolution=1.0 # 1-meter vertical resolution +) + +print(f"Created domain with ID: {domain.id}") +``` + +## Step 2: Create an ALS point cloud within FastFuels + +Next, use 3DEP data to generate a feature point cloud: + +```python +# 1. +from fastfuels_sdk.pointclouds import PointClouds + +# 2. Correct Method Name & 3. Added Required 'sources' argument +als_pointcloud = ( + PointClouds.from_domain_id(domain.id) + .create_als_point_cloud(sources=["3DEP"]) +) + +# This works because create_als_point_cloud returns the child object +als_pointcloud.wait_until_completed() + +``` + +## Step 3: Create Tree Inventory using the point cloud data + +Create a tree inventory and generate the canopy fuel grid: + +```python +from fastfuels_sdk.inventories import Inventories +from fastfuels_sdk.grids import TreeGridBuilder + +# Create tree inventory +# Note: Ensure Step 2 (Point Cloud creation) is fully completed before running this +tree_inventory = Inventories.from_domain_id( + domain.id +).create_tree_inventory_from_point_cloud() +tree_inventory.wait_until_completed() +``` + +## Next Steps +For additional guidance and complementary workflows, refer back to the “Create and Export QUIC-Fire Inputs with FastFuels SDK” tutorial. \ No newline at end of file diff --git a/fastfuels_sdk/inventories.py b/fastfuels_sdk/inventories.py index fb78d22..979517d 100644 --- a/fastfuels_sdk/inventories.py +++ b/fastfuels_sdk/inventories.py @@ -672,6 +672,41 @@ def create_tree_inventory_from_file_upload( return tree_inventory + def create_tree_inventory_from_point_cloud( + self, + in_place: bool = False, + ) -> TreeInventory: + """Create a tree inventory derived from the domain's point cloud data. + This method initiates the creation of a tree inventory by processing the + Airborne Laser Scanning (ALS) point cloud associated with this domain. + Individual trees are segmented from the point cloud to derive tree locations + and heights. + Prerequisites: + - The domain must have an existing ALS point cloud resource (see PointClouds). + - The point cloud status must be 'completed'. + Parameters + ---------- + in_place : bool, optional + If True, updates this Inventories object's tree inventory (self.tree). + If False (default), returns a new inventory object without modifying this instance. + Returns + ------- + TreeInventory + The newly created tree inventory object. Initially in 'pending' status. + Examples + -------- + >>> from fastfuels_sdk import Inventories + >>> inventories = Inventories.from_domain_id("abc123") + >>> + >>> # Create tree inventory from existing point cloud + >>> inventory = inventories.create_tree_inventory_from_point_cloud() + >>> inventory.wait_until_completed() + """ + return self.create_tree_inventory( + sources=[TreeInventorySource.POINTCLOUD], + in_place=in_place, + ) + class TreeInventory(TreeInventoryModel): """ diff --git a/fastfuels_sdk/pointclouds.py b/fastfuels_sdk/pointclouds.py new file mode 100644 index 0000000..343a044 --- /dev/null +++ b/fastfuels_sdk/pointclouds.py @@ -0,0 +1,368 @@ +""" +fastfuels_sdk/pointclouds.py +""" + +# Core imports +from __future__ import annotations +from time import sleep +from typing import Optional, List + +# Internal imports +from fastfuels_sdk.api import ( + get_point_cloud_api, + get_als_point_cloud_api, +) +from fastfuels_sdk.client_library import AlsPointCloudSource +from fastfuels_sdk.utils import format_processing_error +from fastfuels_sdk.client_library.models import ( + PointCloud as PointCloudModel, + AlsPointCloud as AlsPointCloudModel, + CreateAlsPointCloudRequest, +) + + +class PointClouds(PointCloudModel): + """Point cloud resources associated with a domain. + + Attributes + ---------- + domain_id : str + ID of the domain these point clouds belong to + als : AlsPointCloud, optional + Airborne Laser Scanning (ALS) data + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + + >>> # Access ALS data + >>> if pc.als: + ... als_data = pc.als + + >>> # Get updated point cloud data + >>> pc = pc.get() + + See Also + -------- + Domain : Container for point clouds + AlsPointCloud : Airborne Laser Scanning structure + """ + + domain_id: str + als: Optional[AlsPointCloud] + + @classmethod + def from_domain_id(cls, domain_id: str) -> PointClouds: + """Retrieve the point clouds associated with a domain. + + Parameters + ---------- + domain_id : str + The ID of the domain to retrieve point clouds for + + Returns + ------- + PointClouds + A PointClouds object containing: + - als : AlsPointCloud, optional + ALS data within the domain + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + + >>> # Check for specific point clouds. These will be None until created. + >>> if pc.als: + ... print("Domain has ALS point cloud data") + + See Also + -------- + PointClouds.get : Refresh point cloud data + """ + response = get_point_cloud_api().get_pointclouds(domain_id=domain_id) + response_data = _convert_api_models_to_sdk_classes( + domain_id, response.model_dump(), response + ) + + return cls(domain_id=domain_id, **response_data) + + def get(self, in_place: bool = False) -> PointClouds: + """Get the latest point cloud data for this domain. + + Parameters + ---------- + in_place : bool, optional + If True, updates this PointClouds instance with new data. + If False (default), returns a new PointClouds instance. + + Returns + ------- + PointClouds + The updated PointClouds object + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> # Get fresh data in a new instance + >>> updated_pc = pc.get() + >>> + >>> # Or update the existing instance + >>> pc.get(in_place=True) + """ + response = get_point_cloud_api().get_pointclouds(domain_id=self.domain_id) + response_data = response.model_dump() + response_data = _convert_api_models_to_sdk_classes( + self.domain_id, response_data, response + ) + + if in_place: + # Update all attributes of current instance + for key, value in response_data.items(): + setattr(self, key, value) + return self + + return PointClouds(domain_id=self.domain_id, **response_data) + + def create_als_point_cloud( + self, + sources: str | List[str] | AlsPointCloudSource | List[AlsPointCloudSource], + in_place: bool = False, + ) -> AlsPointCloud: + """Create an ALS point cloud for this domain using specified data sources. + + Parameters + ---------- + sources : list[str] + Data sources to use. Supports: + - "file": Generates a signed URL for user upload + - "3DEP": Automatically retrieves public data (e.g. USGS 3DEP) + in_place : bool, optional + If True, updates this PointClouds instance with new data. + If False (default), leaves this instance unchanged. + + Returns + ------- + AlsPointCloud + The newly created ALS point cloud object. If source is "file", check + the `.file` attribute for upload URL and instructions. + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> + >>> # Create from public data + >>> als = pc.create_als_point_cloud(["3DEP"]) + >>> + >>> # Update PointClouds instance + >>> pc.create_als_point_cloud(["3DEP"], in_place=True) + >>> pc.als.status + "pending" + + See Also + -------- + AlsPointCloud.wait_until_completed : Wait for background processing + """ + # Create API request + request = CreateAlsPointCloudRequest(sources=sources) + + # Call API + response = get_als_point_cloud_api().create_als_point_cloud( + domain_id=self.domain_id, create_als_point_cloud_request=request + ) + + # Convert API model to SDK class + als_point_cloud = AlsPointCloud( + domain_id=self.domain_id, **response.model_dump() + ) + + if in_place: + self.als = als_point_cloud + + return als_point_cloud + + +class AlsPointCloud(AlsPointCloudModel): + """Airborne Laser Scanning (ALS) point cloud features. + + Represents high-resolution elevation data used for creating detailed canopy + height models and tree inventories. + + Attributes + ---------- + domain_id : str + ID of the domain this point cloud belongs to + sources : list[str] + Data sources used to create these features ("file", "3DEP") + status : str, optional + Current processing status ("pending", "running", "completed") + created_on : datetime, optional + When these features were created + modified_on : datetime, optional + When these features were last modified + checksum : str, optional + Unique identifier for this version of the features + file : dict, optional + Contains upload instructions (URL, headers) if source is "file" + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> + >>> # Access existing ALS data + >>> if pc.als: + ... als = pc.als + ... print(als.status) + + >>> # Create new ALS data + >>> als = pc.create_als_point_cloud(["3DEP"]) + """ + + domain_id: str + + def get(self, in_place: bool = False) -> AlsPointCloud: + """Get the latest ALS point cloud data. + + Parameters + ---------- + in_place : bool, optional + If True, updates this AlsPointCloud instance with new data. + If False (default), returns a new AlsPointCloud instance. + + Returns + ------- + AlsPointCloud + The updated ALS point cloud object + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> als = pc.create_als_point_cloud(["3DEP"]) + >>> # Get fresh data in a new instance + >>> updated_als = als.get() + >>> + >>> # Or update the existing instance + >>> als.get(in_place=True) + """ + response = get_als_point_cloud_api().get_als_point_cloud( + domain_id=self.domain_id + ) + if in_place: + # Update all attributes of current instance + for key, value in response.model_dump().items(): + setattr(self, key, value) + return self + return AlsPointCloud(domain_id=self.domain_id, **response.model_dump()) + + def wait_until_completed( + self, + step: float = 5, + timeout: float = 600, + in_place: bool = True, + verbose: bool = False, + ) -> AlsPointCloud: + """Wait for ALS point cloud processing to complete. + + ALS point clouds are processed asynchronously and may take several seconds + to minutes to complete depending on domain size and data sources. + + Parameters + ---------- + step : float, optional + Seconds between status checks (default: 5) + timeout : float, optional + Maximum seconds to wait (default: 600) + in_place : bool, optional + If True (default), updates this instance with completed data. + If False, returns a new instance. + verbose : bool, optional + If True, prints status updates (default: False) + + Returns + ------- + AlsPointCloud + The completed ALS point cloud object + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> als = pc.create_als_point_cloud(["3DEP"]) + >>> als.wait_until_completed(verbose=True) + ALS point cloud has status `pending` (5.00s) + ALS point cloud has status `completed` (10.00s) + """ + elapsed_time = 0 + als_point_cloud = self.get(in_place=in_place if in_place else False) + while als_point_cloud.status != "completed": + if als_point_cloud.status == "failed": + error_msg = "ALS point cloud processing failed." + + # Extract detailed error information if available + error_obj = getattr(als_point_cloud, "error", None) + if error_obj: + error_details = format_processing_error(error_obj) + if error_details: + error_msg = f"{error_msg}\n\n{error_details}" + + raise RuntimeError(error_msg) + if elapsed_time >= timeout: + raise TimeoutError("Timed out waiting for ALS point cloud to finish.") + sleep(step) + elapsed_time += step + als_point_cloud = self.get(in_place=in_place if in_place else False) + if verbose: + print( + f"ALS point cloud has status `{als_point_cloud.status}` ({elapsed_time:.2f}s)" + ) + return als_point_cloud + + def delete(self) -> None: + """Delete this ALS point cloud. + + Permanently removes ALS data from the domain. This also cancels + any ongoing processing jobs and removes files from storage. + + Returns + ------- + None + + Examples + -------- + >>> from fastfuels_sdk import PointClouds + >>> pc = PointClouds.from_domain_id("abc123") + >>> als = pc.create_als_point_cloud(["3DEP"]) + >>> # Remove ALS data when no longer needed + >>> als.delete() + >>> # Subsequent operations will raise NotFoundException + >>> als.get() # raises NotFoundException + """ + get_als_point_cloud_api().delete_als_point_cloud(domain_id=self.domain_id) + return None + + +def _convert_api_models_to_sdk_classes( + domain_id, response_data: dict, response_obj=None +) -> dict: + """Convert API models to SDK classes with domain_id. + + Parameters + ---------- + domain_id : str + Domain ID to attach to point clouds + response_data : dict + Dict representation of the response + response_obj : optional + Original response object (to preserve complex nested models) + """ + if "als" in response_data and response_data["als"]: + response_data["als"] = AlsPointCloud( + domain_id=domain_id, **response_data["als"] + ) + + return response_data \ No newline at end of file diff --git a/tests/data/threedep_domain.geojson b/tests/data/threedep_domain.geojson new file mode 100644 index 0000000..b44fe2a --- /dev/null +++ b/tests/data/threedep_domain.geojson @@ -0,0 +1,9 @@ +{"type":"FeatureCollection","features":[{"type":"Feature","properties":{},"geometry":{"coordinates":[ + [ + [-114.19991252294375, 46.7747518267752], + [-114.19991252294375, 46.77023218168779], + [-114.19133479742844, 46.77023218168779], + [-114.19133479742844, 46.7747518267752], + [-114.19991252294375, 46.7747518267752] + ] + ],"type":"Polygon"},"id":0}]} \ No newline at end of file diff --git a/tests/test_inventories.py b/tests/test_inventories.py index 6d08bab..9622b14 100644 --- a/tests/test_inventories.py +++ b/tests/test_inventories.py @@ -8,8 +8,13 @@ from pathlib import Path # Internal imports -from tests.utils import create_default_domain, normalize_datetime +from tests.utils import ( + create_default_domain, + normalize_datetime, + create_default_domain_threedep, +) from fastfuels_sdk.inventories import Inventories, TreeInventory +from fastfuels_sdk.pointclouds import PointClouds from fastfuels_sdk.client_library.exceptions import NotFoundException, ApiException from fastfuels_sdk.client_library import ( TreeInventorySource, @@ -34,6 +39,40 @@ def test_domain(): domain.delete() +@pytest.fixture(scope="module") +def test_point_cloud_domain(): + """Fixture that creates a test domain to be used by the tests""" + domain = create_default_domain_threedep() + + # Return the domain for use in tests + yield domain + + # Cleanup: Delete the domain after the tests + try: + domain.delete() + except NotFoundException: + pass + + +@pytest.fixture(scope="module") +def test_point_cloud_resource(test_point_cloud_domain): + """Fixture that ensures an ALS point cloud exists for the test domain""" + # Initialize the PointClouds interface + pc = PointClouds.from_domain_id(test_point_cloud_domain.id) + + # Create the ALS point cloud using 3DEP (public data) + # We use 3DEP because it triggers a job processing workflow similar to production + als = pc.create_als_point_cloud(sources=["3DEP"]) + + # Wait for it to complete so tests don't hit 404s + als.wait_until_completed() + + yield als + + # Cleanup (Optional, as the domain deletion in test_point_cloud_domain + # will cascade delete this resource anyway) + + @pytest.fixture(scope="module") def test_inventories(test_domain): """Fixture that creates a test inventories object to be used by the tests""" @@ -826,3 +865,47 @@ def test_delete_existing_inventory(self, test_inventories, test_tree_inventory): test_inventories.get(in_place=True) assert test_inventories.tree is None + + +class TestCreateTreeInventoryFromPointCloud: + def test_create_from_point_cloud( + self, test_point_cloud_domain, test_point_cloud_resource + ): + """Test creating a tree inventory from an existing point cloud""" + # 1. Initialize Inventories interface + # Note: We pass test_point_cloud_resource to ensure the prerequisite exists, + # but we use the domain ID to initialize the Inventories object. + inventories = Inventories.from_domain_id(test_point_cloud_domain.id) + + # 2. Call the new method + tree_inventory = inventories.create_tree_inventory_from_point_cloud() + + # 3. Validation + assert tree_inventory is not None + assert isinstance(tree_inventory, TreeInventory) + assert tree_inventory.domain_id == test_point_cloud_domain.id + # Status should start as pending/running + assert tree_inventory.status in ["pending", "running", "completed"] + + # 4. Verify Source + actual_sources = [ + src.actual_instance.value if hasattr(src, "actual_instance") else src + for src in tree_inventory.sources + ] + assert "pointcloud" in actual_sources + + def test_in_place_true(self, test_point_cloud_domain, test_point_cloud_resource): + """Test creation with in_place=True""" + inventories = Inventories.from_domain_id(test_point_cloud_domain.id) + original_tree = inventories.tree + + # Create with in_place=True + tree_inventory = inventories.create_tree_inventory_from_point_cloud( + in_place=True + ) + + # Verify update + assert inventories.tree is not None + assert inventories.tree is tree_inventory + assert inventories.tree is not original_tree + assert tree_inventory.domain_id == test_point_cloud_domain.id \ No newline at end of file diff --git a/tests/test_pointclouds.py b/tests/test_pointclouds.py new file mode 100644 index 0000000..501b26a --- /dev/null +++ b/tests/test_pointclouds.py @@ -0,0 +1,278 @@ +""" +test_pointclouds.py +""" + +# Core imports +from uuid import uuid4 +from datetime import datetime + +# Internal imports +from tests.utils import create_default_domain_threedep, normalize_datetime +from fastfuels_sdk.pointclouds import PointClouds, AlsPointCloud +from fastfuels_sdk.client_library.exceptions import NotFoundException + +# External imports +import pytest + + +@pytest.fixture(scope="module") +def test_domain(): + """Fixture that creates a test domain to be used by the tests""" + domain = create_default_domain_threedep() + + # Return the domain for use in tests + yield domain + + # Cleanup: Delete the domain after the tests + try: + domain.delete() + except NotFoundException: + pass + + +@pytest.fixture(scope="module") +def test_pointclouds(test_domain): + """Fixture that creates a test pointclouds object to be used by the tests""" + pc = PointClouds.from_domain_id(test_domain.id) + # The pointclouds resource exists implicitly or is empty by default + yield pc + + +class TestPointCloudsFromDomain: + def test_success(self, test_domain): + """Test successful retrieval of pointclouds from a domain""" + pc = PointClouds.from_domain_id(test_domain.id) + + # Verify basic structure and domain relationship + assert pc is not None + assert isinstance(pc, PointClouds) + assert pc.domain_id == test_domain.id + + # ALS should be None initially if not created + # (Assuming the API returns empty/null for als initially) + # If the API returns an empty object, adjust this assertion. + if pc.als: + assert isinstance(pc.als, AlsPointCloud) + + def test_bad_domain_id(self, test_domain): + """Test error handling for a non-existent domain ID""" + bad_test_domain = test_domain.model_copy(deep=True) + bad_test_domain.id = uuid4().hex + + # Depending on API implementation, this might raise NotFound immediately + # or return a valid object that fails on subsequent calls. + # Assuming strictly mirrored behavior from Features: + with pytest.raises(NotFoundException): + PointClouds.from_domain_id(bad_test_domain.id) + + +class TestGetPointClouds: + def test_get_update_default(self, test_pointclouds): + """Test get() returns new instance with updated data by default""" + original_pc = test_pointclouds + updated_pc = original_pc.get() + + # Verify new instance returned + assert updated_pc is not original_pc + assert isinstance(updated_pc, PointClouds) + + # Verify data matches + assert updated_pc.domain_id == original_pc.domain_id + if original_pc.als: + assert updated_pc.als is not None + assert updated_pc.als.checksum == original_pc.als.checksum + + def test_get_update_in_place(self, test_pointclouds): + """Test get(in_place=True) updates existing instance""" + pc = test_pointclouds + result = pc.get(in_place=True) + + # Verify same instance returned + assert result is pc + + # Verify instance was updated/refreshed + assert isinstance(result, PointClouds) + assert result.domain_id == test_pointclouds.domain_id + + def test_get_nonexistent_domain(self, test_pointclouds): + """Test error handling when domain no longer exists""" + # Store original domain_id + original_id = test_pointclouds.domain_id + + # Set invalid domain_id + test_pointclouds.domain_id = uuid4().hex + + # Verify exception raised + with pytest.raises(NotFoundException): + test_pointclouds.get() + + # Restore original domain_id + test_pointclouds.domain_id = original_id + + +class TestCreateAlsPointCloud: + @staticmethod + def assert_data_validity(als, domain_id): + """Validates the basic ALS point cloud data""" + assert als is not None + assert isinstance(als, AlsPointCloud) + assert als.domain_id == domain_id + assert als.status in ["pending", "running", "completed", "failed"] + # Checksum is typically set upon creation + assert isinstance(als.checksum, str) and len(als.checksum) > 0 + + def test_create_with_3dep_source(self, test_pointclouds): + """Tests ALS creation with '3DEP' source (public data)""" + # Use '3DEP' as it likely triggers background job + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + self.assert_data_validity(als, test_pointclouds.domain_id) + assert "3DEP" in als.sources + + # Original parent object should rely on subsequent refresh or be independent + # unless in_place was used. + assert test_pointclouds.als is not als + + def test_create_with_file_source(self, test_pointclouds): + """Tests ALS creation with 'file' source (upload instructions)""" + als = test_pointclouds.create_als_point_cloud(sources=["file"]) + + self.assert_data_validity(als, test_pointclouds.domain_id) + # Check if "file" is in the sources list (this is a list of strings/enums, so 'in' works here) + assert "file" in als.sources + + # Verify file upload instructions exist using Dot Notation + assert als.file is not None + assert als.file.url is not None + assert als.file.headers is not None + # Optional: Verify specific header content if needed + assert "x-goog-content-length-range" in als.file.headers + + def test_create_in_place_true(self, test_pointclouds): + """Test ALS creation with in_place=True""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"], in_place=True) + + self.assert_data_validity(als, test_pointclouds.domain_id) + # Parent object should be updated + assert test_pointclouds.als is als + assert test_pointclouds.als.status in ["pending", "running", "completed"] + + def test_create_nonexistent_domain(self, test_pointclouds): + """Test error handling for non-existent domain""" + bad_pc = test_pointclouds.model_copy(deep=True) + bad_pc.domain_id = uuid4().hex + + with pytest.raises(NotFoundException): + bad_pc.create_als_point_cloud(sources=["3DEP"]) + + +class TestGetAlsPointCloud: + def test_default(self, test_pointclouds): + """Tests fetching ALS data returns new instance by default""" + # First create an ALS resource + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + # Get latest data + updated = als.get() + + # Verify new instance returned + assert updated is not None + assert isinstance(updated, AlsPointCloud) + assert updated is not als + assert normalize_datetime(updated) == normalize_datetime(als) + + def test_in_place(self, test_pointclouds): + """Tests fetching ALS data in place""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + result = als.get(in_place=True) + + # Verify same instance returned and updated + assert result is als + assert isinstance(result, AlsPointCloud) + + def test_get_nonexistent_domain(self, test_pointclouds): + """Test error handling when domain no longer exists""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + # Store original domain_id + original_id = als.domain_id + + # Set invalid domain_id + als.domain_id = uuid4().hex + + # Verify exception raised + with pytest.raises(NotFoundException): + als.get() + + # Restore original domain_id + als.domain_id = original_id + + +class TestWaitUntilCompletedAlsPointCloud: + def test_wait_until_completed_default(self, test_pointclouds): + """Test waiting for ALS to complete with default settings""" + # Ensure we use 3DEP so it actually processes (file source waits for upload) + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + # Depending on test environment speed, this might finish instantly or take time + completed = als.wait_until_completed(step=1, timeout=60) + + assert completed is als # In-place by default + assert completed.status == "completed" + assert isinstance(completed.created_on, datetime) + + def test_wait_until_completed_not_in_place(self, test_pointclouds): + """Test waiting without in-place updates""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + completed = als.wait_until_completed(in_place=False) + + assert completed is not als # New instance + assert completed.status == "completed" + # Original might still be pending or completed depending on timing, + # but the key check is that 'completed' is a different object. + + def test_wait_until_completed_timeout(self, test_pointclouds): + """Test timeout handling""" + # Create a new job + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + # If the job is instant in test env, skip check, otherwise verify timeout + if als.status != "completed": + with pytest.raises(TimeoutError): + als.wait_until_completed(timeout=0.001) # Extremely short timeout + + def test_wait_until_completed_nonexistent_domain(self, test_pointclouds): + """Test error handling when domain no longer exists""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + als.domain_id = uuid4().hex # Set invalid ID + + with pytest.raises(NotFoundException): + als.wait_until_completed() + + +class TestDeleteAlsPointCloud: + def test_delete_existing_als(self, test_pointclouds): + """Tests deleting an existing ALS point cloud""" + # Create ALS + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + + # Delete it + als.delete() + + # Verify it was deleted via direct get + # Note: The API might return 404 for the sub-resource or return null + # Assuming standard behavior where GET /als returns 404 if deleted + with pytest.raises(NotFoundException): + als.get() + + # Parent PointClouds should reflect deletion after refresh + test_pointclouds.get(in_place=True) + assert test_pointclouds.als is None + + def test_delete_nonexistent_als(self, test_pointclouds): + """Test deleting an ALS resource that doesn't exist""" + als = test_pointclouds.create_als_point_cloud(sources=["3DEP"]) + als.domain_id = uuid4().hex # Set invalid ID + + with pytest.raises(NotFoundException): + als.delete() \ No newline at end of file diff --git a/tests/utils.py b/tests/utils.py index 9e59c5c..00c48bd 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -24,10 +24,28 @@ def create_default_domain() -> Domain: return domain +def create_default_domain_threedep() -> Domain: + """Creates a Domain resource with where 3dep data is available""" + # Load test GeoJSON data + with open(TEST_DATA_DIR / "threedep_domain.geojson") as f: + geojson = json.load(f) + + # Create a domain using the GeoJSON + domain = Domain.from_geojson( + geojson, + name="test_threedep_domain", + description="Domain for testing threedep operations", + horizontal_resolution=1.0, + vertical_resolution=1.0, + ) + + return domain + + def normalize_datetime(resource): """Normalize datetime fields by ensuring consistent timezone handling""" if resource.created_on and resource.created_on.tzinfo: resource.created_on = resource.created_on.replace(tzinfo=None) if resource.modified_on and resource.modified_on.tzinfo: resource.modified_on = resource.modified_on.replace(tzinfo=None) - return resource + return resource \ No newline at end of file From d24b0cf334b122f07f30685e938d761c79f943d3 Mon Sep 17 00:00:00 2001 From: Daithi Date: Tue, 16 Dec 2025 18:29:28 -0700 Subject: [PATCH 2/3] finished changes on init.py and api.py --- fastfuels_sdk/__init__.py | 3 +++ fastfuels_sdk/api.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/fastfuels_sdk/__init__.py b/fastfuels_sdk/__init__.py index 1d46520..7ee25c6 100644 --- a/fastfuels_sdk/__init__.py +++ b/fastfuels_sdk/__init__.py @@ -13,6 +13,7 @@ ) from fastfuels_sdk.exports import Export from fastfuels_sdk.convenience import export_roi, export_roi_to_quicfire +from fastfuels_sdk.pointclouds import PointClouds, AlsPointCloud __all__ = [ @@ -34,4 +35,6 @@ "Export", "export_roi", "export_roi_to_quicfire", + "PointClouds", + "AlsPointCloud", ] diff --git a/fastfuels_sdk/api.py b/fastfuels_sdk/api.py index d3d4eff..65fb456 100644 --- a/fastfuels_sdk/api.py +++ b/fastfuels_sdk/api.py @@ -18,6 +18,8 @@ SurfaceGridApi, TopographyGridApi, FeatureGridApi, + PointCloudApi, + AlsPointCloudApi, ) _client: Optional[ApiClient] = None @@ -32,6 +34,8 @@ _surface_grid_api: Optional[SurfaceGridApi] = None _topography_grid_api: Optional[TopographyGridApi] = None _feature_grid_api: Optional[FeatureGridApi] = None +_point_cloud_api: Optional[PointCloudApi] = None +_als_point_cloud_api: Optional[AlsPointCloudApi] = None def set_api_key(api_key: str) -> None: @@ -46,6 +50,7 @@ def set_api_key(api_key: str) -> None: global _client, _domains_api, _inventories_api, _tree_inventory_api global _features_api, _road_feature_api, _water_feature_api global _grids_api, _tree_grid_api, _surface_grid_api, _topography_grid_api, _feature_grid_api + global _point_cloud_api, _als_point_cloud_api _client = None _domains_api = None @@ -59,6 +64,8 @@ def set_api_key(api_key: str) -> None: _surface_grid_api = None _topography_grid_api = None _feature_grid_api = None + _point_cloud_api = None + _als_point_cloud_api = None os.environ["FASTFUELS_API_KEY"] = api_key @@ -244,3 +251,24 @@ def get_feature_grid_api() -> FeatureGridApi: if _feature_grid_api is None: _feature_grid_api = FeatureGridApi(ensure_client()) return _feature_grid_api + +def get_point_cloud_api() -> PointCloudApi: + """Get the cached PointCloudApi instance, creating it if necessary. + Returns: + The PointCloudApi instance + """ + global _point_cloud_api + if _point_cloud_api is None: + _point_cloud_api = PointCloudApi(ensure_client()) + return _point_cloud_api + + +def get_als_point_cloud_api() -> AlsPointCloudApi: + """Get the cached AlsPointCloudApi instance, creating it if necessary. + Returns: + The AlsPointCloudApi instance + """ + global _als_point_cloud_api + if _als_point_cloud_api is None: + _als_point_cloud_api = AlsPointCloudApi(ensure_client()) + return _als_point_cloud_api \ No newline at end of file From a4face3fe1ff619dbb10267a6e177842d341e43f Mon Sep 17 00:00:00 2001 From: Daithi Date: Tue, 16 Dec 2025 18:30:00 -0700 Subject: [PATCH 3/3] pre commit formatting --- docs/tutorials/point_cloud_example.md | 2 +- fastfuels_sdk/api.py | 3 ++- fastfuels_sdk/inventories.py | 4 ++-- fastfuels_sdk/pointclouds.py | 2 +- tests/data/threedep_domain.geojson | 2 +- tests/test_inventories.py | 2 +- tests/test_pointclouds.py | 2 +- tests/utils.py | 2 +- 8 files changed, 10 insertions(+), 9 deletions(-) diff --git a/docs/tutorials/point_cloud_example.md b/docs/tutorials/point_cloud_example.md index 4e849ff..6423682 100644 --- a/docs/tutorials/point_cloud_example.md +++ b/docs/tutorials/point_cloud_example.md @@ -101,4 +101,4 @@ tree_inventory.wait_until_completed() ``` ## Next Steps -For additional guidance and complementary workflows, refer back to the “Create and Export QUIC-Fire Inputs with FastFuels SDK” tutorial. \ No newline at end of file +For additional guidance and complementary workflows, refer back to the “Create and Export QUIC-Fire Inputs with FastFuels SDK” tutorial. diff --git a/fastfuels_sdk/api.py b/fastfuels_sdk/api.py index 65fb456..0609049 100644 --- a/fastfuels_sdk/api.py +++ b/fastfuels_sdk/api.py @@ -252,6 +252,7 @@ def get_feature_grid_api() -> FeatureGridApi: _feature_grid_api = FeatureGridApi(ensure_client()) return _feature_grid_api + def get_point_cloud_api() -> PointCloudApi: """Get the cached PointCloudApi instance, creating it if necessary. Returns: @@ -271,4 +272,4 @@ def get_als_point_cloud_api() -> AlsPointCloudApi: global _als_point_cloud_api if _als_point_cloud_api is None: _als_point_cloud_api = AlsPointCloudApi(ensure_client()) - return _als_point_cloud_api \ No newline at end of file + return _als_point_cloud_api diff --git a/fastfuels_sdk/inventories.py b/fastfuels_sdk/inventories.py index 979517d..d5743c2 100644 --- a/fastfuels_sdk/inventories.py +++ b/fastfuels_sdk/inventories.py @@ -673,8 +673,8 @@ def create_tree_inventory_from_file_upload( return tree_inventory def create_tree_inventory_from_point_cloud( - self, - in_place: bool = False, + self, + in_place: bool = False, ) -> TreeInventory: """Create a tree inventory derived from the domain's point cloud data. This method initiates the creation of a tree inventory by processing the diff --git a/fastfuels_sdk/pointclouds.py b/fastfuels_sdk/pointclouds.py index 343a044..5f24872 100644 --- a/fastfuels_sdk/pointclouds.py +++ b/fastfuels_sdk/pointclouds.py @@ -365,4 +365,4 @@ def _convert_api_models_to_sdk_classes( domain_id=domain_id, **response_data["als"] ) - return response_data \ No newline at end of file + return response_data diff --git a/tests/data/threedep_domain.geojson b/tests/data/threedep_domain.geojson index b44fe2a..0f996db 100644 --- a/tests/data/threedep_domain.geojson +++ b/tests/data/threedep_domain.geojson @@ -6,4 +6,4 @@ [-114.19133479742844, 46.7747518267752], [-114.19991252294375, 46.7747518267752] ] - ],"type":"Polygon"},"id":0}]} \ No newline at end of file + ],"type":"Polygon"},"id":0}]} diff --git a/tests/test_inventories.py b/tests/test_inventories.py index 9622b14..aa605eb 100644 --- a/tests/test_inventories.py +++ b/tests/test_inventories.py @@ -908,4 +908,4 @@ def test_in_place_true(self, test_point_cloud_domain, test_point_cloud_resource) assert inventories.tree is not None assert inventories.tree is tree_inventory assert inventories.tree is not original_tree - assert tree_inventory.domain_id == test_point_cloud_domain.id \ No newline at end of file + assert tree_inventory.domain_id == test_point_cloud_domain.id diff --git a/tests/test_pointclouds.py b/tests/test_pointclouds.py index 501b26a..3e2ed69 100644 --- a/tests/test_pointclouds.py +++ b/tests/test_pointclouds.py @@ -275,4 +275,4 @@ def test_delete_nonexistent_als(self, test_pointclouds): als.domain_id = uuid4().hex # Set invalid ID with pytest.raises(NotFoundException): - als.delete() \ No newline at end of file + als.delete() diff --git a/tests/utils.py b/tests/utils.py index 00c48bd..41f36ac 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -48,4 +48,4 @@ def normalize_datetime(resource): resource.created_on = resource.created_on.replace(tzinfo=None) if resource.modified_on and resource.modified_on.tzinfo: resource.modified_on = resource.modified_on.replace(tzinfo=None) - return resource \ No newline at end of file + return resource