354 lines
11 KiB
Python
354 lines
11 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 nullptr.models.shipyard import Shipyard
|
|
from .util import *
|
|
from time import sleep, time
|
|
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:
|
|
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 |