A class to retrieve France Oceanographic Fleet vessel positions

french_fleet.py

import pandas as pd
import urllib
import urllib.parse
import json

class FrenchFleet:
    """

    Examples
    --------
    >>> FrenchFleet('POURQUOIPAS').positions()
    >>> FrenchFleet(['THALASSA', 'POURQUOIPAS']).positions()
    >>> FrenchFleet().positions()
    
    >>> FrenchFleet('POURQUOIPAS').last_positions()
    >>> FrenchFleet(['THALASSA', 'POURQUOIPAS']).last_positions()
    >>> FrenchFleet().last_positions()
    
    """

    url = 'https://localisation.flotteoceanographique.fr/api/'
    fleet = ['ANTEA', 'CDLM', 'ATALANTE', 'EUROPE', 'MARION', 'POURQUOIPAS', 'TETHYS', 'THALASSA', 'TH']  # TH is for Thalia
    
    def __init__(self, names=None):
        if names is None:
            self.names = self.fleet
        elif isinstance(names, str) and names in self.fleet:            
            self.names = [names]
        elif isinstance(names, list):
            self.names = [n for n in names if n in self.fleet]
        if len(self.names) == 0:
            raise ValueError("None of the ship names matches the known fleet (%s)" % self.fleet)

    def fetch(self, uri):
        data = None
        with urllib.request.urlopen(uri) as url:
            data = json.load(url)
        return data

    def _convert_type(self, df:pd.DataFrame):
        if 'pos_DATE' in df.columns:
            df["pos_DATE"] = pd.to_datetime(df["pos_DATE"])   

        def str2num(s):
            # 1 deg = 60 minutes = 3600 seconds
            sgn = 1 if s[0] in ['E', 'N'] else -1
            deg = float(s[1:].split('°')[0].strip())
            ms = s[1:].split('°')[1].strip().split('.')
            mins = float(ms[0])
            secs = float(ms[1])
            return sgn*(deg + mins/60. + secs/3600.)

        if 'pos_LAT' in df.columns:
            df['pos_LAT'] = df['pos_LAT'].map(str2num)

        if 'pos_LONG' in df.columns:
            df['pos_LONG'] = df['pos_LONG'].map(str2num)
        
        return df
    
    def positions(self, startdate:pd.Timestamp=None, enddate:pd.Timestamp=None) -> pd.DataFrame:
        """Return Ship positions for a daterange as :class:`pd.DataFrame`

        Parameters
        ----------
        startdate: :class:`pd.Timestamp`, default=None
            If set to None, today's date is used.
        enddate: :class:`pd.Timestamp`, default=None
            If set to None, the startdate is used.
        
        Returns
        -------
        :class:`pd.DataFrame`
        """
        entry = 'positionsbydate'
        url = self.url + entry + "?"
        startdate = pd.to_datetime('now', utc=True) if startdate is None else startdate
        enddate = startdate if enddate is None else enddate

        results = []
        for ship in self.names:
            params = {'name': ship, 
                      'startdate': startdate.strftime("%y-%m-%dT12:00:00.000Z"), 
                      'enddate': enddate.strftime("%y-%m-%dT12:00:00.000Z")}
            uri = url + urllib.parse.urlencode(params)
            data = self.fetch(uri)
            if len(data)>0:
                df = pd.DataFrame(data)
                results.append(df)
        
        results = [r for r in results if r is not None]
        data = pd.concat(results)
        data = self._convert_type(data)
        data = data.sort_values(by='pos_DATE').reset_index().drop('index', axis='columns')
        return data

    def last_positions(self) -> pd.DataFrame:
        """Return Last ship positions as :class:`pd.DataFrame`

        Returns
        -------
        :class:`pd.DataFrame`
        """
        data = self.positions()
        return data.groupby('pos_NAV_ID').last().reset_index()