from nullptr.command_line import CommandLine from nullptr.store import Store from nullptr.analyzer import Analyzer import argparse from nullptr.models.agent import Agent 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.api import Api from .util import * from time import sleep, time from threading import Thread from nullptr.atlas_builder import AtlasBuilder class CommandError(Exception): pass class Commander(CommandLine): def __init__(self, store_dir='data'): self.store_dir = store_dir self.store = Store(store_dir) self.store.load() self.agent = self.select_agent() self.api = Api(self.store, self.agent) self.atlas_builder = AtlasBuilder(self.store, self.api) self.analyzer = Analyzer(self.store) self.ship = None self.stop_auto= False super().__init__() def prompt(self): if self.ship: return f'{self.ship.symbol}> ' else: return '> ' def has_ship(self): if self.ship is not None: return True else: print('set a ship') def ask_obj(self, typ, prompt): obj = None while obj is None: symbol = input(prompt) obj = self.store.get(typ, symbol.upper()) if obj is None: print('not found') return obj def select_agent(self): agents = self.store.all(Agent) agent = next(agents, None) if agent is None: symbol = input('agent name: ') agent = self.store.get(Agent, symbol, create=True) return agent def resolve(self, typ, arg): arg = arg.upper() matches = [c for c in self.store.all(typ) if c.symbol.startswith(arg)] if len(matches) == 1: return matches[0] elif len(matches) > 1: raise CommandError('multiple matches') else: raise CommandError('not found') def after_cmd(self): self.store.flush() def do_info(self, arg=''): if arg.startswith('r'): self.api.info() pprint(self.agent, 100) def do_register(self, faction): self.api.register(faction.upper()) def do_universe(self, page=1): self.atlas_builder.run(page) def do_systems(self, page=1): r = self.api.list_systems(int(page)) pprint(self.api.last_meta) def do_waypoints(self, system_str=''): if system_str == '': if not self.has_ship(): return system = self.ship.location().system() else: system = self.store.get(System, system_str) r = self.store.all_members(system, 'Waypoint') for w in r: traits = [] if 'MARKETPLACE' in w.traits: traits.append('MARKET') if 'SHIPYARD' in w.traits: traits.append('SHIPYARD') if w.type == 'JUMP_GATE': traits.append('JUMP') if w.type == 'ASTEROID_FIELD': traits.append('ASTROIDS') print(w.symbol.split('-')[2], ', '.join(traits)) def do_marketplace(self, waypoint_str): waypoint = self.store.get(Waypoint, waypoint_str.upper()) r = self.api.marketplace(waypoint) def do_jumps(self, waypoint_str=None): if waypoint_str is None: if not self.has_ship(): return waypoint = self.ship.location() else: waypoint = self.store.get(Waypoint, waypoint_str.upper()) r = self.api.jumps(waypoint) pprint(r) def do_query(self): location = self.ask_obj(System, 'Where are you? ') resource = input('what resource?').upper() sellbuy = self.ask_multichoice(['sell','buy'], 'do you want to sell or buy?') print('Found markets:') for m in self.analyzer.find_markets(resource, sellbuy): system = self.store.get(System, m.system()) p = self.analyzer.find_path(location, system) if p is None: continue print(m, f'{len(p)-1} hops') def do_path(self): orig = self.ask_obj(System, 'from: ') dest = self.ask_obj(System, 'to: ') # orig = self.store.get(System, 'X1-KS52') # dest = self.store.get(System, 'X1-DA90') path = self.analyzer.find_path(orig, dest) pprint(path) def do_ships(self, arg=''): if arg.startswith('r'): r = self.api.list_ships() else: r = list(self.store.all('Ship')) pprint(r) def do_contracts(self, arg=''): if arg.startswith('r'): r = self.api.list_contracts() else: r = list(self.store.all('Contract')) pprint(r) def do_ship(self, arg=''): if arg != '': symbol = f'{self.agent.symbol}-{arg}' ship = self.store.get('Ship', symbol) if ship is None: print('not found') return else: self.ship = ship pprint(self.ship) def do_pp(self): pprint(self.api.last_result) def do_go(self, arg): if not self.has_ship(): return system = self.ship.location().system() symbol = f'{system}-{arg}' dest = self.resolve('Waypoint', symbol) self.api.navigate(self.ship, dest) pprint(self.ship) def do_dock(self): if not self.has_ship(): return self.api.dock(self.ship) pprint(self.ship) def do_orbit(self): if not self.has_ship(): return self.api.orbit(self.ship) pprint(self.ship) def do_negotiate(self): if not self.has_ship(): return r = self.api.negotiate(self.ship) pprint(r) def do_refuel(self): if not self.has_ship(): return r = self.api.refuel(self.ship) pprint(self.ship) def do_accept(self, c): contract = self.resolve('Contract', c) r = self.api.accept_contract(contract) pprint(r) def do_market(self, arg=''): if arg == '': if not self.has_ship(): return waypoint = self.ship.location() else: waypoint = self.resolve('Waypoint', arg) r = self.api.marketplace(waypoint) pprint(r) def do_cargo(self): if not self.has_ship(): return for c, units in self.ship.cargo.items(): print(f'{units:4d} {c}') def do_buy(self, resource, amt=None): if not self.has_ship(): return if amt is None: amt = self.ship.cargo_capacity - self.ship.cargo_units self.api.buy(self.ship, resource.upper(), amt) self.do_cargo() def do_sell(self, resource): if not self.has_ship(): return self.api.sell(self.ship, resource.upper()) self.do_cargo() def do_dump(self, resource): if not self.has_ship(): return self.api.jettison(self.ship, resource.upper()) self.do_cargo() def do_shipyard(self): if not self.has_ship(): return location = self.ship.location() pprint(self.api.shipyard(location)) def do_jump(self, system_str): if not self.has_ship(): return system = self.resolve('System', system_str) self.api.jump(self.ship, system) pprint(self.ship) def do_purchase(self, ship_type): if not self.has_ship(): return location = self.ship.location() ship_type = ship_type.upper() if not ship_type.startswith('SHIP'): ship_type = 'SHIP_' + ship_type s = self.api.purchase(ship_type, location) pprint(s) def do_survey(self): if not self.has_ship(): return r = self.api.survey(self.ship) pprint(r)