move networks commands to [str, Optional[dict]] across the board
and some other updates PrintHTML is an experiment and is unlikely the solution I'll go with
This commit is contained in:
Fabian Dill
parent
670b8b4b11
commit
c604dfe509
196
MultiClient.py
196
MultiClient.py
|
|
@ -8,12 +8,9 @@ import functools
|
|||
import webbrowser
|
||||
import multiprocessing
|
||||
import socket
|
||||
import sys
|
||||
import typing
|
||||
import os
|
||||
import subprocess
|
||||
import base64
|
||||
from json import loads, dumps
|
||||
|
||||
from random import randrange
|
||||
|
||||
|
|
@ -26,11 +23,10 @@ import ModuleUpdate
|
|||
ModuleUpdate.update()
|
||||
|
||||
import colorama
|
||||
import websockets
|
||||
import prompt_toolkit
|
||||
|
||||
from prompt_toolkit.patch_stdout import patch_stdout
|
||||
from NetUtils import Endpoint
|
||||
from NetUtils import *
|
||||
import WebUI
|
||||
|
||||
from worlds.alttp import Regions
|
||||
|
|
@ -44,10 +40,10 @@ logger = logging.getLogger("Client")
|
|||
def create_named_task(coro, *args, name=None):
|
||||
if not name:
|
||||
name = coro.__name__
|
||||
if sys.version_info.major > 2 and sys.version_info.minor > 7:
|
||||
return asyncio.create_task(coro, *args, name=name)
|
||||
else:
|
||||
return asyncio.create_task(coro, *args)
|
||||
return asyncio.create_task(coro, *args, name=name)
|
||||
|
||||
|
||||
coloramaparser = HTMLtoColoramaParser()
|
||||
|
||||
|
||||
class Context():
|
||||
|
|
@ -125,19 +121,6 @@ class Context():
|
|||
await self.server.socket.send(dumps(msgs))
|
||||
|
||||
|
||||
color_codes = {'reset': 0, 'bold': 1, 'underline': 4, 'black': 30, 'red': 31, 'green': 32, 'yellow': 33, 'blue': 34,
|
||||
'magenta': 35, 'cyan': 36, 'white': 37, 'black_bg': 40, 'red_bg': 41, 'green_bg': 42, 'yellow_bg': 43,
|
||||
'blue_bg': 44, 'purple_bg': 45, 'cyan_bg': 46, 'white_bg': 47}
|
||||
|
||||
|
||||
def color_code(*args):
|
||||
return '\033[' + ';'.join([str(color_codes[arg]) for arg in args]) + 'm'
|
||||
|
||||
|
||||
def color(text, *args):
|
||||
return color_code(*args) + text + color_code('reset')
|
||||
|
||||
|
||||
START_RECONNECT_DELAY = 5
|
||||
SNES_RECONNECT_DELAY = 5
|
||||
SERVER_RECONNECT_DELAY = 5
|
||||
|
|
@ -516,7 +499,6 @@ async def snes_connect(ctx: Context, address):
|
|||
await snes_disconnect(ctx)
|
||||
return
|
||||
|
||||
|
||||
logger.info("Attaching to " + device)
|
||||
|
||||
Attach_Request = {
|
||||
|
|
@ -532,7 +514,7 @@ async def snes_connect(ctx: Context, address):
|
|||
if 'sd2snes' in device.lower() or 'COM' in device:
|
||||
logger.info("SD2SNES/FXPAK Detected")
|
||||
ctx.is_sd2snes = True
|
||||
await ctx.snes_socket.send(dumps({"Opcode" : "Info", "Space" : "SNES"}))
|
||||
await ctx.snes_socket.send(dumps({"Opcode": "Info", "Space": "SNES"}))
|
||||
reply = loads(await ctx.snes_socket.recv())
|
||||
if reply and 'Results' in reply:
|
||||
logger.info(reply['Results'])
|
||||
|
|
@ -605,7 +587,7 @@ async def snes_recv_loop(ctx: Context):
|
|||
asyncio.create_task(snes_autoreconnect(ctx))
|
||||
|
||||
|
||||
async def snes_read(ctx : Context, address, size):
|
||||
async def snes_read(ctx: Context, address, size):
|
||||
try:
|
||||
await ctx.snes_request_lock.acquire()
|
||||
|
||||
|
|
@ -613,9 +595,9 @@ async def snes_read(ctx : Context, address, size):
|
|||
return None
|
||||
|
||||
GetAddress_Request = {
|
||||
"Opcode" : "GetAddress",
|
||||
"Space" : "SNES",
|
||||
"Operands" : [hex(address)[2:], hex(size)[2:]]
|
||||
"Opcode": "GetAddress",
|
||||
"Space": "SNES",
|
||||
"Operands": [hex(address)[2:], hex(size)[2:]]
|
||||
}
|
||||
try:
|
||||
await ctx.snes_socket.send(dumps(GetAddress_Request))
|
||||
|
|
@ -634,7 +616,7 @@ async def snes_read(ctx : Context, address, size):
|
|||
if len(data):
|
||||
logger.error(str(data))
|
||||
logger.warning('Unable to connect to SNES Device because QUsb2Snes broke temporarily.'
|
||||
'Try un-selecting and re-selecting the SNES Device.')
|
||||
'Try un-selecting and re-selecting the SNES Device.')
|
||||
if ctx.snes_socket is not None and not ctx.snes_socket.closed:
|
||||
await ctx.snes_socket.close()
|
||||
return None
|
||||
|
|
@ -644,7 +626,7 @@ async def snes_read(ctx : Context, address, size):
|
|||
ctx.snes_request_lock.release()
|
||||
|
||||
|
||||
async def snes_write(ctx : Context, write_list):
|
||||
async def snes_write(ctx: Context, write_list):
|
||||
try:
|
||||
await ctx.snes_request_lock.acquire()
|
||||
|
||||
|
|
@ -661,15 +643,15 @@ async def snes_write(ctx : Context, write_list):
|
|||
logger.error("SD2SNES: Write out of range %s (%d)" % (hex(address), len(data)))
|
||||
return False
|
||||
for ptr, byte in enumerate(data, address + 0x7E0000 - WRAM_START):
|
||||
cmd += b'\xA9' # LDA
|
||||
cmd += b'\xA9' # LDA
|
||||
cmd += bytes([byte])
|
||||
cmd += b'\x8F' # STA.l
|
||||
cmd += b'\x8F' # STA.l
|
||||
cmd += bytes([ptr & 0xFF, (ptr >> 8) & 0xFF, (ptr >> 16) & 0xFF])
|
||||
|
||||
cmd += b'\xA9\x00\x8F\x00\x2C\x00\x68\xEB\x68\x28\x6C\xEA\xFF\x08'
|
||||
|
||||
PutAddress_Request['Space'] = 'CMD'
|
||||
PutAddress_Request['Operands'] = ["2C00", hex(len(cmd)-1)[2:], "2C00", "1"]
|
||||
PutAddress_Request['Operands'] = ["2C00", hex(len(cmd) - 1)[2:], "2C00", "1"]
|
||||
try:
|
||||
if ctx.snes_socket is not None:
|
||||
await ctx.snes_socket.send(dumps(PutAddress_Request))
|
||||
|
|
@ -681,7 +663,7 @@ async def snes_write(ctx : Context, write_list):
|
|||
else:
|
||||
PutAddress_Request['Space'] = 'SNES'
|
||||
try:
|
||||
#will pack those requests as soon as qusb2snes actually supports that for real
|
||||
# will pack those requests as soon as qusb2snes actually supports that for real
|
||||
for address, data in write_list:
|
||||
PutAddress_Request['Operands'] = [hex(address)[2:], hex(len(data))[2:]]
|
||||
if ctx.snes_socket is not None:
|
||||
|
|
@ -697,7 +679,7 @@ async def snes_write(ctx : Context, write_list):
|
|||
ctx.snes_request_lock.release()
|
||||
|
||||
|
||||
def snes_buffered_write(ctx : Context, address, data):
|
||||
def snes_buffered_write(ctx: Context, address, data):
|
||||
if ctx.snes_write_buffer and (ctx.snes_write_buffer[-1][0] + len(ctx.snes_write_buffer[-1][1])) == address:
|
||||
# append to existing write command, bundling them
|
||||
ctx.snes_write_buffer[-1] = (ctx.snes_write_buffer[-1][0], ctx.snes_write_buffer[-1][1] + data)
|
||||
|
|
@ -705,7 +687,7 @@ def snes_buffered_write(ctx : Context, address, data):
|
|||
ctx.snes_write_buffer.append((address, data))
|
||||
|
||||
|
||||
async def snes_flush_writes(ctx : Context):
|
||||
async def snes_flush_writes(ctx: Context):
|
||||
if not ctx.snes_write_buffer:
|
||||
return
|
||||
|
||||
|
|
@ -733,11 +715,11 @@ async def server_loop(ctx: Context, address=None):
|
|||
if address is None: # see if this is an old connection
|
||||
await asyncio.sleep(0.5) # wait for snes connection to succeed if possible.
|
||||
rom = ctx.rom if ctx.rom else None
|
||||
|
||||
servers = Utils.persistent_load()["servers"]
|
||||
if rom in servers:
|
||||
address = servers[rom]
|
||||
cached_address = True
|
||||
if rom:
|
||||
servers = Utils.persistent_load()["servers"]
|
||||
if rom in servers:
|
||||
address = servers[rom]
|
||||
cached_address = True
|
||||
|
||||
# Wait for the user to provide a multiworld server address
|
||||
if not address:
|
||||
|
|
@ -758,15 +740,14 @@ async def server_loop(ctx: Context, address=None):
|
|||
SERVER_RECONNECT_DELAY = START_RECONNECT_DELAY
|
||||
async for data in ctx.server.socket:
|
||||
for msg in loads(data):
|
||||
cmd, args = (msg[0], msg[1]) if len(msg) > 1 else (msg, None)
|
||||
await process_server_cmd(ctx, cmd, args)
|
||||
await process_server_cmd(ctx, msg[0], msg[1])
|
||||
logger.warning('Disconnected from multiworld server, type /connect to reconnect')
|
||||
except WebUI.WaitingForUiException:
|
||||
pass
|
||||
except ConnectionRefusedError:
|
||||
if cached_address:
|
||||
logger.error('Unable to connect to multiworld server at cached address. '
|
||||
'Please use the connect button above.')
|
||||
'Please use the connect button above.')
|
||||
else:
|
||||
logger.error('Connection refused by the multiworld server')
|
||||
except (OSError, websockets.InvalidURI):
|
||||
|
|
@ -791,6 +772,7 @@ async def server_loop(ctx: Context, address=None):
|
|||
asyncio.create_task(server_autoreconnect(ctx))
|
||||
SERVER_RECONNECT_DELAY *= 2
|
||||
|
||||
|
||||
async def server_autoreconnect(ctx: Context):
|
||||
# unfortunately currently broken. See: https://github.com/prompt-toolkit/python-prompt-toolkit/issues/1033
|
||||
# with prompt_toolkit.shortcuts.ProgressBar() as pb:
|
||||
|
|
@ -801,7 +783,8 @@ async def server_autoreconnect(ctx: Context):
|
|||
ctx.server_task = asyncio.create_task(server_loop(ctx))
|
||||
|
||||
|
||||
async def process_server_cmd(ctx: Context, cmd, args):
|
||||
async def process_server_cmd(ctx: Context, cmd: str, args: typing.Optional[dict]):
|
||||
|
||||
if cmd == 'RoomInfo':
|
||||
logger.info('--------------------------------')
|
||||
logger.info('Room Information:')
|
||||
|
|
@ -815,7 +798,7 @@ async def process_server_cmd(ctx: Context, cmd, args):
|
|||
logger.info("Server protocol tags: " + ", ".join(args["tags"]))
|
||||
if args['password']:
|
||||
logger.info('Password required')
|
||||
if "forfeit_mode" in args: # could also be version > 2.2.1, but going with implicit content here
|
||||
if "forfeit_mode" in args: # could also be version > 2.2.1, but going with implicit content here
|
||||
logging.info(f"Forfeit setting: {args['forfeit_mode']}")
|
||||
logging.info(f"Remaining setting: {args['remaining_mode']}")
|
||||
logging.info(f"A !hint costs {args['hint_cost']} points and you get {args['location_check_points']}"
|
||||
|
|
@ -839,64 +822,71 @@ async def process_server_cmd(ctx: Context, cmd, args):
|
|||
await server_auth(ctx, args['password'])
|
||||
|
||||
elif cmd == 'ConnectionRefused':
|
||||
if 'InvalidRom' in args:
|
||||
errors = args["errors"]
|
||||
if 'InvalidRom' in errors:
|
||||
if ctx.snes_socket is not None and not ctx.snes_socket.closed:
|
||||
asyncio.create_task(ctx.snes_socket.close())
|
||||
raise Exception('Invalid ROM detected, '
|
||||
'please verify that you have loaded the correct rom and reconnect your snes (/snes)')
|
||||
elif 'SlotAlreadyTaken' in args:
|
||||
elif 'SlotAlreadyTaken' in errors:
|
||||
Utils.persistent_store("servers", ctx.rom, ctx.server_address)
|
||||
raise Exception('Player slot already in use for that team')
|
||||
elif 'IncompatibleVersion' in args:
|
||||
elif 'IncompatibleVersion' in errors:
|
||||
raise Exception('Server reported your client version as incompatible')
|
||||
#last to check, recoverable problem
|
||||
elif 'InvalidPassword' in args:
|
||||
# last to check, recoverable problem
|
||||
elif 'InvalidPassword' in errors:
|
||||
logger.error('Invalid password')
|
||||
ctx.password = None
|
||||
await server_auth(ctx, True)
|
||||
else:
|
||||
raise Exception("Unknown connection errors: "+str(args))
|
||||
raise Exception("Unknown connection errors: " + str(errors))
|
||||
raise Exception('Connection refused by the multiworld host, no reason provided')
|
||||
|
||||
elif cmd == 'Connected':
|
||||
if ctx.send_unsafe:
|
||||
ctx.send_unsafe = False
|
||||
logger.info(f'Turning off sending of ALL location checks not declared as missing. If you want it on, please use /send_unsafe true')
|
||||
logger.info(
|
||||
f'Turning off sending of ALL location checks not declared as missing. If you want it on, please use /send_unsafe true')
|
||||
Utils.persistent_store("servers", ctx.rom, ctx.server_address)
|
||||
ctx.team, ctx.slot = args[0]
|
||||
ctx.player_names = {p: n for p, n in args[1]}
|
||||
ctx.team = args["team"]
|
||||
ctx.slot = args["slot"]
|
||||
ctx.player_names = {p: n for p, n in args["playernames"]}
|
||||
msgs = []
|
||||
if ctx.locations_checked:
|
||||
msgs.append(['LocationChecks', [Regions.lookup_name_to_id[loc] for loc in ctx.locations_checked]])
|
||||
msgs.append(['LocationChecks',
|
||||
{"locations": [Regions.lookup_name_to_id[loc] for loc in ctx.locations_checked]}])
|
||||
if ctx.locations_scouted:
|
||||
msgs.append(['LocationScouts', list(ctx.locations_scouted)])
|
||||
msgs.append(['LocationScouts', {"locations": list(ctx.locations_scouted)}])
|
||||
if msgs:
|
||||
await ctx.send_msgs(msgs)
|
||||
if ctx.finished_game:
|
||||
await send_finished_game(ctx)
|
||||
ctx.items_missing = args[2] if len(args) >= 3 else [] # Get the server side view of missing as of time of connecting.
|
||||
|
||||
# Get the server side view of missing as of time of connecting.
|
||||
# This list is used to only send to the server what is reported as ACTUALLY Missing.
|
||||
# This also serves to allow an easy visual of what locations were already checked previously
|
||||
# when /missing is used for the client side view of what is missing.
|
||||
if not ctx.items_missing:
|
||||
asyncio.create_task(ctx.send_msgs([['Say', '!missing']]))
|
||||
ctx.items_missing = args["missing_checks"]
|
||||
|
||||
elif cmd == 'ReceivedItems':
|
||||
start_index, items = args
|
||||
start_index = args["index"]
|
||||
|
||||
if start_index == 0:
|
||||
ctx.items_received = []
|
||||
elif start_index != len(ctx.items_received):
|
||||
sync_msg = [['Sync']]
|
||||
sync_msg = [['Sync', None]]
|
||||
if ctx.locations_checked:
|
||||
sync_msg.append(['LocationChecks', [Regions.lookup_name_to_id[loc] for loc in ctx.locations_checked]])
|
||||
sync_msg.append(['LocationChecks',
|
||||
{"locations": [Regions.lookup_name_to_id[loc] for loc in ctx.locations_checked]}])
|
||||
await ctx.send_msgs(sync_msg)
|
||||
if start_index == len(ctx.items_received):
|
||||
for item in items:
|
||||
|
||||
for item in args['items']:
|
||||
ctx.items_received.append(ReceivedItem(*item))
|
||||
ctx.watcher_event.set()
|
||||
|
||||
elif cmd == 'LocationInfo':
|
||||
for location, item, player in args:
|
||||
for item, location, player in args['locations']:
|
||||
if location not in ctx.locations_info:
|
||||
replacements = {0xA2: 'Small Key', 0x9D: 'Big Key', 0x8D: 'Compass', 0x7D: 'Map'}
|
||||
item_name = replacements.get(item, get_item_name_from_id(item))
|
||||
|
|
@ -906,19 +896,20 @@ async def process_server_cmd(ctx: Context, cmd, args):
|
|||
ctx.watcher_event.set()
|
||||
|
||||
elif cmd == 'ItemSent':
|
||||
player_sent, location, player_recvd, item = args
|
||||
ctx.ui_node.notify_item_sent(ctx.player_names[player_sent], ctx.player_names[player_recvd],
|
||||
get_item_name_from_id(item), get_location_name_from_address(location),
|
||||
player_sent == ctx.slot, player_recvd == ctx.slot)
|
||||
item = color(get_item_name_from_id(item), 'cyan' if player_sent != ctx.slot else 'green')
|
||||
player_sent = color(ctx.player_names[player_sent], 'yellow' if player_sent != ctx.slot else 'magenta')
|
||||
player_recvd = color(ctx.player_names[player_recvd], 'yellow' if player_recvd != ctx.slot else 'magenta')
|
||||
found = ReceivedItem(*args["item"])
|
||||
receiving_player = args["receiver"]
|
||||
ctx.ui_node.notify_item_sent(ctx.player_names[found.player], ctx.player_names[receiving_player],
|
||||
get_item_name_from_id(found.item), get_location_name_from_address(found.location),
|
||||
found.player == ctx.slot, receiving_player == ctx.slot)
|
||||
item = color(get_item_name_from_id(found.item), 'cyan' if found.player != ctx.slot else 'green')
|
||||
found_player = color(ctx.player_names[found.player], 'yellow' if found.player != ctx.slot else 'magenta')
|
||||
receiving_player = color(ctx.player_names[receiving_player], 'yellow' if receiving_player != ctx.slot else 'magenta')
|
||||
logging.info(
|
||||
'%s sent %s to %s (%s)' % (player_sent, item, player_recvd, color(get_location_name_from_address(location),
|
||||
'blue_bg', 'white')))
|
||||
'%s sent %s to %s (%s)' % (found_player, item, receiving_player,
|
||||
color(get_location_name_from_address(found.location), 'blue_bg', 'white')))
|
||||
|
||||
elif cmd == 'ItemFound':
|
||||
found = ReceivedItem(*args)
|
||||
found = ReceivedItem(*args["item"])
|
||||
ctx.ui_node.notify_item_found(ctx.player_names[found.player], get_item_name_from_id(found.item),
|
||||
get_location_name_from_address(found.location), found.player == ctx.slot)
|
||||
item = color(get_item_name_from_id(found.item), 'cyan' if found.player != ctx.slot else 'green')
|
||||
|
|
@ -926,9 +917,8 @@ async def process_server_cmd(ctx: Context, cmd, args):
|
|||
logging.info('%s found %s (%s)' % (player_sent, item, color(get_location_name_from_address(found.location),
|
||||
'blue_bg', 'white')))
|
||||
|
||||
|
||||
elif cmd == 'Hint':
|
||||
hints = [Utils.Hint(*hint) for hint in args]
|
||||
hints = [Utils.Hint(*hint) for hint in args["hints"]]
|
||||
for hint in hints:
|
||||
ctx.ui_node.send_hint(ctx.player_names[hint.finding_player], ctx.player_names[hint.receiving_player],
|
||||
get_item_name_from_id(hint.item), get_location_name_from_address(hint.location),
|
||||
|
|
@ -947,17 +937,24 @@ async def process_server_cmd(ctx: Context, cmd, args):
|
|||
text += " at " + color(hint.entrance, 'white_bg', 'black')
|
||||
logging.info(text + (f". {color('(found)', 'green_bg', 'black')} " if hint.found else "."))
|
||||
|
||||
elif cmd == "AliasUpdate":
|
||||
ctx.player_names = {p: n for p, n in args}
|
||||
elif cmd == "RoomUpdate":
|
||||
if "playernames" in args:
|
||||
ctx.player_names = {p: n for p, n in args["playernames"]}
|
||||
|
||||
elif cmd == 'Print':
|
||||
logger.info(args)
|
||||
logger.info(args["text"])
|
||||
|
||||
elif cmd == 'PrintHTML':
|
||||
logger.info(coloramaparser.get_colorama_text(args["text"]))
|
||||
|
||||
elif cmd == 'HintPointUpdate':
|
||||
ctx.hint_points = args[0]
|
||||
ctx.hint_points = args['points']
|
||||
|
||||
elif cmd == 'InvalidArguments':
|
||||
logger.warning(f"Invalid Arguments: {args['text']}")
|
||||
|
||||
else:
|
||||
logger.debug(f"unknown command {args}")
|
||||
logger.debug(f"unknown command {cmd}")
|
||||
|
||||
|
||||
def get_tags(ctx: Context):
|
||||
|
|
@ -981,11 +978,11 @@ async def server_auth(ctx: Context, password_requested):
|
|||
auth = base64.b64encode(ctx.rom).decode()
|
||||
await ctx.send_msgs([['Connect', {
|
||||
'password': ctx.password, 'rom': auth, 'version': Utils._version_tuple, 'tags': get_tags(ctx),
|
||||
'uuid': Utils.get_unique_identifier()
|
||||
'uuid': Utils.get_unique_identifier(), 'game': "ALTTP"
|
||||
}]])
|
||||
|
||||
|
||||
async def console_input(ctx : Context):
|
||||
async def console_input(ctx: Context):
|
||||
ctx.input_requests += 1
|
||||
return await ctx.input_queue.get()
|
||||
|
||||
|
|
@ -1081,7 +1078,8 @@ class ClientCommandProcessor(CommandProcessor):
|
|||
count += 1
|
||||
|
||||
if count:
|
||||
self.output(f"Found {count} missing location checks{f'. {checked_count} locations checks previously visited.' if checked_count else ''}")
|
||||
self.output(
|
||||
f"Found {count} missing location checks{f'. {checked_count} locations checks previously visited.' if checked_count else ''}")
|
||||
else:
|
||||
self.output("No missing location checks found.")
|
||||
return True
|
||||
|
|
@ -1115,13 +1113,14 @@ class ClientCommandProcessor(CommandProcessor):
|
|||
"""Force sending of locations the server did not specify was actually missing. WARNING: This may brick online trackers. Turned off on reconnect."""
|
||||
if toggle:
|
||||
self.ctx.send_unsafe = toggle.lower() in {"1", "true", "on"}
|
||||
logger.info(f'Turning {("on" if self.ctx.send_unsafe else "off")} the option to send ALL location checks to the multiserver.')
|
||||
logger.info(
|
||||
f'Turning {("on" if self.ctx.send_unsafe else "off")} the option to send ALL location checks to the multiserver.')
|
||||
else:
|
||||
logger.info("You must specify /send_unsafe true explicitly.")
|
||||
self.ctx.send_unsafe = False
|
||||
|
||||
def default(self, raw: str):
|
||||
asyncio.create_task(self.ctx.send_msgs([['Say', raw]]))
|
||||
asyncio.create_task(self.ctx.send_msgs([['Say', {"text": raw}]]))
|
||||
|
||||
|
||||
async def console_loop(ctx: Context):
|
||||
|
|
@ -1145,7 +1144,7 @@ async def console_loop(ctx: Context):
|
|||
await snes_flush_writes(ctx)
|
||||
|
||||
|
||||
async def track_locations(ctx : Context, roomid, roomdata):
|
||||
async def track_locations(ctx: Context, roomid, roomdata):
|
||||
new_locations = []
|
||||
|
||||
def new_check(location):
|
||||
|
|
@ -1155,13 +1154,14 @@ async def track_locations(ctx : Context, roomid, roomdata):
|
|||
|
||||
for location, (loc_roomid, loc_mask) in location_table_uw.items():
|
||||
try:
|
||||
if location not in ctx.unsafe_locations_checked and loc_roomid == roomid and (roomdata << 4) & loc_mask != 0:
|
||||
if location not in ctx.unsafe_locations_checked and loc_roomid == roomid and (
|
||||
roomdata << 4) & loc_mask != 0:
|
||||
new_check(location)
|
||||
except Exception as e:
|
||||
logger.exception(f"Exception: {e}")
|
||||
|
||||
uw_begin = 0x129
|
||||
uw_end = 0
|
||||
ow_end = uw_end = 0
|
||||
uw_unchecked = {}
|
||||
for location, (roomid, mask) in location_table_uw.items():
|
||||
if location not in ctx.unsafe_locations_checked:
|
||||
|
|
@ -1178,7 +1178,6 @@ async def track_locations(ctx : Context, roomid, roomdata):
|
|||
new_check(location)
|
||||
|
||||
ow_begin = 0x82
|
||||
ow_end = 0
|
||||
ow_unchecked = {}
|
||||
for location, screenid in location_table_ow.items():
|
||||
if location not in ctx.unsafe_locations_checked:
|
||||
|
|
@ -1204,7 +1203,7 @@ async def track_locations(ctx : Context, roomid, roomdata):
|
|||
misc_data = await snes_read(ctx, SAVEDATA_START + 0x3c6, 4)
|
||||
if misc_data is not None:
|
||||
for location, (offset, mask) in location_table_misc.items():
|
||||
assert(0x3c6 <= offset <= 0x3c9)
|
||||
assert (0x3c6 <= offset <= 0x3c9)
|
||||
if misc_data[offset - 0x3c6] & mask != 0 and location not in ctx.unsafe_locations_checked:
|
||||
new_check(location)
|
||||
|
||||
|
|
@ -1213,18 +1212,19 @@ async def track_locations(ctx : Context, roomid, roomdata):
|
|||
ctx.locations_checked.add(location)
|
||||
new_locations.append(Regions.lookup_name_to_id[location])
|
||||
|
||||
await ctx.send_msgs([['LocationChecks', new_locations]])
|
||||
if new_locations:
|
||||
await ctx.send_msgs([['LocationChecks', {"locations": new_locations}]])
|
||||
|
||||
|
||||
async def send_finished_game(ctx: Context):
|
||||
try:
|
||||
await ctx.send_msgs([['GameFinished', '']])
|
||||
await ctx.send_msgs([['StatusUpdate', {"status": CLIENT_GOAL}]])
|
||||
ctx.finished_game = True
|
||||
except Exception as ex:
|
||||
logger.exception(ex)
|
||||
|
||||
|
||||
async def game_watcher(ctx : Context):
|
||||
async def game_watcher(ctx: Context):
|
||||
prev_game_timer = 0
|
||||
perf_counter = time.perf_counter()
|
||||
while not ctx.exit_event.is_set():
|
||||
|
|
@ -1316,7 +1316,7 @@ async def game_watcher(ctx : Context):
|
|||
if scout_location > 0 and scout_location not in ctx.locations_scouted:
|
||||
ctx.locations_scouted.add(scout_location)
|
||||
logger.info(f'Scouting item at {list(Regions.location_table.keys())[scout_location - 1]}')
|
||||
await ctx.send_msgs([['LocationScouts', [scout_location]]])
|
||||
await ctx.send_msgs([['LocationScouts', {"locations": [scout_location]}]])
|
||||
await track_locations(ctx, roomid, roomdata)
|
||||
|
||||
|
||||
|
|
@ -1391,7 +1391,8 @@ async def main():
|
|||
parser.add_argument('--loglevel', default='info', choices=['debug', 'info', 'warning', 'error', 'critical'])
|
||||
parser.add_argument('--founditems', default=False, action='store_true',
|
||||
help='Show items found by other players for themselves.')
|
||||
parser.add_argument('--disable_web_ui', default=False, action='store_true', help="Turn off emitting a webserver for the webbrowser based user interface.")
|
||||
parser.add_argument('--disable_web_ui', default=False, action='store_true',
|
||||
help="Turn off emitting a webserver for the webbrowser based user interface.")
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(format='%(message)s', level=getattr(logging, args.loglevel.upper(), logging.INFO))
|
||||
|
||||
|
|
@ -1454,6 +1455,7 @@ async def main():
|
|||
|
||||
await input_task
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
colorama.init()
|
||||
loop = asyncio.get_event_loop()
|
||||
|
|
|
|||
147
MultiServer.py
147
MultiServer.py
|
|
@ -20,6 +20,7 @@ import ModuleUpdate
|
|||
ModuleUpdate.update()
|
||||
|
||||
import websockets
|
||||
import colorama
|
||||
import prompt_toolkit
|
||||
from prompt_toolkit.patch_stdout import patch_stdout
|
||||
from fuzzywuzzy import process as fuzzy_process
|
||||
|
|
@ -28,14 +29,11 @@ from worlds.alttp import Items, Regions
|
|||
import Utils
|
||||
from Utils import get_item_name_from_id, get_location_name_from_address, \
|
||||
ReceivedItem, _version_tuple, restricted_loads
|
||||
from NetUtils import Node, Endpoint
|
||||
from NetUtils import Node, Endpoint, CLIENT_GOAL
|
||||
|
||||
colorama.init()
|
||||
console_names = frozenset(set(Items.item_table) | set(Regions.location_table) | set(Items.item_name_groups) | set(Regions.key_drop_data))
|
||||
|
||||
CLIENT_PLAYING = 0
|
||||
CLIENT_GOAL = 1
|
||||
|
||||
|
||||
|
||||
class Client(Endpoint):
|
||||
version: typing.List[int] = [0, 0, 0]
|
||||
|
|
@ -196,8 +194,8 @@ class Context(Node):
|
|||
self.saving = enabled
|
||||
if self.saving:
|
||||
if not self.save_filename:
|
||||
self.save_filename = (self.data_filename[:-9] if self.data_filename.endswith('.archipelago') else (
|
||||
self.data_filename + '_')) + 'save'
|
||||
self.save_filename = (self.data_filename[:-11] if self.data_filename.endswith('.archipelago') else (
|
||||
self.data_filename + '_')) + 'apsave'
|
||||
try:
|
||||
with open(self.save_filename, 'rb') as f:
|
||||
save_data = restricted_loads(zlib.decompress(f.read()))
|
||||
|
|
@ -301,18 +299,18 @@ class Context(Node):
|
|||
|
||||
def notify_all(self, text):
|
||||
logging.info("Notice (all): %s" % text)
|
||||
self.broadcast_all([['Print', text]])
|
||||
self.broadcast_all([['Print', {"text": text}]])
|
||||
|
||||
def notify_client(self, client: Client, text: str):
|
||||
if not client.auth:
|
||||
return
|
||||
logging.info("Notice (Player %s in team %d): %s" % (client.name, client.team + 1, text))
|
||||
asyncio.create_task(self.send_msgs(client, [['Print', text]]))
|
||||
asyncio.create_task(self.send_msgs(client, [['Print', {"text": text}]]))
|
||||
|
||||
def notify_client_multiple(self, client: Client, texts: typing.List[str]):
|
||||
if not client.auth:
|
||||
return
|
||||
asyncio.create_task(self.send_msgs(client, [['Print', text] for text in texts]))
|
||||
asyncio.create_task(self.send_msgs(client, [['Print', {"text": text}] for text in texts]))
|
||||
|
||||
def broadcast_team(self, team, msgs):
|
||||
for client in self.endpoints:
|
||||
|
|
@ -332,8 +330,8 @@ class Context(Node):
|
|||
|
||||
# separated out, due to compatibilty between clients
|
||||
def notify_hints(ctx: Context, team: int, hints: typing.List[Utils.Hint]):
|
||||
cmd = dumps([["Hint", hints]]) # make sure it is a list, as it can be set internally
|
||||
texts = [['Print', format_hint(ctx, team, hint)] for hint in hints]
|
||||
cmd = dumps([["Hint", {"hints", hints}]])
|
||||
texts = [['PrintHTML', format_hint(ctx, team, hint)] for hint in hints]
|
||||
for _, text in texts:
|
||||
logging.info("Notice (Team #%d): %s" % (team + 1, text))
|
||||
for client in ctx.endpoints:
|
||||
|
|
@ -342,9 +340,9 @@ def notify_hints(ctx: Context, team: int, hints: typing.List[Utils.Hint]):
|
|||
|
||||
|
||||
def update_aliases(ctx: Context, team: int, client: typing.Optional[Client] = None):
|
||||
cmd = dumps([["AliasUpdate",
|
||||
[(key[1], ctx.get_aliased_name(*key)) for key, value in ctx.player_names.items() if
|
||||
key[0] == team]]])
|
||||
cmd = dumps([["RoomUpdate",
|
||||
{"playernames": [(key[1], ctx.get_aliased_name(*key)) for key, value in ctx.player_names.items() if
|
||||
key[0] == team]}]])
|
||||
if client is None:
|
||||
for client in ctx.endpoints:
|
||||
if client.team == team and client.auth:
|
||||
|
|
@ -361,13 +359,7 @@ async def server(websocket, path, ctx: Context):
|
|||
await on_client_connected(ctx, client)
|
||||
async for data in websocket:
|
||||
for msg in loads(data):
|
||||
if len(msg) == 1:
|
||||
cmd = msg
|
||||
args = None
|
||||
else:
|
||||
cmd = msg[0]
|
||||
args = msg[1]
|
||||
await process_client_cmd(ctx, client, cmd, args)
|
||||
await process_client_cmd(ctx, client, msg[0], msg[1])
|
||||
except Exception as e:
|
||||
if not isinstance(e, websockets.WebSocketException):
|
||||
logging.exception(e)
|
||||
|
|
@ -401,12 +393,6 @@ async def on_client_joined(ctx: Context, client: Client):
|
|||
ctx.notify_all(
|
||||
f"{ctx.get_aliased_name(client.team, client.slot)} (Team #{client.team + 1}) has joined the game. "
|
||||
f"Client({version_str}), {client.tags}).")
|
||||
if client.version < [2, 1, 0] and "auto" in ctx.forfeit_mode:
|
||||
ctx.notify_client(
|
||||
client,
|
||||
"Your client is too old to send game beaten information. "
|
||||
"The automatic forfeit feature will not work."
|
||||
)
|
||||
|
||||
ctx.client_connection_timers[client.team, client.slot] = datetime.datetime.now(datetime.timezone.utc)
|
||||
|
||||
|
|
@ -469,7 +455,8 @@ def send_new_items(ctx: Context):
|
|||
items = get_received_items(ctx, client.team, client.slot)
|
||||
if len(items) > client.send_index:
|
||||
asyncio.create_task(ctx.send_msgs(client, [
|
||||
['ReceivedItems', (client.send_index, tuplize_received_items(items)[client.send_index:])]]))
|
||||
['ReceivedItems', {"index": client.send_index,
|
||||
"items": tuplize_received_items(items)[client.send_index:]}]]))
|
||||
client.send_index = len(items)
|
||||
|
||||
|
||||
|
|
@ -488,7 +475,7 @@ def get_remaining(ctx: Context, team: int, slot: int) -> typing.List[int]:
|
|||
return sorted(items)
|
||||
|
||||
|
||||
def register_location_checks(ctx: Context, team: int, slot: int, locations):
|
||||
def register_location_checks(ctx: Context, team: int, slot: int, locations: typing.Iterable[int]):
|
||||
found_items = False
|
||||
new_locations = set(locations) - ctx.location_checks[team, slot]
|
||||
known_locations = set()
|
||||
|
|
@ -499,7 +486,7 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations):
|
|||
known_locations.add(location)
|
||||
target_item, target_player = ctx.locations[(location, slot)]
|
||||
if target_player != slot or slot in ctx.remote_items:
|
||||
found = False
|
||||
found: bool = False
|
||||
recvd_items = get_received_items(ctx, team, target_player)
|
||||
for recvd_item in recvd_items:
|
||||
if recvd_item.location == location and recvd_item.player == slot:
|
||||
|
|
@ -510,7 +497,9 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations):
|
|||
new_item = ReceivedItem(target_item, location, slot)
|
||||
recvd_items.append(new_item)
|
||||
if slot != target_player:
|
||||
ctx.broadcast_team(team, [['ItemSent', (slot, location, target_player, target_item)]])
|
||||
ctx.broadcast_team(team,
|
||||
[['ItemSent', {"item": new_item,
|
||||
"receiver" : target_player}]])
|
||||
logging.info('(Team #%d) %s sent %s to %s (%s)' % (
|
||||
team + 1, ctx.player_names[(team, slot)], get_item_name_from_id(target_item),
|
||||
ctx.player_names[(team, target_player)], get_location_name_from_address(location)))
|
||||
|
|
@ -520,20 +509,21 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations):
|
|||
for client in ctx.endpoints:
|
||||
if client.team == team and client.wants_item_notification:
|
||||
asyncio.create_task(
|
||||
ctx.send_msgs(client, [['ItemFound', (target_item, location, slot)]]))
|
||||
ctx.send_msgs(client, [['ItemFound',
|
||||
{"item": ReceivedItem(target_item, location, slot)}]]))
|
||||
ctx.location_checks[team, slot] |= known_locations
|
||||
send_new_items(ctx)
|
||||
|
||||
if found_items:
|
||||
for client in ctx.endpoints:
|
||||
if client.team == team and client.slot == slot:
|
||||
asyncio.create_task(ctx.send_msgs(client, [["HintPointUpdate", (get_client_points(ctx, client),)]]))
|
||||
asyncio.create_task(ctx.send_msgs(client, [["HintPointUpdate", {"points": get_client_points(ctx, client)}]]))
|
||||
ctx.save()
|
||||
|
||||
|
||||
def notify_team(ctx: Context, team: int, text: str):
|
||||
logging.info("Notice (Team #%d): %s" % (team + 1, text))
|
||||
ctx.broadcast_team(team, [['Print', text]])
|
||||
ctx.broadcast_team(team, [['Print', {"text": text}]])
|
||||
|
||||
|
||||
|
||||
|
|
@ -974,23 +964,26 @@ def get_client_points(ctx: Context, client: Client) -> int:
|
|||
ctx.hint_cost * ctx.hints_used[client.team, client.slot])
|
||||
|
||||
|
||||
async def process_client_cmd(ctx: Context, client: Client, cmd, args):
|
||||
async def process_client_cmd(ctx: Context, client: Client, cmd: str, args: typing.Optional[dict]):
|
||||
if type(cmd) is not str:
|
||||
await ctx.send_msgs(client, [['InvalidCmd']])
|
||||
await ctx.send_msgs(client, [['InvalidCmd', {"text": f"Command should be str, got {type(cmd)}"}]])
|
||||
return
|
||||
|
||||
if args is not None and type(args) != dict:
|
||||
await ctx.send_msgs(client, [['InvalidArguments',
|
||||
{'text': f'Expected Optional[dict], got {type(args)} for {cmd}'}]])
|
||||
return
|
||||
|
||||
if cmd == 'Connect':
|
||||
if not args or type(args) is not dict or \
|
||||
'password' not in args or type(args['password']) not in [str, type(None)] or \
|
||||
'rom' not in args or type(args['rom']) not in (list, str):
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'Connect']])
|
||||
if not args or 'password' not in args or type(args['password']) not in [str, type(None)] or \
|
||||
'game' not in args:
|
||||
await ctx.send_msgs(client, [['InvalidArguments', {'text': 'Connect'}]])
|
||||
return
|
||||
|
||||
errors = set()
|
||||
if ctx.password is not None and args['password'] != ctx.password:
|
||||
if ctx.password and args['password'] != ctx.password:
|
||||
errors.add('InvalidPassword')
|
||||
if type(args["rom"]) == list:
|
||||
args["rom"] = bytes(letter for letter in args["rom"]).decode()
|
||||
|
||||
if args['rom'] not in ctx.rom_names:
|
||||
logging.info((args["rom"], ctx.rom_names))
|
||||
errors.add('InvalidRom')
|
||||
|
|
@ -1012,28 +1005,29 @@ async def process_client_cmd(ctx: Context, client: Client, cmd, args):
|
|||
client.team = team
|
||||
client.slot = slot
|
||||
minver = Utils.Version(*(ctx.minimum_client_versions.get((team, slot), (0,0,0))))
|
||||
if minver > tuple(args.get('version', Client.version)):
|
||||
if minver > tuple(args['version']):
|
||||
errors.add('IncompatibleVersion')
|
||||
|
||||
if ctx.compatibility == 1 and "AP" not in args.get('tags', Client.tags):
|
||||
if ctx.compatibility == 1 and "AP" not in args['tags']:
|
||||
errors.add('IncompatibleVersion')
|
||||
#only exact version match allowed
|
||||
elif ctx.compatibility == 0 and tuple(args.get('version', Client.version)) != _version_tuple:
|
||||
elif ctx.compatibility == 0 and tuple(args['version']) != _version_tuple:
|
||||
errors.add('IncompatibleVersion')
|
||||
if errors:
|
||||
logging.info(f"A client connection was refused due to: {errors}")
|
||||
await ctx.send_msgs(client, [['ConnectionRefused', list(errors)]])
|
||||
await ctx.send_msgs(client, [['ConnectionRefused', {"errors": list(errors)}]])
|
||||
else:
|
||||
ctx.client_ids[client.team, client.slot] = args.get("uuid", None)
|
||||
ctx.client_ids[client.team, client.slot] = args["uuid"]
|
||||
client.auth = True
|
||||
client.version = args.get('version', Client.version)
|
||||
client.tags = args.get('tags', Client.tags)
|
||||
reply = [['Connected', [(client.team, client.slot),
|
||||
[(p, ctx.get_aliased_name(t, p)) for (t, p), n in ctx.player_names.items() if
|
||||
t == client.team], get_missing_checks(ctx, client)]]]
|
||||
client.version = args['version']
|
||||
client.tags = args['tags']
|
||||
reply = [['Connected', {"team": client.team, "slot": client.slot,
|
||||
"playernames": [(p, ctx.get_aliased_name(t, p)) for (t, p), n in
|
||||
ctx.player_names.items() if t == client.team],
|
||||
"missing_checks": get_missing_checks(ctx, client)}]]
|
||||
items = get_received_items(ctx, client.team, client.slot)
|
||||
if items:
|
||||
reply.append(['ReceivedItems', (0, tuplize_received_items(items))])
|
||||
reply.append(['ReceivedItems', {"index": 0, "items": tuplize_received_items(items)}])
|
||||
client.send_index = len(items)
|
||||
await ctx.send_msgs(client, reply)
|
||||
await on_client_joined(ctx, client)
|
||||
|
|
@ -1043,22 +1037,16 @@ async def process_client_cmd(ctx: Context, client: Client, cmd, args):
|
|||
items = get_received_items(ctx, client.team, client.slot)
|
||||
if items:
|
||||
client.send_index = len(items)
|
||||
await ctx.send_msgs(client, [['ReceivedItems', (0, tuplize_received_items(items))]])
|
||||
await ctx.send_msgs(client, [['ReceivedItems', {"index": 0, "items": tuplize_received_items(items)}]])
|
||||
|
||||
elif cmd == 'LocationChecks':
|
||||
if type(args) is not list:
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'LocationChecks']])
|
||||
return
|
||||
register_location_checks(ctx, client.team, client.slot, args)
|
||||
register_location_checks(ctx, client.team, client.slot, args["locations"])
|
||||
|
||||
elif cmd == 'LocationScouts':
|
||||
if type(args) is not list:
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'LocationScouts']])
|
||||
return
|
||||
locs = []
|
||||
for location in args:
|
||||
for location in args["locations"]:
|
||||
if type(location) is not int or 0 >= location > len(Regions.location_table):
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'LocationScouts']])
|
||||
await ctx.send_msgs(client, [['InvalidArguments', {"text": 'LocationScouts'}]])
|
||||
return
|
||||
loc_name = list(Regions.location_table.keys())[location - 1]
|
||||
target_item, target_player = ctx.locations[(Regions.location_table[loc_name][0], client.slot)]
|
||||
|
|
@ -1068,31 +1056,34 @@ async def process_client_cmd(ctx: Context, client: Client, cmd, args):
|
|||
if item_type:
|
||||
target_item = replacements.get(item_type[0], target_item)
|
||||
|
||||
locs.append([loc_name, location, target_item, target_player])
|
||||
locs.append([target_item, location, target_player])
|
||||
|
||||
# logging.info(f"{client.name} in team {client.team+1} scouted {', '.join([l[0] for l in locs])}")
|
||||
await ctx.send_msgs(client, [['LocationInfo', [l[1:] for l in locs]]])
|
||||
await ctx.send_msgs(client, [['LocationInfo', {'locations': locs}]])
|
||||
|
||||
elif cmd == 'UpdateTags':
|
||||
if not args or type(args) is not list:
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'UpdateTags']])
|
||||
await ctx.send_msgs(client, [['InvalidArguments', {"text": 'UpdateTags'}]])
|
||||
return
|
||||
client.tags = args
|
||||
|
||||
elif cmd == 'GameFinished':
|
||||
if ctx.client_game_state[client.team, client.slot] != CLIENT_GOAL:
|
||||
finished_msg = f'{ctx.get_aliased_name(client.team, client.slot)} (Team #{client.team + 1}) has found the triforce.'
|
||||
ctx.notify_all(finished_msg)
|
||||
ctx.client_game_state[client.team, client.slot] = CLIENT_GOAL
|
||||
if "auto" in ctx.forfeit_mode:
|
||||
forfeit_player(ctx, client.team, client.slot)
|
||||
elif cmd == 'StatusUpdate':
|
||||
current = ctx.client_game_state[client.team, client.slot]
|
||||
if current != CLIENT_GOAL: # can't undo goal completion
|
||||
if args["status"] == CLIENT_GOAL:
|
||||
finished_msg = f'{ctx.get_aliased_name(client.team, client.slot)} (Team #{client.team + 1}) has completed their goal.'
|
||||
ctx.notify_all(finished_msg)
|
||||
if "auto" in ctx.forfeit_mode:
|
||||
forfeit_player(ctx, client.team, client.slot)
|
||||
|
||||
ctx.client_game_state[client.team, client.slot] = args["status"]
|
||||
|
||||
if cmd == 'Say':
|
||||
if type(args) is not str or not args.isprintable():
|
||||
await ctx.send_msgs(client, [['InvalidArguments', 'Say']])
|
||||
if "text" not in args or type(args["text"]) is not str or not args["text"].isprintable():
|
||||
await ctx.send_msgs(client, [['InvalidArguments', {"text" : 'Say'}]])
|
||||
return
|
||||
|
||||
client.messageprocessor(args)
|
||||
client.messageprocessor(args["text"])
|
||||
|
||||
|
||||
class ServerCommandProcessor(CommonCommandProcessor):
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ from worlds.alttp.Main import get_seed, seeddigits
|
|||
from worlds.alttp.Items import item_name_groups, item_table
|
||||
from worlds.alttp import Bosses
|
||||
from worlds.alttp.Text import TextTable
|
||||
from worlds.alttp.Regions import location_table, key_drop_data
|
||||
|
||||
|
||||
def mystery_argparse():
|
||||
|
|
|
|||
62
NetUtils.py
62
NetUtils.py
|
|
@ -2,10 +2,12 @@ from __future__ import annotations
|
|||
import asyncio
|
||||
import logging
|
||||
import typing
|
||||
from html.parser import HTMLParser
|
||||
from json import loads, dumps
|
||||
|
||||
import websockets
|
||||
|
||||
|
||||
class Node:
|
||||
endpoints: typing.List
|
||||
dumper = staticmethod(dumps)
|
||||
|
|
@ -20,7 +22,7 @@ class Node:
|
|||
for endpoint in self.endpoints:
|
||||
asyncio.create_task(self.send_encoded_msgs(endpoint, msgs))
|
||||
|
||||
async def send_msgs(self, endpoint: Endpoint, msgs):
|
||||
async def send_msgs(self, endpoint: Endpoint, msgs: typing.Iterable[typing.Sequence[str, typing.Optional[dict]]]):
|
||||
if not endpoint.socket or not endpoint.socket.open or endpoint.socket.closed:
|
||||
return
|
||||
try:
|
||||
|
|
@ -51,3 +53,61 @@ class Endpoint:
|
|||
|
||||
async def disconnect(self):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class HTMLtoColoramaParser(HTMLParser):
|
||||
def get_colorama_text(self, input_text: str) -> str:
|
||||
self.feed(input_text)
|
||||
self.close()
|
||||
data = self.data
|
||||
self.reset()
|
||||
return data
|
||||
|
||||
def handle_data(self, data):
|
||||
self.data += data
|
||||
|
||||
def handle_starttag(self, tag, attrs):
|
||||
if tag in {"span", "div", "p"}:
|
||||
for attr in attrs:
|
||||
subtag, data = attr
|
||||
if subtag == "style":
|
||||
for subdata in data.split(";"):
|
||||
if subdata.startswith("color"):
|
||||
color = subdata.split(":", 1)[-1].strip()
|
||||
if color in color_codes:
|
||||
self.data += color_code(color)
|
||||
self.colored = tag
|
||||
|
||||
def handle_endtag(self, tag):
|
||||
if tag == self.colored:
|
||||
self.colored = False
|
||||
self.data += color_code("reset")
|
||||
|
||||
def reset(self):
|
||||
super(HTMLtoColoramaParser, self).reset()
|
||||
self.data = ""
|
||||
self.colored = False
|
||||
|
||||
def close(self):
|
||||
super(HTMLtoColoramaParser, self).close()
|
||||
if self.colored:
|
||||
self.handle_endtag(self.colored)
|
||||
|
||||
|
||||
color_codes = {'reset': 0, 'bold': 1, 'underline': 4, 'black': 30, 'red': 31, 'green': 32, 'yellow': 33, 'blue': 34,
|
||||
'magenta': 35, 'cyan': 36, 'white': 37, 'black_bg': 40, 'red_bg': 41, 'green_bg': 42, 'yellow_bg': 43,
|
||||
'blue_bg': 44, 'purple_bg': 45, 'cyan_bg': 46, 'white_bg': 47}
|
||||
|
||||
|
||||
def color_code(*args):
|
||||
return '\033[' + ';'.join([str(color_codes[arg]) for arg in args]) + 'm'
|
||||
|
||||
|
||||
def color(text, *args):
|
||||
return color_code(*args) + text + color_code('reset')
|
||||
|
||||
|
||||
CLIENT_UNKNOWN = 0
|
||||
CLIENT_READY = 10
|
||||
CLIENT_PLAYING = 20
|
||||
CLIENT_GOAL = 30
|
||||
2
Utils.py
2
Utils.py
|
|
@ -384,6 +384,8 @@ class RestrictedUnpickler(pickle.Unpickler):
|
|||
def find_class(self, module, name):
|
||||
if module == "builtins" and name in safe_builtins:
|
||||
return getattr(builtins, name)
|
||||
if module == "Utils" and name in {"ReceivedItem"}:
|
||||
return globals()[name]
|
||||
# Forbid everything else.
|
||||
raise pickle.UnpicklingError("global '%s.%s' is forbidden" %
|
||||
(module, name))
|
||||
|
|
|
|||
|
|
@ -1,11 +1,24 @@
|
|||
from typing import NamedTuple, Union
|
||||
|
||||
import logging
|
||||
|
||||
class PlandoItem(NamedTuple):
|
||||
item: str
|
||||
location: str
|
||||
world: Union[bool, str] = False # False -> own world, True -> not own world
|
||||
from_pool: bool = True # if item should be removed from item pool
|
||||
force: str = 'silent' # false -> warns if item not successfully placed. true -> errors out on failure to place item.
|
||||
|
||||
def warn(self, warning: str):
|
||||
if self.force in ['true', 'fail', 'failure', 'none', 'false', 'warn', 'warning']:
|
||||
logging.warning(f'{warning}')
|
||||
else:
|
||||
logging.debug(f'{warning}')
|
||||
|
||||
def failed(self, warning: str, exception=Exception):
|
||||
if self.force in ['true', 'fail', 'failure']:
|
||||
raise exception(warning)
|
||||
else:
|
||||
self.warn(warning)
|
||||
|
||||
|
||||
class PlandoConnection(NamedTuple):
|
||||
|
|
|
|||
Loading…
Reference in New Issue