from twisted.internet import reactor
from twisted.internet import task
from twisted.internet import defer
from colorama import Fore, Back, Style
from twisted.python import log
from twisted.internet.task import coiterate
from twisted.internet.defer import gatherResults
from itertools import cycle
import pendulum
from pprint import pformat
def merge(color_string):
""" Given a string of colorama ANSI, merge them if you can. """
return color_string.replace("m\x1b[", ";")
class PlayerInput(object):
""" Player Input
Example:
from flexible import PlayerInput
ask = PlayerInput(self.game)
# abort_blank means, if the input field is blank, abort. Use error_back.
d = ask.prompt("What is your quest?", 40, name="quest", abort_blank=True)
# Display the user's input / but not needed.
d.addCallback(ask.output)
d.addCallback(
lambda ignore: ask.prompt(
"What is your favorite color?", 10, name="color"
)
)
d.addCallback(ask.output)
d.addCallback(
lambda ignore: ask.prompt(
"What is your least favorite number?",
12,
name="number",
digits=True,
)
)
d.addCallback(ask.output)
def show_values(show):
log.msg(show)
self.queue_game.put(pformat(show).replace("\n", "\n\r") + self.nl)
d.addCallback(lambda ignore: show_values(ask.keep))
d.addCallback(self.welcome_back)
# On error, just return back
d.addErrback(self.welcome_back)
"""
def __init__(self, game):
# I think game gives us access to everything we need
self.game = game
self.observer = self.game.observer
self.save = None
self.deferred = None
self.queue_game = game.queue_game
self.keep = {}
# default colors
self.c = merge(Style.BRIGHT + Fore.WHITE + Back.BLUE)
self.cp = merge(Style.BRIGHT + Fore.YELLOW + Back.BLUE)
# useful consts
self.r = Style.RESET_ALL
self.nl = "\n\r"
self.bsb = "\b \b"
self.keepalive = None
def color(self, c):
self.c = c
def colorp(self, cp):
self.cp = cp
def alive(self):
log.msg("PlayerInput.alive()")
self.game.queue_player.put(" ")
def prompt(self, user_prompt, limit, **kw):
""" Generate prompt for user input.
Note: This returns deferred.
prompt = text displayed.
limit = # of characters allowed.
default = (text to default to)
keywords:
abort_blank : Abort if they give us blank text.
name : Stores the input in self.keep dict.
digits : Only allow 0-9 to be entered.
"""
log.msg("PlayerInput({0}, {1}, {2}".format(user_prompt, limit, kw))
self.limit = limit
self.input = ""
self.kw = kw
assert self.save is None
assert self.keepalive is None
# Note: This clears out the server "keep alive"
self.save = self.observer.save()
self.observer.connect("player", self.get_input)
self.keepalive = task.LoopingCall(self.alive)
self.keepalive.start(30)
# We need to "hide" the game output.
# Otherwise it WITH mess up the user input display.
self.to_player = self.game.to_player
self.game.to_player = False
# Display prompt
# self.queue_game.put(self.r + self.nl + self.c + user_prompt + " " + self.cp)
self.queue_game.put(self.r + self.c + user_prompt + self.r + " " + self.cp)
# Set "Background of prompt"
self.queue_game.put(" " * limit + "\b" * limit)
assert self.deferred is None
d = defer.Deferred()
self.deferred = d
log.msg("Return deferred ...", self.deferred)
return d
def get_input(self, chunk):
""" Data from player (in bytes) """
chunk = chunk.decode("utf-8", "ignore")
for ch in chunk:
if ch == "\b":
if len(self.input) > 0:
self.queue_game.put(self.bsb)
self.input = self.input[0:-1]
else:
self.queue_game.put("\a")
elif ch == "\r":
self.queue_game.put(self.r + self.nl)
log.msg("Restore observer dispatch", self.save)
assert not self.save is None
self.observer.load(self.save)
self.save = None
log.msg("Disable keepalive")
self.keepalive.stop()
self.keepalive = None
line = self.input
self.input = ""
assert not self.deferred is None
self.game.to_player = self.to_player
# If they gave us the keyword name, save the value as that name
if "name" in self.kw:
self.keep[self.kw["name"]] = line
if "abort_blank" in self.kw and self.kw["abort_blank"]:
# Abort on blank input
if line.strip() == "":
# Yes, input is blank, abort.
log.msg("errback, abort_blank")
reactor.callLater(
0, self.deferred.errback, Exception("abort_blank")
)
self.deferred = None
return
# Ok, use deferred.callback, or reactor.callLater?
# self.deferred.callback(line)
reactor.callLater(0, self.deferred.callback, line)
self.deferred = None
return
elif ch.isprintable():
# Printable, but is it acceptable?
if "digits" in self.kw:
if not ch.isdigit():
self.queue_game.put("\a")
continue
if len(self.input) + 1 <= self.limit:
self.input += ch
self.queue_game.put(ch)
else:
self.queue_game.put("\a")
def output(self, line):
""" A default display of what they just input. """
log.msg("PlayerInput.output({0})".format(line))
self.game.queue_game.put(self.r + "[{0}]".format(line) + self.nl)
return line
PORT_CLASSES = {
1: "BBS",
2: "BSB",
3: "SBB",
4: "SSB",
5: "SBS",
6: "BSS",
7: "SSS",
8: "BBB",
}
CLASSES_PORT = {v: k for k, v in PORT_CLASSES.items()}
import re
class CIMWarpReport(object):
def __init__(self, game):
self.game = game
self.queue_game = game.queue_game
self.queue_player = game.queue_player
self.observer = game.observer
# Yes, at this point we would activate
self.prompt = game.buffer
self.save = self.observer.save()
# I actually don't want the player input, but I'll grab it anyway.
self.observer.connect("player", self.player)
self.observer.connect("prompt", self.game_prompt)
self.observer.connect("game-line", self.game_line)
# If we want it, it's here.
self.defer = None
self.to_player = self.game.to_player
# Hide what's happening from the player
self.game.to_player = False
self.queue_player.put("^") # Activate CIM
self.state = 1
self.warpdata = {}
self.warpcycle = cycle(["/", "-", "\\", "|"])
def game_prompt(self, prompt):
if prompt == ": ":
if self.state == 1:
# Ok, then we're ready to request the port report
self.warpcycle = cycle(["/", "-", "\\", "|"])
self.queue_player.put("I")
self.state = 2
elif self.state == 2:
self.queue_player.put("Q")
self.state = 3
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
if self.state == 3:
# Ok, time to exit
# exit from this...
self.game.to_player = self.to_player
self.observer.load(self.save)
self.save = None
self.game.warpdata = self.warpdata
self.queue_game.put("\b \b\r\n")
if not self.defer is None:
self.defer.callback(self.warpdata)
self.defer = None
def game_line(self, line):
if line == "" or line == ": ":
return
if line == ": ENDINTERROG":
return
if line.startswith('Command [TL='):
return
# This should be the CIM Report Data -- parse it
if self.warpcycle:
if len(self.warpdata) % 10 == 0:
self.queue_game.put("\b" + next(self.warpcycle))
work = line.strip()
parts = re.split(r"(?<=\d)\s", work)
parts = [int(x) for x in parts]
sector = parts.pop(0)
# tuples are nicer on memory, and the warpdata map isn't going to be changing.
self.warpdata[sector] = tuple(parts)
def __del__(self):
log.msg("CIMWarpReport {0} RIP".format(self))
def whenDone(self):
self.defer = defer.Deferred()
# Call this to chain something after we exit.
return self.defer
def player(self, chunk):
""" Data from player (in bytes). """
chunk = chunk.decode("utf-8", "ignore")
key = chunk.upper()
log.msg("CIMWarpReport.player({0}) : I AM stopping...".format(key))
# Stop the keepalive if we are activating something else
# or leaving...
# self.keepalive.stop()
self.queue_game.put("\b \b\r\n")
if not self.defer is None:
# We have something, so:
self.game.to_player = self.to_player
self.observer.load(self.save)
self.save = None
self.defer.errback(Exception("User Abort"))
self.defer = None
else:
# Still "exit" out.
self.game.to_player = self.to_player
self.observer.load(self.save)
class CIMPortReport(object):
""" Parse data from CIM Port Report
Example:
from flexible import CIMPortReport
report = CIMPortReport(self.game)
d = report.whenDone()
d.addCallback(self.port_report)
d.addErrback(self.welcome_back)
def port_report(self, portdata):
self.portdata = portdata
self.queue_game.put("Loaded {0} records.".format(len(portdata)) + self.nl)
self.welcome_back()
def welcome_back(self,*_):
... restore keep alive timers, etc.
"""
def __init__(self, game):
self.game = game
self.queue_game = game.queue_game
self.queue_player = game.queue_player
self.observer = game.observer
# Yes, at this point we would activate
self.prompt = game.buffer
self.save = self.observer.save()
# I actually don't want the player input, but I'll grab it anyway.
self.observer.connect("player", self.player)
self.observer.connect("prompt", self.game_prompt)
self.observer.connect("game-line", self.game_line)
# If we want it, it's here.
self.defer = None
self.to_player = self.game.to_player
log.msg("to_player (stored)", self.to_player)
# Hide what's happening from the player
self.game.to_player = False
self.queue_player.put("^") # Activate CIM
self.state = 1
self.portdata = {}
self.portcycle = cycle(["/", "-", "\\", "|"])
def game_prompt(self, prompt):
if prompt == ": ":
if self.state == 1:
# Ok, then we're ready to request the port report
self.portcycle = cycle(["/", "-", "\\", "|"])
self.queue_player.put("R")
self.state = 2
elif self.state == 2:
self.queue_player.put("Q")
self.state = 3
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
if self.state == 3:
# Ok, time to exit
# exit from this...
self.game.to_player = self.to_player
self.observer.load(self.save)
self.save = None
self.game.portdata = self.portdata
self.queue_game.put("\b \b\r\n")
if not self.defer is None:
self.defer.callback(self.portdata)
self.defer = None
def game_line(self, line):
if line == "" or line == ": ":
return
if line == ": ENDINTERROG":
return
# This should be the CIM Report Data -- parse it
if self.portcycle:
if len(self.portdata) % 10 == 0:
self.queue_game.put("\b" + next(self.portcycle))
work = line.replace("%", "")
parts = re.split(r"(?<=\d)\s", work)
if len(parts) == 8:
port = int(parts[0].strip())
data = dict()
def portBS(info):
if info[0] == "-":
bs = "B"
else:
bs = "S"
return (bs, int(info[1:].strip()))
data["fuel"] = dict()
data["fuel"]["sale"], data["fuel"]["units"] = portBS(parts[1])
data["fuel"]["pct"] = int(parts[2].strip())
data["org"] = dict()
data["org"]["sale"], data["org"]["units"] = portBS(parts[3])
data["org"]["pct"] = int(parts[4].strip())
data["equ"] = dict()
data["equ"]["sale"], data["equ"]["units"] = portBS(parts[5])
data["equ"]["pct"] = int(parts[6].strip())
# Store what this port is buying/selling
data["port"] = (
data["fuel"]["sale"] + data["org"]["sale"] + data["equ"]["sale"]
)
# Convert BBS/SBB to Class number 1-8
data["class"] = CLASSES_PORT[data["port"]]
self.portdata[port] = data
else:
log.msg("CIMPortReport:", line, "???")
def __del__(self):
log.msg("CIMPortReport {0} RIP".format(self))
def whenDone(self):
self.defer = defer.Deferred()
# Call this to chain something after we exit.
return self.defer
def player(self, chunk):
""" Data from player (in bytes). """
chunk = chunk.decode("utf-8", "ignore")
key = chunk.upper()
log.msg("CIMPortReport.player({0}) : I AM stopping...".format(key))
# Stop the keepalive if we are activating something else
# or leaving...
# self.keepalive.stop()
self.queue_game.put("\b \b\r\n")
if not self.defer is None:
# We have something, so:
self.game.to_player = self.to_player
self.observer.load(self.save)
self.save = None
self.defer.errback(Exception("User Abort"))
self.defer = None
else:
# Still "exit" out.
self.game.to_player = self.to_player
self.observer.load(self.save)
def port_burnt(port):
""" Is this port burned out? """
if all( x in port for x in ['fuel', 'org', 'equ']):
if all( 'pct' in port[x] for x in ['fuel', 'org', 'equ']):
if port['equ']['pct'] <= 20 or port['fuel']['pct'] <= 20 or port['org']['pct'] <= 20:
return True
return False
# Since we don't have any port information, hope for the best, assume it isn't burnt.
return False
def flip(port):
return port.replace('S', 'W').replace('B', 'S').replace('W', 'B')
def port_trading(port1, port2):
""" Are there possible trades at these ports? """
if port1 == port2:
return False
p1 = [ c for c in port1]
p2 = [ c for c in port2]
# Any that are the same? Remove them.
rem = False
for i in range(3):
if p1[i] == p2[i]:
p1[i] = 'X'
p2[i] = 'X'
rem = True
if rem:
j1 = "".join(p1).replace('X', '')
j2 = "".join(p2).replace('X', '')
if j1 == 'BS' and j2 == 'SB':
return True
if j1 == 'SB' and j2 == 'BS':
return True
# Matching 2 of them.
rport1 = flip(port1)
c = 0
match = []
for i in range(3):
if rport1[i] == port2[i]:
match.append(port2[i])
c += 1
if c > 1:
f = flip(match.pop(0))
if f in match:
return True
return False
return False
class ScriptPort(object):
""" Performs the Port script.
This is close to the original.
We don't ask for the port to trade with --
because that information is available to us after "D" (display).
We look at the adjacent sectors, and see if we know any ports.
If the ports are burnt (< 20%), we remove them from the list.
If there's just one, we use it. Otherwise we ask them to choose.
"""
def __init__(self, game):
self.game = game
self.queue_game = game.queue_game
self.queue_player = game.queue_player
self.observer = game.observer
self.r = Style.RESET_ALL
self.nl = "\n\r"
self.this_sector = None # Starting sector
self.sector1 = None # Current Sector
self.sector2 = None # Next Sector Stop
self.percent = 5 # Stick with the good default.
self.credits = 0
self.last_credits = None
self.times_left = 0
# Activate
self.prompt = game.buffer
self.save = self.observer.save()
self.observer.connect('player', self.player)
self.observer.connect("prompt", self.game_prompt)
self.observer.connect("game-line", self.game_line)
self.defer = None
self.queue_game.put(
self.nl + "Script based on: Port Pair Trading v2.00" + self.r + self.nl
)
self.possible_sectors = None
self.state = 1
self.queue_player.put("D")
# Original, send 'D' to display current sector.
# We could get the sector number from the self.prompt string -- HOWEVER:
# IF! We send 'D', we can also get the sectors around -- we might not even need to
# prompt for sector to trade with (we could possibly figure it out ourselves).
# [Command [TL=00:00:00]:[967] (?=Help)? : D]
# [<Re-Display>]
# []
# [Sector : 967 in uncharted space.]
# [Planets : (M) Into the Darkness]
# [Warps to Sector(s) : 397 - (562) - (639)]
# []
def whenDone(self):
self.defer = defer.Deferred()
# Call this to chain something after we exit.
return self.defer
def deactivate(self):
self.state = 0
log.msg("ScriptPort.deactivate ({0})".format(self.times_left))
assert(not self.save is None)
self.observer.load(self.save)
self.save = None
if self.defer:
self.defer.callback('done')
self.defer = None
def player(self, chunk: bytes):
# If we receive anything -- ABORT!
self.deactivate()
def game_prompt(self, prompt: str):
log.msg("{0} : {1}".format(self.state, prompt))
if self.state == 3:
log.msg("game_prompt: ", prompt)
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
self.state = 4
log.msg("Ok, state 4")
if self.sector2 is None:
# Ok, we need to prompt for this.
self.queue_game.put(self.r + self.nl +
"Which sector to trade with? {0}".format(port_show_part(self.this_sector, self.game.portdata[self.this_sector])) +
self.nl + Fore.CYAN)
for i, p in enumerate(self.possible):
self.queue_game.put(" " + str(i + 1) + " : " + port_show_part(p, self.game.portdata[p]) + self.nl)
pi = PlayerInput(self.game)
def got_need1(*_):
log.msg("Ok, I have:", pi.keep)
if pi.keep['count'].strip() == '':
self.deactivate()
return
self.times_left = int(pi.keep['count'])
if pi.keep['choice'].strip() == '':
self.deactivate()
return
c = int(pi.keep['choice']) -1
if c < 0 or c >= len(self.possible):
self.deactivate()
return
self.sector2 = self.possible[int(pi.keep['choice']) -1]
# self.queue_game.put(pformat(pi.keep).replace("\n", "\n\r"))
self.state = 5
self.trade()
d = pi.prompt("Choose -=>", 5, name='choice', digits=True)
d.addCallback(lambda ignore: pi.prompt("Times to execute script:", 5, name='count', digits=True))
d.addCallback(got_need1)
else:
# We already have our target port, so...
self.queue_game.put(self.r + self.nl + "Trading from {0} ({1}) to default {2} ({3}).".format(
self.this_sector,
self.game.portdata[self.this_sector]['port'],
self.sector2, self.game.portdata[self.sector2]['port']) + self.nl
)
self.queue_game.put(self.r + self.nl + "Trading {0}".format(
port_show(self.this_sector,
self.game.portdata[self.this_sector],
self.sector2,
self.game.portdata[self.sector2])
))
# The code is smart enough now, that this can't happen. :( Drat!
if self.game.portdata[self.this_sector]['port'] == self.game.portdata[self.sector2]['port']:
self.queue_game.put("Hey dummy! Look out the window! These ports are the same class!" + nl)
self.deactivate()
return
pi = PlayerInput(self.game)
def got_need2(*_):
if pi.keep['count'].strip() == '':
self.deactivate()
return
self.times_left = int(pi.keep['count'])
log.msg("Ok, I have:", pi.keep)
# self.queue_game.put(pformat(pi.keep).replace("\n", "\n\r"))
self.state = 5
self.trade()
self.queue_game.put(self.r + self.nl)
d = pi.prompt("Times to execute script", 5, name='count')
d.addCallback(got_need2)
elif self.state == 6:
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
if self.fixable:
log.msg("Fixing...")
if self.this_sector == self.sector1:
self.this_sector = self.sector2
self.queue_player.put("{0}\r".format(self.sector2))
self.state = 5
self.trade()
else:
self.this_sector = self.sector1
self.queue_player.put("{0}\r".format(self.sector1))
self.state = 5
self.trade()
elif self.state == 7:
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
# Done
if self.this_sector == self.sector1:
self.this_sector = self.sector2
self.queue_player.put("{0}\r".format(self.sector2))
self.state = 10
else:
self.times_left -= 1
if self.times_left <= 0:
# Ok, exit out
self.deactivate()
return
if self.last_credits is None:
self.last_credits = self.credits
else:
if self.credits <= self.last_credits:
log.msg("We don't seem to be making any money here...")
self.queue_game.put(self.r + self.nl + "We don't seem to be making any money here. I'm stopping!" + self.nl)
self.deactivate()
return
self.this_sector = self.sector1
self.queue_player.put("{0}\r".format(self.sector1))
self.state = 10
elif self.state == 8:
# What are we trading
# How many holds of Equipment do you want to buy [75]?
if re.match(r"How many holds of .+ do you want to buy \[\d+\]\?", prompt):
parts = prompt.split()
trade_type = parts[4]
if trade_type == 'Fuel':
if (self.tpc in (5,7)) and (self.opc in (2,3,4,8)):
# Can buy equipment - fuel ore is worthless.
self.queue_player.put("0\r")
return
if (self.tpc in(4,7)) and (self.opc in (1,3,5,8)):
# Can buy organics - fuel ore is worthless.
self.queue_player.put("0\r")
return
if (self.tpc in (4,7,3,5)) and (self.opc in (3,4,5,7)):
# No point in buying fuel ore if it can't be sold.
self.queue_player.put("0\r")
return
elif trade_type == 'Organics':
if (self.tpc in (6,7)) and (self.opc in (2,3,4,8)):
# Can buy equipment - organics is worthless.
self.queue_player.put("0\r")
return
if (self.tpc in (2,4,6,7)) and (self.opc in (2,4,6,7)):
# No point in buying organics if it can't be sold.
self.queue_player.put("0\r")
return
elif trade_type == 'Equipment':
if (self.opc in (1,5,6,7)):
# No point in buying equipment if it can't be sold.
self.queue_player.put("0\r")
return
self.queue_player.put("\r")
elif re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
# Done
if self.this_sector == self.sector1:
self.this_sector = self.sector2
self.queue_player.put("{0}\r".format(self.sector2))
self.state = 10
else:
self.times_left -= 1
if self.times_left <= 0:
# Ok, exit out
self.deactivate()
return
if self.last_credits is None:
self.last_credits = self.credits
else:
if self.credits <= self.last_credits:
log.msg("We don't seem to be making any money here...")
self.deactivate()
return
self.this_sector = self.sector1
self.queue_player.put("{0}\r".format(self.sector1))
self.state = 10
def trade(self, *_):
# state 5
log.msg("trade!")
self.queue_player.put("pt") # Port Trade
self.this_port = self.game.portdata[self.this_sector]
if self.this_sector == self.sector1:
self.other_port = self.game.portdata[self.sector2]
else:
self.other_port = self.game.portdata[self.sector1]
# Ok, perform some calculations
self.tpc = self.this_port['class']
self.opc = self.other_port['class']
self.fixable = 0
# [ Items Status Trading % of max OnBoard]
# [ ----- ------ ------- -------- -------]
# [Fuel Ore Selling 2573 93% 0]
# [Organics Buying 2960 100% 0]
# [Equipment Buying 1958 86% 0]
# []
# []
# [You have 1,000 credits and 20 empty cargo holds.]
# []
# [We are selling up to 2573. You have 0 in your holds.]
# [How many holds of Fuel Ore do you want to buy [20]? 0]
def game_line(self, line: str):
if line.startswith("You have ") and 'credits and' in line:
parts = line.replace(',', '').split()
credits = int(parts[2])
log.msg("Credits: {0}".format(credits))
self.credits = credits
if self.state == 1:
# First exploration
if line.startswith("Sector :"):
# We have starting sector information
parts = re.split("\s+", line)
self.this_sector = int(parts[2])
# These will be the ones swapped around as we trade back and forth.
self.sector1 = self.this_sector
elif line.startswith("Warps to Sector(s) : "):
# Warps to Sector(s) : 397 - (562) - (639)
_, _, warps = line.partition(':')
warps = warps.replace('-', '').replace('(', '').replace(')', '').strip()
log.msg("Warps: [{0}]".format(warps))
self.warps = [ int(x) for x in re.split("\s+", warps)]
log.msg("Warps: [{0}]".format(self.warps))
self.state = 2
elif self.state == 2:
if line == "":
# Ok, we're done
self.state = 3
# Check to see if we have information on any possible ports
if hasattr(self.game, 'portdata'):
if not self.this_sector in self.game.portdata:
self.state = 0
log.msg("Current sector {0} not in portdata.".format(self.this_sector))
self.queue_game.put(self.r + self.nl + "I can't find the current sector in the portdata." + self.nl)
self.deactivate()
return
else:
# Ok, we are in the portdata
pd = self.game.portdata[self.this_sector]
if port_burnt(pd):
log.msg("Current sector {0} port is burnt (<= 20%).".format(self.this_sector))
self.queue_game.put(self.r + self.nl + "Current sector port is burnt out. <= 20%." + self.nl)
self.deactivate()
return
possible = [ x for x in self.warps if x in self.game.portdata ]
log.msg("Possible:", possible)
# BUG: Sometimes links to another sector, don't link back!
# This causes the game to plot a course / autopilot.
if hasattr(self.game, 'warpdata'):
# Great! verify that those warps link back to us!
possible = [ x for x in possible if self.this_sector in self.game.warpdata[x]]
if len(possible) == 0:
self.state = 0
self.queue_game.put(self.r + self.nl + "I don't see any ports in [{0}].".format(self.warps) + self.nl)
self.deactivate()
return
possible = [ x for x in possible if not port_burnt(self.game.portdata[x]) ]
log.msg("Possible:", possible)
if len(possible) == 0:
self.state = 0
self.queue_game.put(self.r + self.nl + "I don't see any unburnt ports in [{0}].".format(self.warps) + self.nl)
self.deactivate()
return
possible = [ x for x in possible if port_trading(self.game.portdata[self.this_sector]['port'], self.game.portdata[x]['port'])]
self.possible = possible
if len(possible) == 0:
self.state = 0
self.queue_game.put(self.r + self.nl + "I don't see any possible port trades in [{0}].".format(self.warps) + self.nl)
self.deactivate()
return
elif len(possible) == 1:
# Ok! there's only one!
self.sector2 = possible[0]
# Display possible ports:
# spos = [ str(x) for x in possible]
# self.queue_game.put(self.r + self.nl + self.nl.join(spos) + self.nl)
# At state 3, we only get a prompt.
return
else:
self.state = 0
log.msg("We don't have any portdata!")
self.queue_game.put(self.r + self.nl + "I have no portdata. Please run CIM Port Report." + self.nl)
self.deactivate()
return
elif self.state == 5:
if "-----" in line:
self.state = 6
elif self.state == 6:
if "We are buying up to" in line:
# Sell
self.state = 7
self.queue_player.put("\r")
self.sell_perc = 100 + self.percent
if "We are selling up to" in line:
# Buy
self.state = 8
self.sell_perc = 100 - self.percent
if line.startswith('Fuel Ore') or line.startswith('Organics') or line.startswith('Equipment'):
work = line.replace('Fuel Ore', 'Fuel')
# parts = re.split(r"\s+", work)
log.msg(parts)
if parts[0] != '0' and parts[1] != 'Buying':
log.msg("We have a problem -- they aren't buying what we have in stock!")
stuff = line[0] # F O or E.
if stuff == 'F':
spos = 0
elif stuff == 'O':
spos = 1
else:
spos = 2
other = self.other_port['port']
if other[spos] == 'B':
self.fixable = 1
if "You don't have anything they want" in line:
# Neither! DRAT!
if not self.fixable:
self.deactivate()
return
if "We're not interested." in line:
log.msg("Try, try again. :(")
self.state = 5
self.trade()
elif self.state == 7:
# Haggle Sell
if "We'll buy them for" in line or "Our final offer" in line:
if "Our final offer" in line:
self.sell_perc -= 1
parts = line.replace(',', '').split()
start_price = int(parts[4])
price = start_price * self.sell_perc // 100
log.msg("start: {0} % {1} price {2}".format(start_price, self.sell_perc, price))
self.sell_perc -= 1
self.queue_player.put("{0}\r".format(price))
if "We are selling up to" in line:
# Buy
self.state = 8
self.sell_perc = 100 - self.percent
if "We're not interested." in line:
log.msg("Try, try again. :(")
self.state = 5
self.trade()
elif self.state == 8:
# Haggle Buy
if "We'll sell them for" in line or "Our final offer" in line:
if "Our final offer" in line:
self.sell_perc += 1
parts = line.replace(',', '').split()
start_price = int(parts[4])
price = start_price * self.sell_perc // 100
log.msg("start: {0} % {1} price {2}".format(start_price, self.sell_perc, price))
self.sell_perc += 1
self.queue_player.put("{0}\r".format(price))
if "We're not interested." in line:
log.msg("Try, try again. :(")
self.state = 5
self.trade()
elif self.state == 10:
if "Sector : " in line:
# Trade
self.state = 5
reactor.callLater(0, self.trade, 0)
# self.trade()
# elif self.state == 3:
# log.msg("At state 3 [{0}]".format(line))
# self.queue_game.put("At state 3.")
# self.deactivate()
# return
def port_pct(port):
# Make sure these exist in the port data given.
if all( x in port for x in ['fuel', 'org', 'equ']):
return "{0:3},{1:3},{2:3}%".format(
port['fuel']['pct'],
port['org']['pct'],
port['equ']['pct'])
else:
return "---,---,---%"
def port_show_part(sector, sector_port):
return "{0:5} ({1}) {2}".format(sector, sector_port['port'], port_pct(sector_port))
def port_show(sector, sector_port, warp, warp_port):
sector_pct = port_pct(sector_port)
warp_pct = port_pct(warp_port)
return "{0} -=- {1}".format(
port_show_part(sector, sector_port),
port_show_part(warp, warp_port)
)
class ProxyMenu(object):
""" Display ProxyMenu
Example:
from flexible import ProxyMenu
if re.match(r"Command \[TL=.* \(\?=Help\)\? :", prompt):
menu = ProxyMenu(self.game)
"""
def __init__(self, game):
self.nl = "\n\r"
self.c = merge(Style.BRIGHT + Fore.YELLOW + Back.BLUE)
self.r = Style.RESET_ALL
self.c1 = merge(Style.BRIGHT + Fore.BLUE)
self.c2 = merge(Style.NORMAL + Fore.CYAN)
self.portdata = None
self.game = game
self.queue_game = game.queue_game
self.observer = game.observer
# Am I using self or game? (I think I want game, not self.)
if hasattr(self.game, "portdata"):
self.portdata = self.game.portdata
else:
self.portdata = {}
if hasattr(self.game, 'warpdata'):
self.warpdata = self.game.warpdata
else:
self.warpdata = {}
if hasattr(self.game, 'trade_report'):
self.trade_report = self.game.trade_report
else:
self.trade_report = []
# Yes, at this point we would activate
self.prompt = game.buffer
self.save = self.observer.save()
self.observer.connect("player", self.player)
# If we want it, it's here.
self.defer = None
self.keepalive = task.LoopingCall(self.awake)
self.keepalive.start(30)
self.menu()
def __del__(self):
log.msg("ProxyMenu {0} RIP".format(self))
def whenDone(self):
self.defer = defer.Deferred()
# Call this to chain something after we exit.
return self.defer
def menu(self):
self.queue_game.put(
self.nl + self.c + "TradeWars Proxy active." + self.r + self.nl
)
def menu_item(ch: str, desc: str):
self.queue_game.put(
" " + self.c1 + ch + self.c2 + " - " + self.c1 + desc + self.nl
)
menu_item("D", "Display Report again")
# menu_item("Q", "Quest")
if hasattr(self.game, 'portdata'):
ports = len(self.game.portdata)
else:
ports = '?'
menu_item("P", "Port CIM Report ({0})".format(ports))
if hasattr(self.game, 'warpdata'):
warps = len(self.game.warpdata)
else:
warps = '?'
menu_item("W", "Warp CIM Report ({0})".format(warps))
menu_item("T", "Trading Report")
menu_item("S", "Scripts")
menu_item("X", "eXit")
self.queue_game.put(" " + self.c + "-=>" + self.r + " ")
def awake(self):
log.msg("ProxyMenu.awake()")
self.game.queue_player.put(" ")
def port_report(self, portdata: dict):
self.portdata = portdata
self.game.portdata = portdata
self.queue_game.put("Loaded {0} ports.".format(len(portdata)) + self.nl)
self.welcome_back()
def warp_report(self, warpdata: dict):
self.warpdata = warpdata
self.game.warpdata = warpdata
self.queue_game.put("Loaded {0} sectors.".format(len(warpdata)) + self.nl)
self.welcome_back()
def make_trade_report(self):
log.msg("make_trade_report()")
ok_trades = []
best_trades = []
for sector, pd in self.game.portdata.items():
if not port_burnt(pd):
pc = pd['class']
# Ok, let's look into it.
if not sector in self.game.warpdata:
continue
warps = self.game.warpdata[sector]
for w in warps:
# Verify that we have that warp's info, and that the sector is in it.
# (We can get back from it)
if w in self.game.warpdata and sector in self.game.warpdata[w]:
# Ok, we can get there -- and get back!
if w > sector and w in self.game.portdata and not port_burnt(self.game.portdata[w]):
# it is > and has a port.
wd = self.game.portdata[w]
wc = wd['class']
# 1: "BBS",
# 2: "BSB",
# 3: "SBB",
# 4: "SSB",
# 5: "SBS",
# 6: "BSS",
# 7: "SSS",
# 8: "BBB",
if pc in (1,5) and wc in (2,4):
best_trades.append(port_show(sector, pd, w, wd))
# best_trades.append( "{0:5} -=- {1:5}".format(sector, w))
elif pc in (2,4) and wc in (1,5):
best_trades.append(port_show(sector, pd, w, wd))
# best_trades.append( "{0:5} -=- {1:5}".format(sector, w))
elif port_trading(pd['port'], self.game.portdata[w]['port']):
# ok_trades.append( "{0:5} -=- {1:5}".format(sector,w))
ok_trades.append(port_show(sector, pd, w, wd))
yield
self.trade_report.append("Best Trades: (org/equ)")
self.trade_report.extend(best_trades)
self.trade_report.append("Ok Trades:")
self.trade_report.extend(ok_trades)
# self.queue_game.put("BEST TRADES:" + self.nl + self.nl.join(best_trades) + self.nl)
# self.queue_game.put("OK TRADES:" + self.nl + self.nl.join(ok_trades) + self.nl)
def show_trade_report(self, *_):
self.game.trade_report = self.trade_report
for t in self.trade_report:
self.queue_game.put(t + self.nl)
self.welcome_back()
def player(self, chunk: bytes):
""" Data from player (in bytes). """
chunk = chunk.decode("utf-8", "ignore")
key = chunk.upper()
log.msg("ProxyMenu.player({0})".format(key))
# Stop the keepalive if we are activating something else
# or leaving...
self.keepalive.stop()
if key == "T":
self.queue_game.put(self.c + key + self.r + self.nl)
# do we have enough information to do this?
if not hasattr(self.game, 'portdata') and not hasattr(self.game, 'warpdata'):
self.queue_game.put("Missing portdata and warpdata." + self.nl)
elif not hasattr(self.game, 'portdata'):
self.queue_game.put("Missing portdata." + self.nl)
elif not hasattr(self.game, 'warpdata'):
self.queue_game.put("Missing warpdata." + self.nl)
else:
# Yes, so let's start!
self.trade_report = []
d = coiterate(self.make_trade_report())
d.addCallback(self.show_trade_report)
return
elif key == "P":
self.queue_game.put(self.c + key + self.r + self.nl)
# Activate CIM Port Report
report = CIMPortReport(self.game)
d = report.whenDone()
d.addCallback(self.port_report)
d.addErrback(self.welcome_back)
return
elif key == "W":
self.queue_game.put(self.c + key + self.r + self.nl)
# Activate CIM Port Report
report = CIMWarpReport(self.game)
d = report.whenDone()
d.addCallback(self.warp_report)
d.addErrback(self.welcome_back)
return
elif key == "S":
self.queue_game.put(self.c + key + self.r + self.nl)
self.activate_scripts_menu()
return
elif key == "D":
self.queue_game.put(self.c + key + self.r + self.nl)
for t in self.trade_report:
self.queue_game.put(t + self.nl)
# self.queue_game.put(pformat(self.portdata).replace("\n", "\n\r") + self.nl)
# self.queue_game.put(pformat(self.warpdata).replace("\n", "\n\r") + self.nl)
elif key == "Q":
self.queue_game.put(self.c + key + self.r + self.nl)
# This is an example of chaining PlayerInput prompt calls.
ask = PlayerInput(self.game)
d = ask.prompt("What is your quest?", 40, name="quest", abort_blank=True)
# Display the user's input
d.addCallback(ask.output)
d.addCallback(
lambda ignore: ask.prompt(
"What is your favorite color?", 10, name="color"
)
)
d.addCallback(ask.output)
d.addCallback(
lambda ignore: ask.prompt(
"What is the meaning of the squirrel?",
12,
name="squirrel",
digits=True,
)
)
d.addCallback(ask.output)
def show_values(show):
log.msg(show)
self.queue_game.put(pformat(show).replace("\n", "\n\r") + self.nl)
d.addCallback(lambda ignore: show_values(ask.keep))
d.addCallback(self.welcome_back)
# On error, just return back
# This doesn't seem to be getting called.
# d.addErrback(lambda ignore: self.welcome_back)
d.addErrback(self.welcome_back)
return
elif key == "X":
self.queue_game.put(self.c + key + self.r + self.nl)
self.queue_game.put("Proxy done." + self.nl)
self.observer.load(self.save)
self.save = None
# It isn't running (NOW), so don't try to stop it.
# self.keepalive.stop()
self.keepalive = None
# Ok, this is a HORRIBLE idea, because the prompt might be
# outdated.
# self.queue_game.put(self.prompt)
self.prompt = None
# Possibly: Send '\r' to re-display the prompt
# instead of displaying the original one.
self.game.queue_player.put("d")
# Were we asked to do something when we were done here?
if self.defer:
reactor.CallLater(0, self.defer.callback)
# self.defer.callback()
self.defer = None
return
self.keepalive.start(30, True)
self.menu()
def activate_scripts_menu(self):
self.observer.disconnect("player", self.player)
self.observer.connect("player", self.scripts_player)
self.scripts_menu()
def deactivate_scripts_menu(self, *_):
self.observer.disconnect("player", self.scripts_player)
self.observer.connect("player", self.player)
self.welcome_back()
def scripts_menu(self, *_):
c1 = merge(Style.BRIGHT + Fore.CYAN)
c2 = merge(Style.NORMAL + Fore.CYAN)
def menu_item(ch, desc):
self.queue_game.put(
" " + c1 + ch + c2 + " - " + c1 + desc + self.nl
)
menu_item("1", "Ports (Trades between two sectors)")
menu_item("2", "TODO")
menu_item("3", "TODO")
menu_item("X", "eXit")
self.queue_game.put(" " + c1 + "-=>" + self.r + " ")
def scripts_player(self, chunk: bytes):
""" Data from player (in bytes). """
chunk = chunk.decode("utf-8", "ignore")
key = chunk.upper()
if key == '1':
self.queue_game.put(self.c + key + self.r + self.nl)
# Activate this magical event here
ports = ScriptPort(self.game)
d = ports.whenDone()
# d.addCallback(self.scripts_menu)
# d.addErrback(self.scripts_menu)
d.addCallback(self.deactivate_scripts_menu)
d.addErrback(self.deactivate_scripts_menu)
return
elif key == 'X':
self.queue_game.put(self.c + key + self.r + self.nl)
self.deactivate_scripts_menu()
return
else:
self.queue_game.put(self.c + "?" + self.r + self.nl)
self.scripts_menu()
def welcome_back(self, *_):
log.msg("welcome_back")
self.keepalive.start(30, True)
self.menu()