LANDFIRE fuel model creators: expose the disturbance option

docs/v2/guides/creating-grids.md

@@ -384,6 +384,27 @@ The single `annual_disturbance` band holds categorical LDist codes, fetched on
 demand from the LANDFIRE Product Service. Pass `version` to pin a vintage; it
 defaults to the API's current version.
 
+To update a LANDFIRE fuel model grid (FBFM13, FBFM40, or FCCS) for recent
+disturbance where the newest release doesn't fully cover the domain yet, pass
+`disturbance="landfire"`:
+
+```python
+grid = ff.grids.create_fuel_model_grid_from_landfire_fbfm40(
+    domain, disturbance="landfire", output_resolution_m=30
+)
+grid.wait()
+```
+
+```python
+>>> grid.source["disturbance"]
+{'ldist_version': '2025', 'fdist_version': '2024', 'source': 'landfire', 'mode': 'ldist_and_last_year_fdist'}
+```
+
+The API records the plan it applied on `source.disturbance`. Omit `version` (or
+set it to the baseline release); for FBFM40 `disturbance` can't be combined with
+`season`. The API returns a 422 if the newest release already fully covers the
+domain.
+
 ## 3D tree fuel grids (voxelization)
 
 The 3D canopy fuel grid — per-voxel bulk density, the input 3D fire models

fastfuels_sdk/v2/grids.py

@@ -138,6 +138,7 @@
     LandfireCanopyFuelBand,
     LandfireCanopyVersion,
     LandfireCoverageResponse,
+    LandfireDisturbanceRequest,
     LandfireDisturbanceVersion,
     LandfireFbfm13Version,
     LandfireFbfm40Version,
@@ -409,6 +410,28 @@ def _build_boundary_scatter(scatter):
     )
 
 
+def _build_landfire_disturbance(disturbance):
+    """Translate a disturbance kwarg into a LandfireDisturbanceRequest or UNSET.
+
+    - ``None`` -> ``UNSET`` (no disturbance adjustment).
+    - ``"landfire"`` -> ``LandfireDisturbanceRequest(source="landfire")``.
+    - a mapping -> a ``LandfireDisturbanceRequest`` built from its fields.
+    - a ``LandfireDisturbanceRequest`` -> passed through unchanged.
+    """
+    if disturbance is None:
+        return UNSET
+    if isinstance(disturbance, LandfireDisturbanceRequest):
+        return disturbance
+    if disturbance == "landfire":
+        return LandfireDisturbanceRequest(source="landfire")
+    if isinstance(disturbance, Mapping):
+        return LandfireDisturbanceRequest(**disturbance)
+    raise ValueError(
+        'disturbance must be "landfire", a mapping like {"source": "landfire"}, '
+        f"or a LandfireDisturbanceRequest, got {disturbance!r}."
+    )
+
+
 def _build_method_overrides(method_overrides):
     """Translate a resample ``method_overrides`` kwarg into a model or UNSET.
 
@@ -1869,6 +1892,7 @@ def create_fuel_model_grid_from_landfire_fbfm13(
     description: str = "",
     tags: Optional[List[str]] = None,
     modifications: Optional[list] = None,
+    disturbance=None,
 ) -> Grid:
     """Create a fuel model grid from LANDFIRE Anderson 13 fuel models.
 
