Update CMEMS observation scripts

scripts/cmems_obs/fix_obs_tracks.py

@@ -0,0 +1,79 @@
+import argparse
+
+import numpy as np
+import pandas as pd
+from sqlalchemy import create_engine
+from sqlalchemy.dialects.mysql import insert
+
+
+def mysql_upsert(table, conn, keys, data_iter):
+    data = [dict(zip(keys, row)) for row in data_iter]
+    stmt = insert(table.table).values(data)
+    update_stmt = stmt.on_duplicate_key_update(**{c.name: c for c in stmt.inserted})
+    conn.execute(update_stmt)
+
+
+def main(uri: str, platform_ids: str | list | None = None):
+    """
+    Corrects station coordinates at crossing of date line so that tracks are displayed
+    continuously in the Navigator. Arguments should be passed on commandline. e.g:
+    python scripts/data_importers/get_cmems_month.py "uri" -p 12345 45678
+
+    :param uri: The URI string of the MariaDB Observation database
+    :param platform_ids: IDs of platforms to be corrected. If not provided all
+                        platforms will be updated.
+    """
+
+    engine = create_engine(uri)
+
+    if isinstance(platform_ids, str):
+        platform_ids = [platform_ids]
+    elif platform_ids is None:
+        platforms = pd.read_sql("SELECT * FROM platforms;", con=engine)
+        platform_ids = platforms.id.values
+
+    for platform_id in platform_ids:
+        stations = pd.read_sql(
+            f"SELECT * FROM stations where platform_id={platform_id};",
+            con=engine,
+            index_col="id",
+        ).sort_values(by="time")
+
+        corrected_lon = np.copy(stations.longitude.values)
+        diffs = np.diff(corrected_lon)
+        crossings = np.where(np.abs(diffs) > 180)[0]
+
+        if len(crossings) > 0:
+            print(f"Updating track of platform {platform_id}.")
+
+            for crossing in crossings:
+                if diffs[crossing] > 0:
+                    corrected_lon[crossing + 1 :] -= 360
+                else:
+                    corrected_lon[crossing + 1 :] += 360
+            stations.longitude = corrected_lon
+            stations.to_sql(
+                "stations", engine, if_exists="append", method=mysql_upsert, index=False
+            )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="Initialize/append CMEMS Observation data",
+        description=(
+            "Add monthly CMEMS Global Ocean In-Situ Near-Real-Time "
+            "Observations data to ONAV Obs database."
+        ),
+    )
+    parser.add_argument("uri", type=str, help="URI of the observation database.")
+    parser.add_argument(
+        "-p",
+        "--platform_ids",
+        type=str,
+        help="ID of platforms to modify.",
+        default=None,
+    )
+
+    args = parser.parse_args()
+
+    main(args.uri, args.platform_ids)

scripts/cmems_obs/get_cmems_obs_day.py

@@ -0,0 +1,88 @@
+import argparse
+import os
+from datetime import date, timedelta
+
+import copernicusmarine
+
+from import_cmems_obs import import_cmems_obs_mp
+
+
+def main(uri: str, output_dir: str, date_str: str):
+    """
+    Downloads data from CMEMS latest observation dataset matching the provided date
+    string to add to the observation database.
+    Deletes dowloaded files after data is added. Arguments should be passed on
+    commandline. e.g:
+    python scripts/data_importers/init_cmems_obs.py "uri" "/output_dir/" -d 20250101
+
+    :param uri: The URI string of the MariaDB Observation database
+    :param output_dir: The directory that you want to save the data to.
+    :param date_str: The date in YYYYMMDD format.
+    """
+
+    prod_id = "INSITU_GLO_PHYBGCWAV_DISCRETE_MYNRT_013_030"
+    meta = copernicusmarine.describe(product_id=prod_id)
+    version = [v.label for v in meta.products[0].datasets[0].versions][-1]
+
+    obs_types = ["GL", "DB", "PF", "CT"]
+
+    for obs in obs_types:
+        print(obs)
+        obs_filter = f"*/{date_str}/*_*_{obs}_*_*.nc"
+        resp = copernicusmarine.get(
+            dataset_id="cmems_obs-ins_glo_phybgcwav_mynrt_na_irr",
+            dataset_version=version,
+            output_directory=output_dir,
+            filter=obs_filter,
+            dataset_part="latest",
+            sync=True,
+        )
+        file_list = [f.file_path for f in resp.files]
+
+        if len(file_list) == 0:
+            continue
+
+        try:
+            import_cmems_obs_mp(uri, file_list, obs)
+
+            for file in file_list:
+                os.remove(file)
+        except Exception as e:
+            print(e)
+            with open("obs_error.log", "a") as f:
+                f.write(f"{obs_filter}\n")
+                f.write("\n")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="Initialize/append CMEMS Observation data",
+        description=(
+            "Add monthly CMEMS Global Ocean In-Situ Near-Real-Time "
+            "Observations data to ONAV Obs database."
+        ),
+    )
+    parser.add_argument("uri", type=str, help="URI of the observation database.")
+    parser.add_argument(
+        "output_dir",
+        type=str,
+        help="Output directory for observation data.",
+    )
+    parser.add_argument(
+        "-d",
+        "--date",
+        type=str,
+        help=(
+            "The date to add in YYYYMMDD format. If omitted then yesterday's date "
+            "will be used."
+        ),
+        default=None,
+    )
+
+    args = parser.parse_args()
+
+    if args.date:
+        main(args.uri, args.output_dir, args.date)
+    else:
+        date_str = (date.today() - timedelta(days=1)).strftime("%Y%m%d")
+        main(args.uri, args.output_dir, date_str)

