Skip to content

rslearn.data_sources.usda_cdl

usda_cdl

Crop type data from the USDA Cropland Data Layer.

CDL

Bases: DataSource

Data source for crop type data from the USDA Cropland Data Layer.

See https://www.nass.usda.gov/Research_and_Science/Cropland/SARS1a.php for details about the data.

There is one GeoTIFF item per year from 2008. Each GeoTIFF spans the entire continental US, and has a single band.

Source code in rslearn/data_sources/usda_cdl.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class CDL(DataSource):
    """Data source for crop type data from the USDA Cropland Data Layer.

    See https://www.nass.usda.gov/Research_and_Science/Cropland/SARS1a.php for details
    about the data.

    There is one GeoTIFF item per year from 2008. Each GeoTIFF spans the entire
    continental US, and has a single band.
    """

    BASE_URL = (
        "https://www.nass.usda.gov/Research_and_Science/Cropland/Release/datasets/"
    )
    ZIP_FILENAMES = {
        2024: "2024_30m_cdls.zip",
        2023: "2023_30m_cdls.zip",
        2022: "2022_30m_cdls.zip",
        2021: "2021_30m_cdls.zip",
        2020: "2020_30m_cdls.zip",
        2019: "2019_30m_cdls.zip",
        2018: "2018_30m_cdls.zip",
        2017: "2017_30m_cdls.zip",
        2016: "2016_30m_cdls.zip",
        2015: "2015_30m_cdls.zip",
        2014: "2014_30m_cdls.zip",
        2013: "2013_30m_cdls.zip",
        2012: "2012_30m_cdls.zip",
        2011: "2011_30m_cdls.zip",
        2010: "2010_30m_cdls.zip",
        2009: "2009_30m_cdls.zip",
        2008: "2008_30m_cdls.zip",
    }

    # The bounds of each GeoTIFF in WGS84 coordinates, based on the 2023 map.
    BOUNDS = shapely.box(-127.9, 23.0, -65.3, 48.3)

    def __init__(
        self,
        timeout: timedelta = timedelta(seconds=10),
        context: DataSourceContext = DataSourceContext(),
    ):
        """Initialize a new CDL instance.

        Args:
            timeout: timeout for requests.
            context: the data source context.
        """
        self.timeout = timeout

        # Get the band name from the layer config, which should have a single band set
        # with a single band. If the layer config is not available in the context, we
        # default to "cdl".
        if context.layer_config is not None:
            if len(context.layer_config.band_sets) != 1:
                raise ValueError("expected a single band set")
            if len(context.layer_config.band_sets[0].bands) != 1:
                raise ValueError("expected band set to have a single band")
            self.band_name = context.layer_config.band_sets[0].bands[0]
        else:
            self.band_name = "cdl"

    def get_item_by_name(self, name: str) -> Item:
        """Gets an item by name.

        Args:
            name: the name of the item to get. For CDL, the item name is the filename
                of the zip file containing the per-year GeoTIFF.

        Returns:
            the Item object
        """
        year = int(name[0:4])
        geometry = STGeometry(
            WGS84_PROJECTION,
            self.BOUNDS,
            (
                datetime(year, 1, 1, tzinfo=UTC),
                datetime(year + 1, 1, 1, tzinfo=UTC),
            ),
        )
        return Item(name, geometry)

    def get_items(
        self, geometries: list[STGeometry], query_config: QueryConfig
    ) -> list[list[MatchedItemGroup[Item]]]:
        """Get a list of items in the data source intersecting the given geometries.

        Args:
            geometries: the spatiotemporal geometries
            query_config: the query configuration

        Returns:
            List of groups of items that should be retrieved for each geometry.
        """
        # First enumerate all items.
        # Then we simply pass this to match_candidate_items_to_window.
        items: list[Item] = []
        for year, fname in self.ZIP_FILENAMES.items():
            geometry = STGeometry(
                WGS84_PROJECTION,
                self.BOUNDS,
                (
                    datetime(year, 1, 1, tzinfo=UTC),
                    datetime(year + 1, 1, 1, tzinfo=UTC),
                ),
            )
            items.append(Item(fname, geometry))

        groups = []
        for geometry in geometries:
            cur_groups = match_candidate_items_to_window(geometry, items, query_config)
            groups.append(cur_groups)

        return groups

    def deserialize_item(self, serialized_item: dict) -> Item:
        """Deserializes an item from JSON-decoded data."""
        return Item.deserialize(serialized_item)

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[Item],
        geometries: list[list[STGeometry]],
    ) -> None:
        """Ingest items into the given tile store.

        Args:
            tile_store: the tile store to ingest into
            items: the items to ingest
            geometries: a list of geometries needed for each item
        """
        for item in items:
            if tile_store.is_raster_ready(item, [self.band_name]):
                continue

            # Download the zip file.
            url = self.BASE_URL + item.name
            logger.debug(f"Downloading CDL GeoTIFF from {url}")
            response = requests.get(
                url, stream=True, timeout=self.timeout.total_seconds()
            )
            response.raise_for_status()

            with tempfile.TemporaryDirectory() as tmp_dir:
                # Store it in temporary directory.
                zip_fname = os.path.join(tmp_dir, "data.zip")
                with open(zip_fname, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)

                # Extract the .tif file.
                logger.debug(f"Extracting GeoTIFF from {item.name}")
                with zipfile.ZipFile(zip_fname) as zip_f:
                    candidate_member_names = [
                        member_name
                        for member_name in zip_f.namelist()
                        if member_name.endswith(".tif")
                    ]
                    if len(candidate_member_names) != 1:
                        raise ValueError(
                            f"expected CDL zip to have one .tif file but got {candidate_member_names}"
                        )
                    local_fname = zip_f.extract(candidate_member_names[0], path=tmp_dir)

                # Now we can ingest it.
                logger.debug(f"Ingesting data for {item.name}")
                tile_store.write_raster_file(
                    item,
                    [self.band_name],
                    UPath(local_fname),
                    time_range=item.geometry.time_range,
                )