@@ -1905,15 +1929,31 @@ def create_fuel_model_grid_from_landfire_fbfm13(
         Tags for the grid.
     modifications : list, optional
         Modification rules applied after the grid is built.
+    disturbance : str, dict, or LandfireDisturbanceRequest, optional
+        Update the baseline release for recent disturbance where the newest
+        LANDFIRE release doesn't fully cover the domain. Pass ``"landfire"``,
+        ``{"source": "landfire"}``, or a ``LandfireDisturbanceRequest``. The
+        plan the API chose is recorded on the grid as ``source.disturbance``.
+        Omit ``version`` (or set it to the baseline release). ``None``
+        (default) applies no adjustment.
 
     Returns
     -------
     Grid
         The created Grid object (job status ``"pending"`` or ``"running"``).
+
+    Raises
+    ------
+    ValueError
+        If ``disturbance`` isn't a recognized form.
+    UnprocessableEntityException
+        If the API rejects the request, e.g. ``version`` isn't the baseline or
+        the newest release already fully covers the domain.
     """
     request_body = CreateLandfireFbfm13Request(
         version=LandfireFbfm13Version(version) if version is not None else UNSET,
         remove_non_burnable=_enum_list(remove_non_burnable, NonBurnableFuelModel),
+        disturbance=_build_landfire_disturbance(disturbance),
         boundary_scatter=_build_boundary_scatter(boundary_scatter),
         alignment=_build_alignment(output_resolution_m, align_to, align, resampling),
         extent_buffer_cells=extent_buffer_cells,
@@ -1943,6 +1983,7 @@ def create_fuel_model_grid_from_landfire_fbfm40(
     description: str = "",
     tags: Optional[List[str]] = None,
     modifications: Optional[list] = None,
+    disturbance=None,
 ) -> Grid:
     """Create a fuel model grid from LANDFIRE 40 Scott & Burgan fuel models.
 
@@ -1987,16 +2028,32 @@ def create_fuel_model_grid_from_landfire_fbfm40(
         Tags for the grid.
     modifications : list, optional
         Modification rules applied after the grid is built.
+    disturbance : str, dict, or LandfireDisturbanceRequest, optional
+        Update the baseline release for recent disturbance where the newest
+        LANDFIRE release doesn't fully cover the domain. Pass ``"landfire"``,
+        ``{"source": "landfire"}``, or a ``LandfireDisturbanceRequest``. The
+        plan the API chose is recorded on the grid as ``source.disturbance``.
+        Omit ``version`` (or set it to the baseline release), and don't
+        combine with ``season``. ``None`` (default) applies no adjustment.
 
     Returns
     -------
     Grid
         The created Grid object (job status "pending" or "running").
+
+    Raises
+    ------
+    ValueError
+        If ``disturbance`` isn't a recognized form.
+    UnprocessableEntityException
+        If the API rejects the request, e.g. ``version`` isn't the baseline or
+        the newest release already fully covers the domain.
     """
     request_body = CreateLandfireFbfm40Request(
         version=LandfireFbfm40Version(version) if version is not None else UNSET,
         season=LandfireSeason(season) if season is not None else UNSET,
         remove_non_burnable=_enum_list(remove_non_burnable, NonBurnableFuelModel),
+        disturbance=_build_landfire_disturbance(disturbance),
         boundary_scatter=_build_boundary_scatter(boundary_scatter),
         alignment=_build_alignment(output_resolution_m, align_to, align, resampling),
         extent_buffer_cells=extent_buffer_cells,
@@ -2025,6 +2082,7 @@ def create_fuel_model_grid_from_landfire_fccs(
     description: str = "",
     tags: Optional[List[str]] = None,
     modifications: Optional[list] = None,
+    disturbance=None,
 ) -> Grid:
     """Create a fuel model grid from LANDFIRE FCCS fuelbeds.
 
@@ -2060,15 +2118,31 @@ def create_fuel_model_grid_from_landfire_fccs(
         Tags for the grid.
     modifications : list, optional
         Modification rules applied after the grid is built.
+    disturbance : str, dict, or LandfireDisturbanceRequest, optional
+        Update the baseline release for recent disturbance where the newest
+        LANDFIRE release doesn't fully cover the domain. Pass ``"landfire"``,
+        ``{"source": "landfire"}``, or a ``LandfireDisturbanceRequest``. The
+        plan the API chose is recorded on the grid as ``source.disturbance``.
+        Omit ``version`` (or set it to the baseline release). ``None``
+        (default) applies no adjustment.
 
     Returns
     -------
     Grid
         The created Grid object (job status "pending" or "running").
+
+    Raises
+    ------
+    ValueError
+        If ``disturbance`` isn't a recognized form.
+    UnprocessableEntityException
+        If the API rejects the request, e.g. ``version`` isn't the baseline or
+        the newest release already fully covers the domain.
     """
     request_body = CreateLandfireFccsRequest(
         version=LandfireFccsVersion(version) if version is not None else UNSET,
         remove_bare_ground=remove_bare_ground,
+        disturbance=_build_landfire_disturbance(disturbance),
         boundary_scatter=_build_boundary_scatter(boundary_scatter),
         alignment=_build_alignment(output_resolution_m, align_to, align, resampling),
         extent_buffer_cells=extent_buffer_cells,

tests/v2/test_grids.py

@@ -56,6 +56,7 @@
     Band,
     BandType,
     BoundaryScatter,
+    LandfireDisturbanceRequest,
     CanopyCbhPercentile,
     ChmMaxAggregation,
     ChmMeanAggregation,
@@ -1534,6 +1535,120 @@ def test_create_seasonal_reports_projected_year(self):
             domain.delete(force=True)
 
 
+class TestLandfireDisturbance:
+    """The disturbance option on the LANDFIRE FBFM13 / FBFM40 / FCCS creators."""
+
+    CREATORS = [
+        ("create_landfire_fbfm13", create_fuel_model_grid_from_landfire_fbfm13, "2024"),
+        ("create_landfire_fbfm40", create_fuel_model_grid_from_landfire_fbfm40, "2024"),
+        ("create_landfire_fccs", create_fuel_model_grid_from_landfire_fccs, "2023"),
+    ]
+
+    def _capture(self, monkeypatch, endpoint):
+        captured = {}
+
+        def fake_create(domain_id, *, client, body):
+            captured.update(body=body)
+            return Response(
+                status_code=HTTPStatus.CREATED,
+                content=b"",
+                headers={},
+                parsed=Grid(
+                    id="grid-id",
+                    domain_id="domain-id",
+                    status=JobStatus.PENDING,
+                    source=GridSource(),
+                    bands=[],
+                ),
+            )
+
+        monkeypatch.setattr(grids, "ensure_client", lambda: object())
+        monkeypatch.setattr(getattr(grids, endpoint), "sync_detailed", fake_create)
+        return captured
+
+    @pytest.mark.parametrize("endpoint,creator,baseline", CREATORS)
+    @pytest.mark.parametrize(
+        "value",
+        [
+            "landfire",
+            {"source": "landfire"},
+            LandfireDisturbanceRequest(source="landfire"),
+        ],
+    )
+    def test_request_body(self, monkeypatch, endpoint, creator, baseline, value):
+        captured = self._capture(monkeypatch, endpoint)
+        creator(SimpleNamespace(id="domain-id"), version=baseline, disturbance=value)
+        body = captured["body"]
+        assert isinstance(body.disturbance, LandfireDisturbanceRequest)
+        assert body.disturbance.source == "landfire"
+        assert body.to_dict()["disturbance"] == {"source": "landfire"}
+
+    @pytest.mark.parametrize("endpoint,creator,baseline", CREATORS)
+    def test_default_is_unset(self, monkeypatch, endpoint, creator, baseline):
+        captured = self._capture(monkeypatch, endpoint)
+        creator(SimpleNamespace(id="domain-id"))
+        assert captured["body"].disturbance is UNSET
+        assert "disturbance" not in captured["body"].to_dict()
+
+    @pytest.mark.parametrize("endpoint,creator,baseline", CREATORS)
+    @pytest.mark.parametrize("value", ["fire", True, 3])
+    def test_invalid_value_raises(
+        self, monkeypatch, endpoint, creator, baseline, value
+    ):
+        self._capture(monkeypatch, endpoint)
+        with pytest.raises(ValueError, match="disturbance"):
+            creator(SimpleNamespace(id="domain-id"), disturbance=value)
+
+    def test_live_fbfm40_disturbance(self, real_dispatch):
+        # A small domain on the Colorado Front Range, where LANDFIRE's newest
+        # release only partly covers the area. If that release catches up, the
+        # API rejects disturbance and the test skips.
+        x, y, d = -105.30, 40.00, 0.01
+        domain = Domain.from_geojson(
+            {
+                "type": "FeatureCollection",
+                "features": [
+                    {
+                        "type": "Feature",
+                        "properties": {},
+                        "geometry": {
+                            "type": "Polygon",
+                            "coordinates": [
+                                [
+                                    [x, y],
+                                    [x + d, y],
+                                    [x + d, y + d],
+                                    [x, y + d],
+                                    [x, y],
+                                ]
+                            ],
+                        },
+                    }
+                ],
+            },
+            name="throwaway_disturbance_domain",
+            tags=["sdk-test"],
+        )
+        try:
+            try:
+                grid = create_fuel_model_grid_from_landfire_fbfm40(
+                    domain,
+                    disturbance="landfire",
+                    output_resolution_m=30,
+                    name="throwaway_fbfm40_disturbance",
+                )
+            except UnprocessableEntityException as exc:
+                pytest.skip(f"Disturbance not applicable to this domain: {exc}")
+            try:
+                grid.wait()
+                assert grid.status == JobStatus.COMPLETED
+                assert grid.source["disturbance"]["source"] == "landfire"
+            finally:
+                grid.delete()
+        finally:
+            domain.delete(force=True)
+
+
 class TestRepresentedYear:
     """Unit tests for the Grid.represented_year accessor (no API)."""