laurent.deleers@gmail.com
9b5fcd0100
update interface adapt time base before converting to grib refactor some stuff remove debug comments
108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
import os
|
|
from datetime import datetime, timedelta
|
|
from copernicusmarine import subset
|
|
import subprocess
|
|
|
|
class CopernicusDownloader:
|
|
def __init__(self, username="", password=""):
|
|
self.dataset_id = "cmems_mod_ibi_phy_anfc_0.027deg-2D_PT1H-m"
|
|
self.username = username
|
|
self.password = password
|
|
self.output_file = os.path.join(os.getcwd(), "downloads", "download.nc")
|
|
self.prepared_file = os.path.join(os.getcwd(), "downloads", "download_prepared.nc")
|
|
self.output_grib = os.path.join(os.getcwd(), "downloads", "download.grib2")
|
|
|
|
self.lat_min = 0.0
|
|
self.lat_max = 0.0
|
|
self.lon_min = 0.0
|
|
self.lon_max = 0.0
|
|
self.days = 1
|
|
|
|
def retrieve_grib2(self, lat_min, lat_max, lon_min, lon_max, days):
|
|
self.lat_min = lat_min
|
|
self.lat_max = lat_max
|
|
self.lon_min = lon_min
|
|
self.lon_max = lon_max
|
|
self.days = days
|
|
# print(f"Démarage du téléchargement avec {lat_min} - {lat_max} - {lon_min} - {lon_max} - {days} ")
|
|
|
|
self.download(self.output_file)
|
|
self.apply_setreftime(self.output_file,self.prepared_file)
|
|
self.convert_to_grib2(self.prepared_file,self.output_grib)
|
|
return self.output_grib
|
|
|
|
def download(self,outfile):
|
|
today = datetime.utcnow()
|
|
start_date = today.strftime("%Y-%m-%dT00:00:00")
|
|
end_date = (today + timedelta(days=self.days)).strftime("%Y-%m-%dT23:00:00")
|
|
|
|
save_dir = os.path.join(os.getcwd(), "downloads")
|
|
os.makedirs(save_dir, exist_ok=True)
|
|
|
|
subset(
|
|
dataset_id=self.dataset_id,
|
|
output_filename=outfile,
|
|
variables=["uo", "vo"], # removed "thetao" (temperature) and "zos" (elevation)
|
|
minimum_longitude=self.lon_min,
|
|
maximum_longitude=self.lon_max,
|
|
minimum_latitude=self.lat_min,
|
|
maximum_latitude=self.lat_max,
|
|
start_datetime=start_date,
|
|
end_datetime=end_date,
|
|
username=self.username,
|
|
password=self.password
|
|
)
|
|
print(f"✅ File downloaded to : {outfile}")
|
|
|
|
|
|
def extract_date_and_time(self,infile):
|
|
# Extract the first timecode of the nc file
|
|
try:
|
|
result = subprocess.run(
|
|
["cdo", "showtimestamp", infile],
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True
|
|
)
|
|
|
|
# Récupérer le premier timestamp : ex. 2025-05-21T00:00:00
|
|
first_timestamp = result.stdout.strip().split()[0]
|
|
date, time = first_timestamp.split('T')
|
|
return date, time
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"[CDO Error] {e.stderr}")
|
|
except Exception as e:
|
|
print(f"[Error] {e}")
|
|
|
|
return None, None
|
|
|
|
def apply_setreftime(self,infile,outfile):
|
|
# Apply the command cdo -setreftime,<date>,<heure> <input> <output> to define a reference time
|
|
date, time = self.extract_date_and_time(infile)
|
|
if not date or not time:
|
|
print("Cannot retrieve date/time from this file.")
|
|
return
|
|
|
|
try:
|
|
subprocess.run(
|
|
["cdo", f"-setreftime,{date},{time}", infile, outfile],
|
|
check=True
|
|
)
|
|
print(f"✅ File adapted with correct time ref {date} - {time} : {outfile}")
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"[Error when setting reference time with CDO -setreftime] {e.stderr}")
|
|
|
|
|
|
def convert_to_grib2(self,infile,outfile):
|
|
if not os.path.exists(infile):
|
|
raise FileNotFoundError(f"File not found : {infile}")
|
|
try:
|
|
subprocess.run(["cdo", "-f", "grb2", "copy", infile, outfile], check=True)
|
|
print(f"✅ File converted to grib : {outfile}")
|
|
return outfile
|
|
except subprocess.CalledProcessError as e:
|
|
raise RuntimeError(f"Error when converting with CDO : {e}")
|