How to integrate vessel location and weather data using APIs?

On a previous post, I wrote a tutorial on how to create a web app of ETAs here http://www.batechniques.com/blog/how-to-create-a-simple-web-app-for-vessels-eta-api-in-python/

By using low code development platform anvi.works, and following the steps on user interface within client code, and server code in the mentioned post, we can integrate maritime data of vessel locations and weather data such wind and wave values. Also, the vessel’s location is displayed on a google map. We do this by using APIs from marinetraffic.com and Stormglass, and calling Google maps using anvil works (more information here https://anvil.works/docs/client/components/maps) .

Below, we display the code of the demo.

Client code in anvil.works

from ._anvil_designer import GeneralTemplate
from anvil import *
import anvil.tables as tables
import anvil.tables.query as q
from anvil.tables import app_tables
import anvil.server

class General(GeneralTemplate):
  def __init__(self, **properties):
    # Set Form properties and Data Bindings.
    self.init_components(**properties)
    
    self.label_2.visible = False
    
    self.repeating_panel_1.items = anvil.server.call('get_marine_api_records')
    self.repeating_panel_2.items = anvil.server.call('get_storm_api_records')


  def button_1_click(self, **event_args):
    """ This method is called when the button is clicked """
    ship_id = self.text_box_2.text
    lat, long = anvil.server.call('make_marine_request', ship_id=ship_id)
    
    if lat is None and long is None:
      self.label_2.visible = True
     
    else:
      self.label_2.visible = False
    
      self.map.center = GoogleMap.LatLng(lat, long)
      self.map.zoom = 10
    
      marker = GoogleMap.Marker(
        animation=GoogleMap.Animation.DROP,
        position=GoogleMap.LatLng(lat, long)
      )

      self.map.add_component(marker)
    
      self.repeating_panel_1.items = anvil.server.call('get_marine_api_records')
      self.repeating_panel_2.items = anvil.server.call('get_storm_api_records')

Server code in anvil.works

from anvil.tables import app_tables
import anvil.tables.query as q
import anvil.tables as tables
import anvil.server
import datetime
import requests
import json

@anvil.server.callable
def make_marine_request(ship_id):
  response = requests.get(f"https://services.marinetraffic.com/api/exportvessel/v:5/API key/shipid:{ship_id}/protocol:jsono/msgtype:simple").json()
  if not response:
    return None, None
  data_marine = response[0]
  
  lat = float(data_marine['LAT'])
  long = float(data_marine['LON'])
  
  url = f"https://stormglass.p.rapidapi.com/forecast?lat={lat}&lng={long}"

  headers = {
    'x-rapidapi-host': "stormglass.p.rapidapi.com",
    'x-rapidapi-key': "API key"
  }

  response = requests.request("GET", url, headers=headers)
  data_storm = response.json()
  
  app_tables.marine.add_row(
    SHIP_ID=ship_id,
    MMSI=data_marine['MMSI'],
    LAT=data_marine['LAT'],
    LONG=data_marine['LON'],
    SPEED=data_marine['SPEED'],
    HEADING=data_marine['HEADING'],
    COURSE=data_marine['COURSE'],
    CREATED=datetime.datetime.utcnow()
  )
  
  for hour in data_storm['hours']:
    wave_direction = ''
    wave_height = ''
    wind_speed = ''
    wind_direction = ''
    time = datetime.datetime.strptime(hour['time'][:19], "%Y-%m-%dT%H:%M:%S")
    
    for source in hour['waveDirection']:
      if source['source'] == 'sg':
        wave_direction = str(source['value'])
        break
        
    for source in hour['waveHeight']:
      if source['source'] == 'sg':
        wave_height = str(source['value'])
        break
        
    for source in hour['windDirection']:
      if source['source'] == 'sg':
        wind_direction = str(source['value'])
        break
        
    for source in hour['windSpeed']:
      if source['source'] == 'sg':
        wind_speed = str(source['value'])
        break
    
    app_tables.storm.add_row(
      SHIP_ID=ship_id,
      WAVE_DIRECTION=wave_direction,
      WAVE_HEIGHT=wave_height,
      WIND_DIRECTION=wind_direction,
      WIND_SPEED=wind_speed,
      TIME=time,
    )
  
  return lat, long


@anvil.server.callable
def get_marine_api_records():
  return app_tables.marine.search(tables.order_by("CREATED", ascending=False))

@anvil.server.callable
def get_storm_api_records():
  return app_tables.storm.search(tables.order_by("TIME"))