<jupyter_start><jupyter_text>Post processing of aboveground biomass dataset InputRandom forest model prediction results from inference.ipynb. These are parquet files (1 for eachlandsat scene x year) with columns x, y, biomass. x, y are in lat/lon coordinates, and biomass is inunit of Mg biomass / ha and only accounts for aboveground, live, woody biomass. ProcessesFor each 10x10 degree tile in our template1. merge and mosaic all landsat scenes within a 10x10 degree tile for all years available and store   the data in zarr format2. fill gaps within the biomass dataset by xarray interpolate_na with linear method (first through   dim time, then through dim x, then dim y)3. mask with MODIS MCD12Q1 land cover dataset to only select the forest pixels4. calculate belowground biomass and deadwood and litter<jupyter_code>%load_ext autoreload
%autoreload 2

from pyproj import CRS
import boto3
from rasterio.session import AWSSession
from s3fs import S3FileSystem
aws_session = AWSSession(boto3.Session(),#profile_name='default'), 
                         requester_pays=True)
fs = S3FileSystem(requester_pays=True)
import xgboost as xgb

from osgeo.gdal import VSICurlClearCache
import rasterio as rio
import numpy as np
import xarray as xr
import dask
import os
import fsspec

import rioxarray # for the extension to load
import pandas as pd
from datetime import datetime

from dask_gateway import Gateway
from carbonplan_trace.v1.landsat_preprocess import access_credentials, test_credentials
from carbonplan_trace.v1.inference import predict, predict_delayed 
from carbonplan_trace.v1 import utils, postprocess, load
from carbonplan_trace.tiles import tiles
from carbonplan_trace.v1.landsat_preprocess import access_credentials, test_credentials
import prefect
from prefect import task, Flow, Parameter
from prefect.executors import DaskExecutor
from prefect.utilities.debug import raise_on_exception
from datetime import datetime as time

from carbonplan_trace import version
%reload_ext watermark
print(version)
watermark -d -n -t -u -v -p carbonplan_trace -h -m -g -r -b
dask.config.set({"array.slicing.split_large_chunks": False})
dask.config.set({"distributed.comm.timeouts.tcp": "50s"})
dask.config.set({"distributed.comm.timeouts.connect": "50s"})
kind_of_cluster = "local"
if kind_of_cluster == "local":
    # spin up local cluster. must be on big enough machine
    from dask.distributed import Client

    local_cluster_client = Client(n_workers=5, threads_per_worker=1, resources={"workertoken": 1})

    local_cluster_client
elif kind_of_cluster == "remote":
    gateway = Gateway()
    options = gateway.cluster_options()
    options.environment = {
        "AWS_REQUEST_PAYER": "requester",
        "AWS_REGION_NAME": "us-west-2",
        "DASK_DISTRIBUTED__WORKER__RESOURCES__WORKERTOKEN": "1",
    }
    options.worker_cores = 1
    options.worker_memory = 31

    options.image = "carbonplan/trace-python-notebook:latest"
    cluster = gateway.new_cluster(cluster_options=options)
    cluster.adapt(minimum=0, maximum=150)
