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 .util import * from time import sleep class ApiError(Exception): def __init__(self, msg, code): super().__init__(msg) self.code = code class ApiLimitError(Exception): pass class Api: def __init__(self, store, agent): self.agent = agent self.store = 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: return self.request_once(method, path, data, need_token, params) 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'] 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 info(self): data = self.request('get', 'my/agent') self.agent.update(data) return self.agent 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/') # 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 list_ships(self): data = self.request('get', 'my/ships') return self.store.update_list(Ship, data) 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 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 def navigate(self, ship, wp): data = {'waypointSymbol': str(wp)} response = self.request('post', f'my/ships/{ship}/navigate', data) 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 refuel(self, ship): data = self.request('post', f'my/ships/{ship}/refuel') if 'fuel' in data: ship.update(data) if 'agent' in data: self.agent.update(data['agent']) return data 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 sell(self, ship, typ): units = ship.get_cargo(typ) data = { 'symbol': typ, 'units': units } data = self.request('post', f'my/ships/{ship}/sell', 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) 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) if 'cargo' in data: ship.update(data) if 'agent' in data: self.agent.update(data['agent']) return data 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 def jump(self, ship, system): if type(system) == System: system = system.symbol data = { "systemSymbol": system } data = self.request('post', f'my/ships/{ship}/jump', data) if 'nav' in data: ship.update(data) return ship def shipyard(self, wp): return self.request('get', f'systems/{wp.system}/waypoints/{wp}/shipyard') def extract(self, ship, survey=None): data = {} url = f'my/ships/{ship}/extract' if survey is not None: data['survey'] = 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) 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