trade.py

#!/usr/bin/env python3

import code
import datetime
import enum
import logging
import pathlib
import readline
import rlcompleter
import time

import rich.logging
import rich.pretty
from rich import print

import spess.actor
import spess.client
import spess.models
from spess.models import *

# market: 'X1-VK73-B7'
#    0.0 X1-VK73-B7   ASTEROID_BASE  HOLLOWED_INTERIOR OUTPOST MARKETPLACE
#    4.5 X1-VK73-B10  ASTEROID       COMMON_METAL HOLLOWED_INTERIOR SHALLOW_CRATERS
#    9.5 X1-VK73-B12  ASTEROID       COMMON_METAL
#  118.1 X1-VK73-B45  ASTEROID       COMMON_METAL
#  144.5 X1-VK73-B9   ASTEROID       COMMON_METAL
#  173.5 X1-VK73-B6   FUEL_STATION   MARKETPLACE
#  181.5 X1-VK73-B42  ASTEROID       COMMON_METAL DEBRIS_CLUSTER SHALLOW_CRATERS RADIOACTIVE
#  184.1 X1-VK73-B13  ASTEROID       MINERAL DEEP_CRATERS EXPLOSIVE_GASES
def mining(c: spess.client.Client) -> None:
    manager = Manager(
        c,
        miners = ['AGRIF-3', 'AGRIF-4', 'AGRIF-6', 'AGRIF-7', 'AGRIF-8', 'AGRIF-9'],
        surveyors = ['AGRIF-5'],
        dropoff = 'X1-VK73-B7',
        mines = ['X1-VK73-B10', 'X1-VK73-B12'],
    ).handle.start()

    try:
        manager.join()
    finally:
        manager.stop()

class Manager(spess.actor.ActorLoop, spess.actor.ActorSpawnMixin):
    miners: spess.actor.HandleGroup[spess.models.Survey]

    def __init__(self,
                 c: spess.client.Client,
                 miners: list[str],
                 surveyors: list[str],
                 dropoff: str,
                 mines: list[str],
                 ) -> None:
        super().__init__('manager', block=True)
        self.c = c
        self.miner_names = miners
        self.surveyor_names = surveyors
        self.dropoff_name = dropoff
        self.mine_names = mines

        self.miners = spess.actor.HandleGroup()

    def on_start(self) -> None:
        dropoff = self.c.waypoint(self.dropoff_name)
        mines = [self.c.waypoint(m) for m in self.mine_names]

        market = self.c.market(dropoff)
        allowed = [good.symbol for good in market.trade_goods or []]

        for i, miner in enumerate(self.miner_names):
            ship = self.c.ship(miner)
            h = self.spawn(Miner(ship, dropoff, mines[i % len(mines)], allowed))
            self.miners.add(h)

        for surveyor in self.surveyor_names:
            ship = self.c.ship(surveyor)
            self.spawn(Surveyor(ship, dropoff, mines, self.miners))