#     cluster.scale(200)
postprocess._set_thread_settings()
# cluster.shutdown()
# local_cluster_client.shutdown()
# gateway = Gateway()
# clusters = gateway.list_clusters()
# cluster = gateway.connect(clusters[0].name)
# cluster.shutdown()
# cluster
# client = cluster.get_client()
# client
local_cluster_client
# cluster.shutdown()
access_key_id, secret_access_key = access_credentials()
tasks = []
# define starting and ending years (will want to go back to 2014 but that might not be ready right now)
year0, year1 = 2014, 2021
# define the size of subtile you want to work in (2 degrees recommended)
tile_degree_size = 2
# if you want to write the metadata for the zarr store
write_tile_metadata = True
chunks_dict = {"lat": 1000, "lon": 1000}
log_bucket = "s3://carbonplan-climatetrace/v1.2/postprocess_log/"
completed_subtiles = fs.ls(log_bucket)
completed_subtiles = [subtile.split("/")[-1].split(".txt")[0] for subtile in completed_subtiles]
len(completed_subtiles)
len(completed_subtiles) - 6995
print(datetime.now())
running_tiles = [tile for tile in tiles]  # if ("E" in tile and "N" in tile)
# running_tiles = ["40N_120W"]
len(running_tiles)
parameters_list = []
# for tile in tiles:
for tile in running_tiles:
    lat_tag, lon_tag = utils.get_lat_lon_tags_from_tile_path(tile)
    lat_lon_box = utils.parse_bounding_box_from_lat_lon_tags(lat_tag, lon_tag)
    # find the lat_lon_box for that tile
    min_lat, max_lat, min_lon, max_lon = lat_lon_box
    # initialize empty dataset. only need to do this once, and not if the tile has already been processed
    data_path = postprocess.initialize_empty_dataset(
        lat_tag, lon_tag, year0, year1, write_tile_metadata=write_tile_metadata
    )
    # now we'll split up each of those tiles into smaller subtiles of length `tile_degree_size`
    # and run through those. In this case since we've specified 2, we'll have 25 in each box

    prefect_parameters = {
        "MIN_LAT": min_lat,
        "MIN_LON": min_lon,
        "YEAR_0": year0,
        "YEAR_1": year1,
        "TILE_DEGREE_SIZE": tile_degree_size,
        "DATA_PATH": data_path,
        "ACCESS_KEY_ID": access_key_id,
        "SECRET_ACCESS_KEY": secret_access_key,
        "CHUNKS_DICT": chunks_dict,
    }

    for lat_increment in np.arange(0, 10, tile_degree_size):
        for lon_increment in np.arange(0, 10, tile_degree_size):
            task_tag = "{}_{}_{}_{}".format(min_lat, min_lon, lat_increment, lon_increment)
            if task_tag in completed_subtiles:
                continue
            else:
                increment_parameters = prefect_parameters.copy()
                increment_parameters["LAT_INCREMENT"] = lat_increment
                increment_parameters["LON_INCREMENT"] = lon_increment
                parameters_list.append(increment_parameters)
#         tasks.append(client.compute(postprocess_delayed(subtile_ul_lat, subtile_ul_lon, year0, year1, tile_degree_size, mapper)))
len(parameters_list)
# print(datetime.now())
# postprocess.postprocess_subtile(parameters_list[4])
print(datetime.now())
import random
random.shuffle(parameters_list)
if len(parameters_list) > 1000:
    parameters_list = parameters_list[:900]
# postprocess.postprocess_subtile(parameters_list[4])
if kind_of_cluster == "local":
    executor = DaskExecutor(address=local_cluster_client.scheduler.address)
elif kind_of_cluster == "remote":
    executor = DaskExecutor(
        address=client.scheduler.address,
        client_kwargs={"security": cluster.security},
        debug=True,
    )
def fail_nicely(task, old_state, new_state):
    if new_state.is_running():
        print("running!")
    if new_state.is_failed():
        print("this task {} failed".format(task))
        raise ValueError("OH NO")  # function that sends a notification
    return new_state
# prefect.engine.signals.state.Skipped()
postprocess_task = task(
    postprocess.postprocess_subtile,  # .test_to_zarr,#
    tags=["dask-resource:workertoken=1"],
    state_handlers=[fail_nicely],
)
with Flow("Postprocessing") as flow:
    # Run postprocess
    postprocess_task.map(parameters_list)
# with raise_on_exception():
# if running locally (no cluster)
#     flow.run()
# if running on cluster
flow.run(executor=executor)
cluster.shutdown()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08844371763423275, 'fraction_numerical': 0.015126737530662305, 'mean_word_length': 4.497378277153558, 'pattern_counts': {'":': 23, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '473786', 'license': 'MIT', 'n_tokens_mistral': 2393, 'n_tokens_neox': 2280, 'n_words': 635, 'path': 'notebooks/processing/postprocess.ipynb', 'repo_name': 'carbonplan/trace', 'size': '7338'}