monorail
/
Archipelago
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

FactorioClient.py formatting

This commit is contained in:
Fabian Dill 2021-06-06 17:50:48 +02:00
parent dd3c612dec
commit 4555b77204
1 changed files with 15 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
FactorioClient.py
View File

@ -1,8 +1,10 @@
from __future__ import annotations
import os import os
import logging import logging
import json import json
import string import string
import copy import copy
import sys
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
import colorama import colorama
@ -21,7 +23,6 @@ rcon_port = 24242
rcon_password = ''.join(random.choice(string.ascii_letters) for x in range(32)) rcon_password = ''.join(random.choice(string.ascii_letters) for x in range(32))
save_name = "Archipelago" save_name = "Archipelago"
logging.basicConfig(format='[%(name)s]: %(message)s', level=logging.INFO) logging.basicConfig(format='[%(name)s]: %(message)s', level=logging.INFO)
options = Utils.get_options() options = Utils.get_options()
executable = options["factorio_options"]["executable"] executable = options["factorio_options"]["executable"]
@ -34,13 +35,14 @@ if not os.path.exists(executable):
else: else:
raise FileNotFoundError(executable) raise FileNotFoundError(executable)
import sys
server_args = (save_name, "--rcon-port", rcon_port, "--rcon-password", rcon_password, *sys.argv[1:]) server_args = (save_name, "--rcon-port", rcon_port, "--rcon-password", rcon_password, *sys.argv[1:])
threadpool = ThreadPoolExecutor(10) thread_pool = ThreadPoolExecutor(10)
class FactorioCommandProcessor(ClientCommandProcessor): class FactorioCommandProcessor(ClientCommandProcessor):
ctx: FactorioContext
@mark_raw @mark_raw
def _cmd_factorio(self, text: str) -> bool: def _cmd_factorio(self, text: str) -> bool:
"""Send the following command to the bound Factorio Server.""" """Send the following command to the bound Factorio Server."""
@ -83,7 +85,8 @@ class FactorioContext(CommonContext):
logger.info(args["text"]) logger.info(args["text"])
if self.rcon_client: if self.rcon_client:
cleaned_text = args['text'].replace('"', '') cleaned_text = args['text'].replace('"', '')
self.rcon_client.send_command(f"/sc game.print(\"[font=default-large-bold]Archipelago:[/font] {cleaned_text}\")") self.rcon_client.send_command(f"/sc game.print(\"[font=default-large-bold]Archipelago:[/font] "
f"{cleaned_text}\")")
def on_print_json(self, args: dict): def on_print_json(self, args: dict):
if not self.found_items and args.get("type", None) == "ItemSend" and args["receiving"] == args["sending"]: if not self.found_items and args.get("type", None) == "ItemSend" and args["receiving"] == args["sending"]:
@ -94,7 +97,9 @@ class FactorioContext(CommonContext):
text = self.factorio_json_text_parser(args["data"]) text = self.factorio_json_text_parser(args["data"])
logger.info(text) logger.info(text)
cleaned_text = text.replace('"', '') cleaned_text = text.replace('"', '')
self.rcon_client.send_command(f"/sc game.print(\"[font=default-large-bold]Archipelago:[/font] {cleaned_text}\")") self.rcon_client.send_command(f"/sc game.print(\"[font=default-large-bold]Archipelago:[/font] "
f"{cleaned_text}\")")
async def game_watcher(ctx: FactorioContext, bridge_file: str): async def game_watcher(ctx: FactorioContext, bridge_file: str):
bridge_logger = logging.getLogger("FactorioWatcher") bridge_logger = logging.getLogger("FactorioWatcher")
@ -120,8 +125,9 @@ async def game_watcher(ctx: FactorioContext, bridge_file: str):
ctx.finished_game = True ctx.finished_game = True
if ctx.locations_checked != research_data: if ctx.locations_checked != research_data:
bridge_logger.info(f"New researches done: " bridge_logger.info(
f"{[lookup_id_to_name[rid] for rid in research_data - ctx.locations_checked]}") f"New researches done: "
f"{[lookup_id_to_name[rid] for rid in research_data - ctx.locations_checked]}")
ctx.locations_checked = research_data ctx.locations_checked = research_data
await ctx.send_msgs([{"cmd": 'LocationChecks', "locations": tuple(research_data)}]) await ctx.send_msgs([{"cmd": 'LocationChecks', "locations": tuple(research_data)}])
await asyncio.sleep(1) await asyncio.sleep(1)
@ -184,7 +190,7 @@ async def factorio_server_watcher(ctx: FactorioContext):
if os.path.exists(bridge_file): if os.path.exists(bridge_file):
os.remove(bridge_file) os.remove(bridge_file)
logging.info(f"Bridge File Path: {bridge_file}") logging.info(f"Bridge File Path: {bridge_file}")
progression_watcher= asyncio.create_task( progression_watcher = asyncio.create_task(
game_watcher(ctx, bridge_file), name="FactorioProgressionWatcher") game_watcher(ctx, bridge_file), name="FactorioProgressionWatcher")
if not ctx.awaiting_bridge and "Archipelago Bridge File written for game tick " in msg: if not ctx.awaiting_bridge and "Archipelago Bridge File written for game tick " in msg:
ctx.awaiting_bridge = True ctx.awaiting_bridge = True
@ -202,7 +208,6 @@ async def factorio_server_watcher(ctx: FactorioContext):
ctx.send_index += 1 ctx.send_index += 1
await asyncio.sleep(1) await asyncio.sleep(1)
except Exception as e: except Exception as e:
logging.exception(e) logging.exception(e)
logging.error("Aborted Factorio Server Bridge") logging.error("Aborted Factorio Server Bridge")
@ -261,4 +266,4 @@ class FactorioJSONtoTextParser(JSONtoTextParser):
elif color == "magenta": elif color == "magenta":
node["text"] = f"[color=pink]{node['text']}[/color]" node["text"] = f"[color=pink]{node['text']}[/color]"
return self._handle_text(node) return self._handle_text(node)
return self._handle_text(node) return self._handle_text(node)