class Miner(spess.actor.ActorLoop[spess.models.Survey]):
    class State(enum.Enum):
        MINING = enum.auto()
        DROPOFF = enum.auto()

    state: State
    survey: spess.models.Survey | None

    def __init__(self,
                 ship: spess.models.Ship,
                 dropoff: spess.models.Waypoint,
                 mine: spess.models.Waypoint,
                 allowed: list[spess.models.TradeSymbol],
                 ) -> None:
        super().__init__(f'miner/{ship.symbol.split("-", 1)[1]}')
        self.state = self.State.MINING
        self.ship = ship
        self.dropoff = dropoff
        self.mine = mine
        self.allowed = allowed
        self.survey = None

    def on_start(self) -> None:
        if self.ship.fuel.current < self.ship.fuel.capacity / 2:
            # prioritize refueling
            self.state = self.State.DROPOFF
        elif self.ship.cargo.units == self.ship.cargo.capacity:
            # full, go to dropoff
            self.state = self.State.DROPOFF
        elif self.ship.cargo.units < self.ship.cargo.capacity:
            # have some room, go mining
            self.state = self.State.MINING

    def on_message(self, msg: spess.models.Survey) -> None:
        if msg == self.mine:
            self.log.info(f'got new survey for [green]{msg.symbol}[/] expiring at {msg.expiration}')
            self.survey = msg

    def loop(self) -> None:
        if self.state == self.State.MINING:
            if self.ship.nav != self.mine:
                # mining but not at mine
                self.log.info(f'flying to mine [green]{self.mine.symbol}[/]')
                self.ship.orbit()
                self.ship.navigate(self.mine)
                self.ship.nav.wait()

            elif self.ship.cargo.units < self.ship.cargo.capacity:
                # mining, at mine, have room
                self.ship.orbit()
                self.ship.cooldown.wait()

                # do we have a valid survey
                survey = None
                if self.survey is not None:
                    nowmargin = datetime.datetime.now(datetime.UTC)
                    nowmargin += datetime.timedelta(seconds=10)
                    if self.survey.expiration > nowmargin:
                        survey = self.survey

                # use survey if we have it
                if survey is None:
                    # extraction without survey can't fail
                    ext = self.ship.extract_resources().extraction.yield_
                    self.log.info(f'got {ext.units} units of [yellow]{ext.symbol}[/]')
                else:
                    # extraction with survey *can* fail
                    try:
                        ext = self.ship.extract_resources(survey=survey).extraction.yield_
                    except spess.client.ClientError:
                        self.log.warning(f'survey depleted in [green]{self.mine.symbol}[/]')
                        self.survey = None
                    else:
                        self.log.info(f'got {ext.units} units of [yellow]{ext.symbol}[/] using survey')

                # dump unwanted items
                for item in self.ship.cargo.inventory:
                    if item.symbol not in self.allowed:
                        self.log.info(f'jettisoning {item.units} units of [yellod]{item.symbol}[/]')
                        self.ship.jettison_cargo(item.symbol, item.units)

                # debug inventory
                self.log.debug(', '.join(f'{item.symbol}={item.units}' for item in self.ship.cargo.inventory))

            else:
                # mining, at mine, no more room
                self.state = self.State.DROPOFF
                
        if self.state == self.State.DROPOFF:
            if self.ship.nav != self.dropoff:
                # dropoff, not at dropoff though
                self.log.info(f'flying to dropoff [green]{self.dropoff.symbol}[/]')
                self.ship.orbit()
                self.ship.navigate(self.dropoff)
                self.ship.nav.wait()

            elif self.ship.fuel.current < self.ship.fuel.capacity / 2:
                # dropoff, at dropoff, need fuel
                self.ship.dock()
                refuel = self.ship.refuel()
                self.log.info(f'bought {refuel.transaction.units} fuel for {refuel.transaction.total_price} credits')

            elif self.ship.cargo.units > 0:
                # dropoff, at dropoff, have cargo. sell it.
                self.ship.dock()
                amt = 0
                for item in self.ship.cargo.inventory:
                    if item.symbol in self.allowed:
                        sold = self.ship.sell_cargo(item.symbol, item.units)
                        self.log.info(f'sold {sold.transaction.units} units of [yellow]{item.symbol}[/] for {sold.transaction.total_price} credits ({sold.transaction.price_per_unit}/u)')
                        amt += sold.transaction.total_price
                self.log.info(f'profit {amt} credits')

                # go mining for sure after selling
                self.state = self.State.MINING

            else:
                # dropoff, at dropoff, no cargo. go get some.
                self.state = self.State.MINING