scripts/cmems_obs/get_cmems_obs_month.py

@@ -0,0 +1,90 @@
+import argparse
+import os
+from datetime import date, datetime, timedelta
+
+import copernicusmarine
+
+from import_cmems_obs import import_cmems_obs_mp
+
+
+def main(uri: str, output_dir: str, month_key: str):
+    """
+    Downloads one month of CMEMS monthly observation data to append to the observation
+    database. Deletes dowloaded files after data is added. Arguments should be passed
+    on commandline. e.g:
+    python scripts/data_importers/get_cmems_month.py "uri" "/output_dir/" -m "month_key"
+
+    :param uri: The URI string of the MariaDB Observation database
+    :param output_dir: The directory that you want to save the data to.
+    :param month_key: The month in YYYYMM format
+    """
+
+    prod_id = "INSITU_GLO_PHYBGCWAV_DISCRETE_MYNRT_013_030"
+    meta = copernicusmarine.describe(product_id=prod_id)
+    version = [v.label for v in meta.products[0].datasets[0].versions][-1]
+
+    obs_types = ["GL", "DB", "PF", "CT"]
+
+    for obs in obs_types:
+        obs_filter = f"*/{obs}/{month_key}/*.nc"
+        resp = copernicusmarine.get(
+            dataset_id="cmems_obs-ins_glo_phybgcwav_mynrt_na_irr",
+            dataset_version=version,
+            output_directory=output_dir,
+            filter=obs_filter,
+            dataset_part="monthly",
+            sync=True
+        )
+
+        file_list = [f.file_path for f in resp.files]
+
+        try:
+            import_cmems_obs_mp(uri, file_list, obs)
+
+            for file in file_list:
+                os.remove(file)
+        except Exception as e:
+            print(e)
+            with open("obs_error.log", "a") as f:
+                f.write(f"{obs_filter}\n")
+                f.write("\n")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="Initialize/append CMEMS Observation data",
+        description=(
+            "Add monthly CMEMS Global Ocean In-Situ Near-Real-Time "
+            "Observations data to ONAV Obs database."
+        ),
+    )
+    parser.add_argument("uri", type=str, help="URI of the observation database.")
+    parser.add_argument(
+        "output_dir",
+        type=str,
+        help="Output directory for observation data.",
+    )
+    parser.add_argument(
+        "-m",
+        "--month",
+        type=str,
+        help=(
+            "The month to add in YYYYMM format. If omitted then all months from 2017 "
+            "onward will be downloaded."
+        ),
+        default=None,
+    )
+
+    args = parser.parse_args()
+
+    if args.month:
+        main(args.uri, args.output_dir, args.month)
+    else:
+        start_month = date.today().replace(day=1)
+        end_month = datetime(2017, 1, 1)
+        while end_month >= end_month:
+            month_key = start_month.strftime("%Y%m")
+
+            main(args.uri, args.output_dir, month_key)
+
+            start_month = (start_month - timedelta(days=28)).replace(day=1)