File size: 7,310 Bytes
64ab846 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 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 | from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.logging import get_run_logger
import pandas as pd
import os
import yaml
import pickle
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from xgboost import XGBRegressor
from sklearn.metrics import r2_score
from pipelines import xgcast_pipeline
from pipelines.data_collection_pipeline import get_satellite_data
from pipelines.preprocessing_pipeline import save_to_file, read_satellite_data, update_historical_weather_data
from pipelines.resample_impute_pipeline import interpolate_satellite_data, get_tabular_satellite, generate_weather_forecast_df
from pipelines.xgcast_pipeline import normalise_df
from pipelines.model_preparation_pipeline import model_preparation_pipeline, data_availability
with open('config/params.yml') as file:
config = yaml.safe_load(file)
with open('config/xgcast_params.yml') as file:
model_params = yaml.safe_load(file)
datetime_col = config['datetime_col']
value_col = config['value_col']
datastream_name_col = config['datastream_name_col']
resampling_window = config['resampling_window']
resampling = int(resampling_window.split('h')[0])
num_predictions = 3 * (24 // resampling)
### Flows
@flow(name='xgcast_preparation_pipeline', retries=1, task_runner=ThreadPoolTaskRunner())
def xgcast_run(consortium_name):
logger = get_run_logger()
logger.info(f'Starting XGCast run for consortium {consortium_name}!')
has_weather_data, has_crop_data, has_soil_data, has_remote_sensing_data = True, True, True, True
if not os.path.exists(f'data//03_primary//field_sensor_data_{consortium_name}.parquet'):
logger.error(f'Table of field sensor measurements not found for consortium {consortium_name}. Stopping.')
return
if not os.path.exists(f'data//03_primary//irrigation_data_{consortium_name}.parquet'):
logger.error(f'Table of irrigation data not found for consortium {consortium_name}. Stopping.')
return
if not os.path.exists(f'data//03_primary//locations_ids_{consortium_name}.parquet'):
logger.error(f'Table of locations ids not found for consortium {consortium_name}. Stopping.')
return
if not os.path.exists(f'data//03_primary//crop_type_data_{consortium_name}.pickle'):
has_crop_data = False
logger.info(f'Crop data not found for consortium {consortium_name}')
if not os.path.exists(f'data//03_primary//soil_type_data_{consortium_name}.parquet'):
has_soil_data = False
logger.info(f'Soil data not found for consortium {consortium_name}')
if not os.path.exists(f'data//03_primary//weather_data_{consortium_name}.parquet'):
has_weather_data = False
logger.info(f'On-field weather data not found for consortium {consortium_name}')
if not os.path.exists(f'data//03_primary//historical_weather_data_{consortium_name}.parquet'):
logger.info(f'Public weather data not found for consortium {consortium_name}. Downloading data...')
locations_ids = pd.read_parquet(f'data//03_primary//locations_ids_{consortium_name}.parquet')
historical_weather_data = update_historical_weather_data(consortium_name=consortium_name, locations_ids=locations_ids)
save_to_file(df=historical_weather_data, output_file=f'data//03_primary//historical_weather_data_{consortium_name}.parquet')
logger.info(f'Public weather data has been downloaded and saved for consortium {consortium_name}.')
if not os.path.exists(f'data//03_primary//forecasted_weather_data_{consortium_name}.parquet'):
# for model training purposes we use 'real' weather forecast data
logger.info(f'Weather forecast data not found for consortium {consortium_name}. Generating data for model training purposes...')
historical_weather_data = pd.read_parquet(f'data//03_primary//historical_weather_data_{consortium_name}.parquet')
forecasted_weather_data = generate_weather_forecast_df(historical_weather_data.reset_index())
save_to_file(df=forecasted_weather_data, output_file=f'data//03_primary//forecasted_weather_data_{consortium_name}.parquet')
logger.info(f'Weather forecast data has been downloaded and saved for consortium {consortium_name}.')
if not os.path.exists(f'data//03_primary//remote_sensing_data_final_{consortium_name}.parquet'):
try:
# TODO: make sure such params exist. AS THE USED TASKS WERE MEANT TO BE PRIVATE, THERE ARE REQUIRED PARAMS THAT MIGHT NOT EXIST
# pre anonmym params are being called in copernicus data request as bbox and data folder are needed
# download satellite data
# todo: pre anonmym params are being called in copernicus data request as bbox and data folder are needed
# todo: start_date end_date and request_script are init in the file, so they pre_anonym_params is needed
get_satellite_data.submit(
consortium=consortium_name,
overwrite=False
)
# transform into nc file containing all data
remote_sensing_data = read_satellite_data.submit(consortium_name=consortium_name).result()
# we save only final
#save_to_file(df=remote_sensing_data,
# output_file=f'data//02_preprocessed//remote_sensing_data_{consortium_name}.nc')
# resample
satellite_interpolated = interpolate_satellite_data.submit(remote_sensing_data).result()
# agg
locations_ids = pd.read_parquet(f'data//03_primary//locations_ids_{consortium_name}.parquet')
# todo: requires pre_anonym_params - 'spatial_agg_method' and 'field_agg'
spatial_agg_method = config['spatial_agg_method']
field_agg = config['field_agg']
remote_sensing_data_final = get_tabular_satellite.submit(
consortium_name,
satellite_interpolated,
locations_ids,
method=spatial_agg_method,
sensor_field_mapping=f'config/sensor_field_mapping.yaml',
field_agg=field_agg).result()
# save
save_to_file(df = remote_sensing_data_final, output_file=f'data//03_primary//remote_sensing_data_final_{consortium_name}.parquet')
except:
has_remote_sensing_data = False
logger.warning('Warning: satellite data could not be downloaded.')
data_availability[consortium_name] = {'has_weather_data': has_weather_data, 'has_crop_data': has_crop_data, 'has_soil_data': has_soil_data, 'has_remote_sensing_data': has_remote_sensing_data}
model_preparation_pipeline(override_consortia=[consortium_name], override_data_availability=data_availability[consortium_name])
xgcast_pipeline.xgcast_preparation_pipeline(override_consortia=[consortium_name], override_data_availability=data_availability[consortium_name])
xgcast_pipeline.xgcast_model_pipeline(override_consortia=[consortium_name])
|