0ptr/nullptr/api.py
2023-07-16 20:50:30 +02:00

260 lines
7.5 KiB
Python

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 = {}
if survey is not None:
data['survey'] = survey.api_dict()
try:
data = self.request('post', f'my/ships/{ship}/extract', 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