class Surveyor(spess.actor.ActorLoop):
    target: spess.models.Waypoint | None
    surveys: dict[str, spess.models.Survey]
    margin: datetime.timedelta

    def __init__(self,
                 ship: spess.models.Ship,
                 home: spess.models.Waypoint,
                 mines: list[spess.models.Waypoint],
                 miners: spess.actor.HandleGroup[spess.models.Survey],
                 )-> None:
        super().__init__(f'surveyor/{ship.symbol.split("-", 1)[1]}')
        self.ship = ship
        self.home = home
        self.mines = mines
        self.miners = miners

        self.target = None
        self.surveys = {}
        self.margin = datetime.timedelta(minutes=5)

    def loop(self) -> None:
        if self.target is None:
            # calculate expired or unset surveys
            expired = []
            for wp in self.mines:
                if wp.symbol in self.surveys:
                    expiretime = datetime.datetime.now(datetime.UTC)
                    expiretime += self.margin
                    if self.surveys[wp.symbol].expiration < expiretime:
                        expired.append(wp)
                else:
                    expired.append(wp)

            if self.ship.fuel.current < self.ship.fuel.capacity / 2:
                if self.ship.nav != self.home:
                    # not surveying, need fuel, not home
                    self.log.info(f'flying to home [green]{self.home.symbol}[/]')
                    self.ship.orbit()
                    self.ship.navigate(self.home)
                    self.ship.nav.wait()

                else:
                    # not surveying, need fuel, at home
                    self.ship.dock()
                    refuel = self.ship.refuel()
                    self.log.info(f'bought {refuel.transaction.units} fuel for {refuel.transaction.total_price} credits')

            elif expired:
                # not surveying, expired surveys. go surveyin'
                self.target = expired[0]

            else:
                if self.ship.nav != self.home:
                    # not surveying, not home, good fuel. idle at home
                    self.log.info(f'flying to home [green]{self.home.symbol}[/]')
                    self.ship.orbit()
                    self.ship.navigate(self.home)
                    self.ship.nav.wait()
                    self.ship.dock()
                else:
                    # not surveying, at home, good fuel. just wait.
                    time.sleep(self.margin.total_seconds() / 10)

        else:
            if self.ship.nav != self.target:
                # surveying, not at target. fly there
                self.log.info(f'flying to mine [green]{self.target.symbol}[/]')
                self.ship.orbit()
                self.ship.navigate(self.target)
                self.ship.nav.wait()

            else:
                # surveying, at target. do a survey.
                self.ship.cooldown.wait()
                self.ship.orbit()
                surveys = self.ship.create_survey().surveys

                # set a default survey
                if surveys:
                    self.surveys[self.target.symbol] = surveys[0]
                else:
                    self.log.warning(f'surveying failed!')

                # send the surveys to the miners
                # keep track of which one expires first
                for survey in surveys:
                    if survey.expiration < self.surveys[self.target.symbol].expiration:
                        self.surveys[self.target.symbol] = survey
                    deposits = ', '.join([f'[yellow]{d.symbol}[/]' for d in survey.deposits])
                    self.log.info(f'survey for [green]{self.target.symbol}[/]: {str(survey.size)} {deposits}')
                    self.log.info(f'survey for [green]{self.target.symbol}[/] expires {survey.expiration}')
                    self.miners.send(survey)

                # unset target
                self.target = None

if __name__ == '__main__':
    # set up pretty logger
    logging.basicConfig(
        format='[bold]%(threadName)s[/]: %(message)s',
        level=logging.INFO,
        handlers=[rich.logging.RichHandler(markup=True)],
    )

    # load token
    here = pathlib.Path(__file__).parent
    with open(here / 'token') as f:
        token = f.read().strip()

    # create client
    client = spess.client.Client(token)

    # set up interactive namespace
    vars = globals()
    vars.update({
        'client': client,
        'c': client,
        'models': spess.models,
        'm': spess.models,
    })

    # run a repl
    readline.set_completer(rlcompleter.Completer(vars).complete)
    readline.parse_and_bind('tab: complete')
    rich.pretty.install()
    code.InteractiveConsole(vars).interact()