get_item_by_name 

get_item_by_name(name: str) -> Item

Gets an item by name.

Parameters:

Name Type Description Default
name str

the name of the item to get. For CDL, the item name is the filename of the zip file containing the per-year GeoTIFF.

 required

Returns:

Type Description
Item

the Item object
Source code in rslearn/data_sources/usda_cdl.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_item_by_name(self, name: str) -> Item:
    """Gets an item by name.

    Args:
        name: the name of the item to get. For CDL, the item name is the filename
            of the zip file containing the per-year GeoTIFF.

    Returns:
        the Item object
    """
    year = int(name[0:4])
    geometry = STGeometry(
        WGS84_PROJECTION,
        self.BOUNDS,
        (
            datetime(year, 1, 1, tzinfo=UTC),
            datetime(year + 1, 1, 1, tzinfo=UTC),
        ),
    )
    return Item(name, geometry)

get_items 

get_items(geometries: list[STGeometry], query_config: QueryConfig) -> list[list[MatchedItemGroup[Item]]]

Get a list of items in the data source intersecting the given geometries.

Parameters:

Name Type Description Default
geometries list[STGeometry]

the spatiotemporal geometries

 required
query_config QueryConfig

the query configuration

 required

Returns:

Type Description
list[list[MatchedItemGroup[Item]]]

List of groups of items that should be retrieved for each geometry.
Source code in rslearn/data_sources/usda_cdl.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def get_items(
    self, geometries: list[STGeometry], query_config: QueryConfig
) -> list[list[MatchedItemGroup[Item]]]:
    """Get a list of items in the data source intersecting the given geometries.

    Args:
        geometries: the spatiotemporal geometries
        query_config: the query configuration

    Returns:
        List of groups of items that should be retrieved for each geometry.
    """
    # First enumerate all items.
    # Then we simply pass this to match_candidate_items_to_window.
    items: list[Item] = []
    for year, fname in self.ZIP_FILENAMES.items():
        geometry = STGeometry(
            WGS84_PROJECTION,
            self.BOUNDS,
            (
                datetime(year, 1, 1, tzinfo=UTC),
                datetime(year + 1, 1, 1, tzinfo=UTC),
            ),
        )
        items.append(Item(fname, geometry))

    groups = []
    for geometry in geometries:
        cur_groups = match_candidate_items_to_window(geometry, items, query_config)
        groups.append(cur_groups)

    return groups

deserialize_item 

deserialize_item(serialized_item: dict) -> Item

Deserializes an item from JSON-decoded data.

Source code in rslearn/data_sources/usda_cdl.py 
139
140
141
def deserialize_item(self, serialized_item: dict) -> Item:
    """Deserializes an item from JSON-decoded data."""
    return Item.deserialize(serialized_item)

ingest 

ingest(tile_store: TileStoreWithLayer, items: list[Item], geometries: list[list[STGeometry]]) -> None

Ingest items into the given tile store.

Parameters:

Name Type Description Default
tile_store TileStoreWithLayer

the tile store to ingest into

 required
items list[Item]

the items to ingest

 required
geometries list[list[STGeometry]]

a list of geometries needed for each item

 required
Source code in rslearn/data_sources/usda_cdl.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[Item],
    geometries: list[list[STGeometry]],
) -> None:
    """Ingest items into the given tile store.

    Args:
        tile_store: the tile store to ingest into
        items: the items to ingest
        geometries: a list of geometries needed for each item
    """
    for item in items:
        if tile_store.is_raster_ready(item, [self.band_name]):
            continue

        # Download the zip file.
        url = self.BASE_URL + item.name
        logger.debug(f"Downloading CDL GeoTIFF from {url}")
        response = requests.get(
            url, stream=True, timeout=self.timeout.total_seconds()
        )
        response.raise_for_status()

        with tempfile.TemporaryDirectory() as tmp_dir:
            # Store it in temporary directory.
            zip_fname = os.path.join(tmp_dir, "data.zip")
            with open(zip_fname, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)

            # Extract the .tif file.
            logger.debug(f"Extracting GeoTIFF from {item.name}")
            with zipfile.ZipFile(zip_fname) as zip_f:
                candidate_member_names = [
                    member_name
                    for member_name in zip_f.namelist()
                    if member_name.endswith(".tif")
                ]
                if len(candidate_member_names) != 1:
                    raise ValueError(
                        f"expected CDL zip to have one .tif file but got {candidate_member_names}"
                    )
                local_fname = zip_f.extract(candidate_member_names[0], path=tmp_dir)

            # Now we can ingest it.
            logger.debug(f"Ingesting data for {item.name}")
            tile_store.write_raster_file(
                item,
                [self.band_name],
                UPath(local_fname),
                time_range=item.geometry.time_range,
            )