tasks.py

import os
import redis
import pandas as pd
import numpy as np
import json
from celery import Celery
from sdig.erddap.info import Info
import urllib.parse
import constants
import db
from celery.utils.log import get_task_logger
import ssl
import urllib.parse
import plotly.express as px

import time

import requests
import time
import io

timeout = 1800

ssl._create_default_https_context = ssl._create_unverified_context
logger = get_task_logger(__name__)


def flush():
    constants.redis_instance.flushall()


def update_mission(mid, mission):
   
    logger.debug('Pulling locations for mission ' + str(mission['ui']['title']))
         
    start_dates = []
    end_dates = []
    long_names = {}
    units = {}
    dsg_ids = []
    drones = mission['drones']
    mission_dfs = []
    for dix, d in enumerate(sorted(drones)):
        color = px.colors.qualitative.Alphabet[dix]
        logger.debug('Reading drone ' + str(d))
        drone = drones[d]
        info_url = drone['url']
        info = Info(info_url) 
        depth_name, dsg_var = info.get_dsg_info()
        drone_vars, d_long_names, d_units, standard_names, var_types = info.get_variables()
        dsg_id = dsg_var[info.get_dsg_type()]
        base_url = drone['url'] + '.csv?'
        time_url = base_url + urllib.parse.quote_plus('time&orderByMinMax("time")')
        print("time range:", time_url)
        tdf = read_csv_with_retries_requests(time_url, retries=10, delay=60)
        # tdf = pd.read_csv(time_url, skiprows=[1])
        start_date = tdf['time'].min()
        end_date = tdf['time'].max()
        extra_var = 'SOG'
        for extra_var in drone_vars:
            if extra_var != 'latitude' and extra_var != 'longitude' and extra_var != 'time' and extra_var != dsg_id:
                break
        req_vars = f'latitude,longitude,time,{dsg_id},{extra_var}'
        query = '&orderByClosest("time,1day")&'+dsg_id+'="'+d+'"'
        q = urllib.parse.quote(query)
        url = base_url + req_vars + q
        print("Locations:", url)
        df = read_csv_with_retries_requests(url, retries=10, delay=60)
        df.drop(extra_var, axis=1, inplace=True)
        # df = pd.read_csv(url, skiprows=[1])
        # Don't drop, just take the rows where lat or lon is not NA:
        df = df[df['latitude'].notna()]
        df = df[df['longitude'].notna()]
        df['mission_id'] = mid
        df['title'] = mission['ui']['title']
        df[dsg_id] = df[dsg_id].astype(str)
        drones[d]['variables'] = drone_vars
        drones[d]['start_date'] = start_date
        drones[d]['color'] = color
        start_dates.append(start_date)
        drones[d]['end_date'] = end_date
        end_dates.append(end_date)
        depth_name, dsg_id = info.get_dsg_info()
        dsg_ids.append(dsg_id['trajectory'])
        long_names = {**long_names, **d_long_names}
        units = {**units, **d_units}
        mission_dfs.append(df)
    uids = list(set(dsg_ids))
    if len(uids) == 1:
        mission['dsg_id'] = uids[0]
    else:
        print('Mission has non-unique DSG ID names.')
    long_names = dict(sorted(long_names.items(), key=lambda item: item[1]))
    mission['long_names'] = long_names
    mission['units'] = units
    start_dates.sort()
    end_dates.sort()
    mission['start_date'] = start_dates[0]
    mission['end_date'] = end_dates[-1]
    constants.redis_instance.hset("mission", mid, json.dumps(mission)) 
    full_df = pd.concat(mission_dfs).reset_index()
    return full_df

# Run this once from the workspace before deploying the application

def load_missions(force=False):
    with open('config/missions.json') as missions_config:
        config_json = json.load(missions_config)
    collections = config_json['collections']
    outeridx = 0 
    for collection in collections:
        logger.info('Processing missions for ' + collection)
        member = collections[collection]     
        for idx, mid in enumerate(member['missions']):
            mission = member['missions'][mid]
            mission_exists_df = db.get_mission_locations_notsorted(mid)
            print(mission_exists_df['mission_id'].unique())
            if mission_exists_df.empty or force or mission['active'] == 'true':
                df = update_mission(mid, mission)
                print(f'{df.shape[0]} records found for {mid}')
                if not df.empty:
                    if outeridx == 0 and force:
                        df.to_sql(constants.locations_table, constants.postgres_engine, if_exists='replace', index=False)
                    else:
                        print(f'deleting previous records for {mid}')
                        db.delete_ds_drones(mid)
                        print(f'Saving {df.shape[0]} new records for {mid}')
                        df.rename(columns={mission['dsg_id']: 'trajectory'}, inplace=True)
                        df.to_sql(constants.locations_table, constants.postgres_engine, if_exists='append', index=False)
                    outeridx = outeridx + 1
                
    logger.info('Setting the mission locations...')


def read_csv_with_retries_requests(url, retries=10, delay=5):
    """
    Downloads a CSV file from a URL using the requests library, and converts
    it to a pandas DataFrame with retry logic.

    Args:
        url (str): The URL of the CSV file.
        retries (int): The number of times to retry the download. Defaults to 10.
        delay (int): The delay in seconds between retries. Defaults to 5.

    Returns:
        pandas.DataFrame: The DataFrame containing the CSV data, or None if the download fails.
    """
    for i in range(retries):
        try:
            response = requests.get(url, timeout=timeout)
            response.raise_for_status()  # Raise an HTTPError for bad responses (4xx or 5xx)

            # Use io.StringIO to treat the string as a file
            s = io.StringIO(response.text)
            df = pd.read_csv(s, skiprows=[1])
            print("Download successful! 🎉")
            return df
        except (requests.exceptions.ConnectTimeout, requests.exceptions.ReadTimeout, requests.exceptions.Timeout, requests.exceptions.RequestException, OSError, TimeoutError) as rexe:
            print(f"Attempt {i+1} failed: {rexe}")
            if i < retries - 1:
                print(f"Retrying in {delay} seconds...")
                time.sleep(delay)
            else:
                print("All retry attempts failed. 😔")
                return None
        except Exception as e: # Catch any other unexpected errors
            print(f"An unexpected error occurred: {e}")
            if i < retries - 1:
                print(f"Retrying in {delay} seconds...")
                time.sleep(delay)
            else:
                print("All retry attempts failed. 😔")
                return None