0ptr/nullptr/analyzer.py

from nullptr.models.marketplace import Marketplace
from nullptr.models.jumpgate import Jumpgate
from nullptr.models.system import System
from nullptr.models.waypoint import Waypoint
from dataclasses import dataclass
from nullptr.util import pprint
from copy import copy

class AnalyzerException(Exception):
  pass

def path_dist(m):
  t = 0
  o = Point(0,0)
  for w in m:
    t +=w.distance(o)
    o = w
  return t
  
@dataclass
class Point:
  x: int
  y: int
  
@dataclass
class TradeOption:
  resource: str
  source: Waypoint
  dest: Waypoint
  buy: int
  margin: int
  dist: int
  score: float
  
@dataclass
class SearchNode:
  system: System
  parent: 'SearchNode'

  def __hash__(self):
    return hash(self.system.symbol)
    
  def path(self):
    result = []
    n = self
    while n is not None:
      result.append(n.system)
      n = n.parent
    result.reverse()
    return result
    
  def __repr__(self):
    return self.system.symbol
 
    
def find_markets(c, resource, sellbuy):
  for m in c.store.all(Marketplace):
    if 'sell' in sellbuy and resource in m.imports:
      yield ('sell', m)
    
    elif 'buy' in sellbuy and resource in m.exports:
      yield ('buy', m)
    
    elif 'exchange' in sellbuy and resource in m.exchange:
      yield ('exchange', m)
      
def find_closest_markets(c, resource, sellbuy, location):
  if type(location) == str:
    location = c.store.get(Waypoint, location)
  mkts = find_markets(resource, sellbuy)
  candidates = []
  origin = location.system
  for typ, m in mkts:
    system = m.waypoint.system
    d = origin.distance(system)
    candidates.append((typ, m, d))
  possibles = sorted(candidates, key=lambda m: m[2])
  possibles = possibles[:10]
  results = []
  for typ,m,d in possibles:
    system = m.waypoint.system
    p = find_jump_path(origin, system)
    if p is None: continue
    results.append((typ,m,d,len(p)))
  return results
  
def solve_tsp(c, waypoints):
  wps = copy(waypoints)
  path = []
  cur = Point(0,0)
  while len(wps) > 0:
    closest = wps[0]
    for w in wps:
      if w.distance(cur) < closest.distance(cur):
        closest = w
    cur = closest
    path.append(closest)
    wps.remove(closest)
  return path

def get_jumpgate(c, system):
  gates = c.store.all_members(system, Jumpgate)
  return next(gates, None)

# dijkstra shmijkstra
def find_nav_path(c, orig, to, ran):
  path = []
  mkts = [m.waypoint for m in c.store.all_members(orig.system, Marketplace)]
  cur = orig
  if orig == to:
    
    return []
  while cur != to:
    best = cur
    bestdist = cur.distance(to)
    if bestdist < ran:
      path.append(to)
      break
    for m in mkts:
      dist = m.distance(to)
      if dist < bestdist and cur.distance(m) < ran:
        best = m
        bestdist = dist
    if best == cur:
      raise AnalyzerException(f'no path to {to}') 
    cur = best
    path.append(cur)
  return path
  
def find_jump_path(c, orig, to, depth=100, seen=None):
  if depth < 1: return None
  if seen is None:
    seen = set()
  if type(orig) == System:
    orig = set([SearchNode(orig,None)])
  result = [n for n in orig if n==to]
  if len(result) > 0:
    return result[0].path()
  dest = set()
  for o in orig:
    jg = get_jumpgate(o)
    if jg is None: continue
    for s in jg.connections:
      if s in seen:  continue
      seen.add(s)
      dest.add(SearchNode(s, o))
  if len(dest) == 0: 
    return None
  return find_jump_path(dest, to, depth-1, seen)

def prices(c, system):
  prices = {}
  for m in c.store.all_members(system, Marketplace):
    for r, p in m.prices.items():
      if not r in prices:
        prices[r] = []
      prices[r].append({
        'wp': m.waypoint,
        'buy': p.buy,
        'sell': p.sell,
        'volume': p.volume,
        'category': m.rtype(r)
      })
  return prices

def find_trade(c, system):
  max_traders = 3
  pcs= prices(c, system)
  occupied_routes = dict()
  for s in c.store.all('Ship'):
    if s.mission != 'trade':
      continue
    k = (s.mission_state['site'], s.mission_state['dest'])
    if k in occupied_routes:
      occupied_routes[k] += 1
    else:
      occupied_routes[k] = 1
  best = None
  for resource, markets in pcs.items():
    source = sorted(markets, key=lambda x: x['buy'])[0]
    dest = sorted(markets, key=lambda x: x['sell'])[-1]
    swp = source['wp']
    dwp = dest['wp']
    margin = dest['sell'] -source['buy']
    k = (swp.symbol,dwp.symbol)
    if k in occupied_routes and occupied_routes[k] > max_traders:
      continue
    dist = swp.distance(dwp)
    dist = max(dist, 0.0001)
    score = margin / dist
    if margin < 2:
      continue
    o = TradeOption(resource, swp, dwp, source['buy'], margin, dist, score)
    if best is None or best.score < o.score:
      best = o
  return best

def find_deal(c, smkt, dmkt):
  best_margin = 0
  best_resource = None
  for r, sp in smkt.prices.items():
    if not r in dmkt.prices:
      continue
    dp = dmkt.prices[r]
    margin = dp.sell - sp.buy
    if margin > best_margin:
      best_margin = margin
      best_resource = r
  return best_resource

def best_sell_market(c, system, r):
  best_price = 0
  best_market = None
  for m in c.store.all_members(system, Marketplace):
    if r not in m.prices: continue
    price = m.prices[r].sell
    if price > best_price:
      best_price = price
      best_market = m
  return best_market
  
def find_gas(c, system):
  m = [w for w in c.store.all_members(system, 'Waypoint') if w.type == 'GAS_GIANT']
  if len(m)==0:
    raise AnalyzerException('no gas giant found')
  return m[0]
  
def find_metal(c, system):
  m = [w for w in c.store.all_members(system, Waypoint) if 'COMMON_METAL_DEPOSITS' in w.traits]
  if len(m) == 0:
    return None
  origin = Point(0,0)
  m = sorted(m, key=lambda w: w.distance(origin))
  return m[0]