[IMPA] Changing process satelllite file to optimize ram.

.gitattributes

@@ -0,0 +1 @@
+*.ckpt filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -1,6 +1,7 @@
 # Folders that Git should ignore
 *.ipynb_checkpoints/
-pipelines/precipitation_model/impa/data/
+pipelines/precipitation_model/impa/data/raw/
+pipelines/precipitation_model/impa/data/processed/satellite/
 pipelines/precipitation_model/impa/models/data/
 pipelines/precipitation_model/impa/models/docs/
 pipelines/precipitation_model/impa/models/eval/
@@ -48,7 +49,6 @@ setup.py
 *.csv
 *.xls
 *.pkl
-*.npy
 *.Rdata
 *.Rds
 *.hdf

pipelines/precipitation_model/impa/flows.py

@@ -100,22 +100,27 @@
     #########################
 
     # Input arguments (These can be passed via Prefect Parameters or CLI)
+    n_historical_days = 1
     dt = get_start_datetime(start_datetime=start_datetime)
-    relevant_dts, days_of_year, years = get_relevant_dates_informations(dt=dt)
+    relevant_dts, days_of_year, years, relevant_times = get_relevant_dates_informations(
+        dt=dt, n_historical_days=n_historical_days
+    )
 
     # Download data from s3
     downloaded_files_rr = download_files_from_s3(
         product="ABI-L2-RRQPEF",
         relevant_dts=relevant_dts,
         days_of_year=days_of_year,
         years=years,
+        relevant_times=relevant_times,
         download_base_path=download_base_path,
     )
     downloaded_files_achaf = download_files_from_s3(
         product="ABI-L2-ACHAF",
         relevant_dts=relevant_dts,
         days_of_year=days_of_year,
         years=years,
+        relevant_times=relevant_times,
         download_base_path=download_base_path,
     )
 
@@ -124,14 +129,17 @@
         day_of_year=days_of_year[0],
         num_workers=num_workers,
         product="ABI-L2-RRQPEF",
+        n_historical_days=n_historical_days,
         wait=downloaded_files_rr,
     )
     data_processed_achaf = process_satellite_task(
         year=years[0],
         day_of_year=days_of_year[0],
         num_workers=num_workers,
         product="ABI-L2-ACHAF",
-        wait=[downloaded_files_achaf, data_processed_rr],
+        n_historical_days=n_historical_days,
+        wait=[downloaded_files_achaf],
+        # wait=[downloaded_files_achaf, data_processed_rr],
     )
     dfr = build_dataframe_task(num_workers, dt, wait=[data_processed_rr, data_processed_achaf])
     output_predict_filepaths = get_predictions(num_workers=num_workers, cuda=cuda, wait=dfr)
@@ -201,9 +209,9 @@
     labels=[
         constants.WEATHER_FORECAST_AGENT_LABEL.value,
     ],
-    cpu_request="1",
+    cpu_request="500m",
     memory_limit="30Gi",
     memory_request="15Gi",
 )
 prediction_previsao_chuva_impa.schedule = prediction_schedule
-prediction_previsao_chuva_impa.executor = LocalDaskExecutor(num_workers=2)  # 10
+prediction_previsao_chuva_impa.executor = LocalDaskExecutor(num_workers=50)

pipelines/precipitation_model/impa/models/EVONET/params.json

@@ -0,0 +1 @@
+{"normalized": 3, "batch": 16, "loss": 2, "weights": 2, "data_modification": null, "n_after": 18, "n_before": 12, "n_epochs": 50, "predict_dataset": null, "lambda_": 0.1, "learning_rate": 0.0002, "model_name": "Evolution_Network"}

pipelines/precipitation_model/impa/models/Mamba/params.json

@@ -0,0 +1 @@
+{"normalized": 2, "learning_rate": 0.00027, "n_after": 20, "n_before": 10, "std_fac": 10.0, "n_epoch": 100, "weight_bool": 1, "data_modification": null, "model_name": "Mamba"}

pipelines/precipitation_model/impa/models/MetNet3/params.json

@@ -0,0 +1 @@
+{"normalized": 3, "batch": 4, "loss": 2, "weights": 2, "data_modification": null, "n_after": 18, "n_before": 12, "n_epochs": 25, "predict_dataset": null, "learning_rate": 0.0001, "weight_decay": 0.01, "depth": 8, "dim": 128, "lead_time_dim_embedding": null, "resnet_block_depth": 2, "correct_context": true, "model_name": "Metnet3"}

pipelines/precipitation_model/impa/models/NowcastNet/params.json

@@ -0,0 +1 @@
+{"normalized": 3, "batch": 16, "loss": 2, "weights": 2, "data_modification": null, "n_after": 18, "n_before": 4, "n_epochs": 30, "predict_dataset": null, "discriminator_learning_rate": 0.0002, "generator_learning_rate": 0.0002, "alpha": 6, "beta": 20, "generation_steps": 4, "predictions": 2, "predictions_option": 2, "ckpt_file_predictions": "1--best-epoch=3", "model_name": "NowcastNet"}

pipelines/precipitation_model/impa/models/UNET/params.json

@@ -0,0 +1,3 @@
+{"normalized": 3, "batch": 16, "loss": 2, "weights": 2, "data_modification": null, "n_after": 18, "n_before": 12, "n_epochs": 50, "predict_dataset": null, "learning_rate": 0.0002, "weight_decay": 0.01, "bilinear": true, "dimension_division": null, "model_name": "UNET"}
+
+

pipelines/precipitation_model/impa/src/data/process/build_dataframe.py

@@ -10,6 +10,7 @@
 
 import h5py
 import numpy as np
+from prefeitura_rio.pipelines_utils.logging import log  # pylint: disable=E0611, E0401
 from tqdm import tqdm
 
 from pipelines.precipitation_model.impa.src.data.process.SatelliteData import (
@@ -109,14 +110,18 @@ def build_dataframe(
             )
             exit(0)
 
+    log("Loading grids")
     grid_small = np.load(
         f"pipelines/precipitation_model/impa/data/dataframe_grids/{location}-res=2km-256x256.npy"
     )
     grid_large = np.load(
         f"pipelines/precipitation_model/impa/data/dataframe_grids/{location}-res=4km-256x256.npy"
     )
+    log(f"grid_small.shape: {grid_small.shape}, grid_large.shape: {grid_large.shape}")
     assert grid_small.shape == grid_large.shape
     ni, nj = grid_small.shape[:2]
+
+    log("Start saving files as h5")
     with h5py.File(output_filepath, "w") as f:
         what = f.create_group("what")
         what.attrs["feature"] = product