0ptr/nullptr/api.py

import requests
from nullptr.models.system import System
from nullptr.models.waypoint import Waypoint
from nullptr.models.marketplace import Marketplace
from nullptr.models.jumpgate import Jumpgate
from nullptr.models.ship import Ship
from nullptr.models.shipyard import Shipyard
from .util import *
from time import sleep, time

class ApiError(AppError):
  def __init__(self, msg, code):
    super().__init__(msg)
    self.code = code
  
class ApiLimitError(Exception):
  pass
  
class Api:
  def __init__(self, c, agent):
    self.agent = agent
    self.store = c.store
    self.requests_sent = 0
    self.last_meta = None
    self.last_result = None
    self.root = 'https://api.spacetraders.io/v2/'
    
  def token(self):
    if self.agent.token is None:
      raise ApiError('no token. Please register', 1337)
    return self.agent.token
  
  def request(self, method, path, data=None, need_token=True, params={}):
    try:
      start = time()
      result = self.request_once(method, path, data, need_token, params)
      dur = time() - start
      # print(f'api {dur:.03}')
      return result
    except (ApiLimitError, requests.exceptions.Timeout):
      # print('oops, hit the limit. take a break')
      sleep(10)
      return self.request_once(method, path, data, need_token, params)
      
  def request_once(self, method, path, data=None, need_token=True, params={}):
    headers = {}
    if need_token:
      headers['Authorization'] = 'Bearer ' + self.token()
    self.requests_sent += 1
    if method == 'get':
      params['limit'] = 20
    r = requests.request(method, self.root+path, json=data, headers=headers, params=params)
    if r.status_code == 429:
      raise ApiLimitError()
    result = r.json()
    self.last_result = result
    if result is None:
      raise ApiError('http call failed', r.status_code)
    if 'meta' in result:
      self.last_meta = result['meta']
    if 'data' not in result:
      error_code = sg(result, 'error.code', -1)
      self.last_error = error_code
      error_message = sg(result, 'error.message')
      raise ApiError(error_message, error_code)
    else:
      self.last_error = 0
      return result['data']
  
  ######## Account #########
  def register(self, faction):
    callsign = self.agent.symbol
    data = {
      'symbol': callsign,
      'faction': faction
    }
    result = self.request('post', 'register', data, need_token=False)
    token = mg(result, 'token')
    self.agent.update(mg(result, 'agent'))
    self.agent.token = token
  
  def status(self):
    try:
      self.request('get', '')
    except ApiError:
      pass
    return self.last_result
    
  def info(self):
    data = self.request('get', 'my/agent')
    self.agent.update(data)
    return self.agent
  
  ######## Atlas #########
  def list_systems(self, page=1):
    data = self.request('get', 'systems', params={'page': page})
    #pprint(self.last_meta)
    systems = self.store.update_list(System, data)
    for s in data:
      self.store.update_list(Waypoint, mg(s, 'waypoints'))
    return systems
  
  def list_waypoints(self, system):
    data = self.request('get', f'systems/{system}/waypoints/')
    tp = total_pages(self.last_meta)
    for p in range(tp):
      data += self.request('get', f'systems/{system}/waypoints/', params={'page': p+1})
    # pprint(data)
    return self.store.update_list(Waypoint, data)
  
  def marketplace(self, waypoint):
    system = waypoint.system
    data = self.request('get', f'systems/{system}/waypoints/{waypoint}/market')
    return self.store.update(Marketplace, data)
  
  def jumps(self, waypoint):
    data = self.request('get', f'systems/{waypoint.system}/waypoints/{waypoint}/jump-gate')
    symbol = str(waypoint)
    return self.store.update(Jumpgate, data, symbol)
  
  def shipyard(self, wp):
    data = self.request('get', f'systems/{wp.system}/waypoints/{wp}/shipyard')
    symbol = str(wp)
    
    return self.store.update(Shipyard, data, symbol)

  ######## Fleet #########
  def list_ships(self):
    data = self.request('get', 'my/ships')
    tp = total_pages(self.last_meta)
    for p in range(1, tp):
      data += self.request('get', 'my/ships', params={'page': p+1})
    return self.store.update_list(Ship, data)
  
  def refuel(self, ship, from_cargo=False):
    fuel_need = ship.fuel_capacity - ship.fuel_current
    fuel_avail = ship.get_cargo('FUEL') * 100
    units = fuel_need
    if from_cargo:
      units = min(units, fuel_avail)
    data = {'fromCargo': from_cargo, 'units': units }
    data =  self.request('post', f'my/ships/{ship}/refuel', data)
    self.log_transaction(data)
    if from_cargo:
      boxes = ceil(float(units) / 100)
      ship.take_cargo('FUEL', boxes)
    if 'fuel' in data:
      ship.update(data)
    if 'agent' in data:
      self.agent.update(data['agent'])
    return data
    
  ######## Contract #########
  def list_contracts(self):
    data = self.request('get', 'my/contracts')
    return self.store.update_list('Contract', data)
  
  def negotiate(self, ship):
    data = self.request('post', f'my/ships/{ship}/negotiate/contract')
    if data is not None and 'contract' in data:
      contract = self.store.update('Contract', data['contract'])
      return contract
      
  def accept_contract(self, contract):
    data = self.request('post', f'my/contracts/{contract.symbol.lower()}/accept')
    if 'contract' in data:
      contract.update(data['contract'])
    if 'agent' in data:
      self.agent.update(data['agent'])
    return contract
    
  def deliver(self, ship, typ, contract):
    units = ship.get_cargo(typ)
    if units == 0:
      print("Resource not in cargo")
      return {}
    data = {
      'shipSymbol': str(ship),
      'tradeSymbol': typ.upper(),
      'units': units
    }
    data = self.request('post', f'my/contracts/{contract.symbol.lower()}/deliver', data)
    if 'cargo' in data:
      ship.update(data)
    if 'contract' in data:
      contract.update(data['contract'])
    return contract
  
  def fulfill(self, contract):
    data = self.request('post', f'my/contracts/{contract.symbol.lower()}/fulfill')
    if 'contract' in data:
      contract.update(data['contract'])
    if 'agent' in data:
      self.agent.update(data['agent'])
    return contract
  
  ######## Nav #########
  def navigate(self, ship, wp):
    data = {'waypointSymbol': str(wp)}
    response = self.request('post', f'my/ships/{ship}/navigate', data)
    ship.log(f'nav to {wp}')
    ship.update(response)
  
  def dock(self, ship):
    data = self.request('post', f'my/ships/{ship}/dock')
    ship.update(data)
    return data
    
  def orbit(self, ship):
    data = self.request('post', f'my/ships/{ship}/orbit')
    ship.update(data)
    return data
  
  def flight_mode(self, ship, mode):
    data = {'flightMode': mode}
    data = self.request('patch', f'my/ships/{ship}/nav', data)
    ship.update({'nav':data})
    return data
    
  def jump(self, ship, waypoint):
    if type(waypoint) == Waypoint:
      waypoint = waypoint.symbol
    data = {
      "waypointSymbol": waypoint
    }
    data = self.request('post', f'my/ships/{ship}/jump', data)
    if 'nav' in data:
      ship.update(data)
    return ship
    
  ######## Extraction #########
  def siphon(self, ship):
    data = self.request('post', f'my/ships/{ship}/siphon')
    ship.update(data)
    amt = mg(data, 'siphon.yield.units')
    rec = mg(data, 'siphon.yield.symbol')
    ship.log(f"siphoned {amt} {rec}")
    ship.location.extracted += amt
    return data['siphon']
  
  def extract(self, ship, survey=None):
    data = {}
    url = f'my/ships/{ship}/extract'
    if survey is not None:
      data= survey.api_dict()
      url += '/survey'
    try:
      data =  self.request('post', url, data=data)
    except ApiError as e:
      if e.code in [ 4221, 4224]:
        survey.exhausted = True
      else:
        raise e
    ship.update(data)
    amt = sg(data, 'extraction.yield.units', 0)
    rec = sg(data, 'extraction.yield.symbol', 'nothing')
    ship.log(f"extracted {amt} {rec}")
    ship.location.extracted += amt
    return data
  
  def survey(self, ship):
    data =  self.request('post', f'my/ships/{ship}/survey')
    ship.update(data)
    result = self.store.update_list('Survey', mg(data, 'surveys'))
    return result
  
    
  ######## Commerce #########
  def transaction_cost(self, data):
    if not 'transaction' in data: return 0
    act = mg(data,'transaction.type')
    minus = -1 if act == 'PURCHASE' else 1
    units = mg(data, 'transaction.units')
    ppu = mg(data, 'transaction.pricePerUnit')
    return ppu * units * minus
    
  def log_transaction(self, data):
    if not 'transaction' in data: return
    typ = mg(data, 'transaction.tradeSymbol')
    ppu = mg(data, 'transaction.pricePerUnit')
    shipsym = mg(data,  'transaction.shipSymbol')
    ship = self.store.get('Ship', shipsym)
    units = mg(data, 'transaction.units')
    act = mg(data,'transaction.type')
    ship.log(f'{act} {units} of {typ} for {ppu} at {ship.location}')
    
  def sell(self, ship, typ,units=None):
    if units is None:
      units = ship.get_cargo(typ)
    data = {
      'symbol': typ,
      'units': units
    }
    data = self.request('post', f'my/ships/{ship}/sell', data)
    self.log_transaction(data)
    if 'cargo' in data:
      ship.update(data)
    if 'agent' in data:
      self.agent.update(data['agent'])
    return data
    
  def buy(self, ship, typ, amt):
    data = {
      'symbol': typ,
      'units': amt
    }
    data = self.request('post', f'my/ships/{ship}/purchase', data)
    self.log_transaction(data)
    if 'cargo' in data:
      ship.update(data)
    if 'agent' in data:
      self.agent.update(data['agent'])
    return data
  
  def jettison(self, ship, typ):
    units = ship.get_cargo(typ)
    if units == 0:
      print('cargo not found')
      return
    data = {
      'symbol': typ,
      'units': units
    }
    data = self.request('post', f'my/ships/{ship.symbol}/jettison', data)
    ship.log(f'drop {units} of {typ}')
    if 'cargo' in data:
      ship.update(data)
    if 'agent' in data:
      self.agent.update(data['agent'])
    return data
  
  def transfer(self, sship, dship, typ, amt):
    data = {
    'tradeSymbol': typ,
    'units': amt,
    'shipSymbol': dship.symbol
    }
    data = self.request('post', f'my/ships/{sship.symbol}/transfer', data)
    sship.log(f'tra {amt} {typ} to {dship}')
    dship.log(f'rec {amt} {typ} from {sship}', 10)
    if 'cargo' in data:
      sship.update(data)
    dship.put_cargo(typ, amt)
    
  def purchase(self, typ, wp):
    data = {
      'shipType': typ,
      'waypointSymbol': str(wp)
    }
    data = self.request('post', 'my/ships', data)
    if 'agent' in data:
      self.agent.update(data['agent'])
    if 'ship' in data:
      ship = self.store.update('Ship', data['ship'])  
      return ship