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

SC2: client revamp (#967) 

SC2 client now relies almost entirely on the map file and server for the locations and just facilitates them, should make it significantly more resilient to objectives being added or removed


* SC2: fix client crash on printjson messages with more [ than ]

* SC2: move text to queue, that actually clears memory

* SC2: Announce which mission is being loaded


Co-authored-by: black-sliver <59490463+black-sliver@users.noreply.github.com>
This commit is contained in:
Fabian Dill 2022-08-31 20:55:15 +02:00 committed by GitHub
parent 8da1cfeeb7
commit c617bba959
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 187 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
CommonClient.py
View File

@ -152,8 +152,9 @@ class CommonContext:
# locations # locations
locations_checked: typing.Set[int] # local state locations_checked: typing.Set[int] # local state
locations_scouted: typing.Set[int] locations_scouted: typing.Set[int]
missing_locations: typing.Set[int] missing_locations: typing.Set[int] # server state
checked_locations: typing.Set[int] # server state checked_locations: typing.Set[int] # server state
server_locations: typing.Set[int] # all locations the server knows of, missing_location | checked_locations
locations_info: typing.Dict[int, NetworkItem] locations_info: typing.Dict[int, NetworkItem]
# internals # internals
@ -184,8 +185,9 @@ class CommonContext:
self.locations_checked = set() # local state self.locations_checked = set() # local state
self.locations_scouted = set() self.locations_scouted = set()
self.items_received = [] self.items_received = []
self.missing_locations = set() self.missing_locations = set() # server state
self.checked_locations = set() # server state self.checked_locations = set() # server state
self.server_locations = set() # all locations the server knows of, missing_location | checked_locations
self.locations_info = {} self.locations_info = {}
self.input_queue = asyncio.Queue() self.input_queue = asyncio.Queue()
@ -634,6 +636,7 @@ async def process_server_cmd(ctx: CommonContext, args: dict):
# when /missing is used for the client side view of what is missing. # when /missing is used for the client side view of what is missing.
ctx.missing_locations = set(args["missing_locations"]) ctx.missing_locations = set(args["missing_locations"])
ctx.checked_locations = set(args["checked_locations"]) ctx.checked_locations = set(args["checked_locations"])
ctx.server_locations = ctx.missing_locations | ctx. checked_locations
elif cmd == 'ReceivedItems': elif cmd == 'ReceivedItems':
start_index = args["index"] start_index = args["index"]

419
Starcraft2Client.py
View File

@ -1,31 +1,31 @@
from __future__ import annotations from __future__ import annotations
import multiprocessing
import logging
import asyncio import asyncio
import copy
import ctypes
import logging
import multiprocessing
import os.path import os.path
import re
import sys
import typing
import queue
from pathlib import Path
import nest_asyncio import nest_asyncio
import sc2 import sc2
from sc2.main import run_game
from sc2.data import Race
from sc2.bot_ai import BotAI from sc2.bot_ai import BotAI
from sc2.data import Race
from sc2.main import run_game
from sc2.player import Bot from sc2.player import Bot
from worlds.sc2wol.Regions import MissionInfo from MultiServer import mark_raw
from worlds.sc2wol.MissionTables import lookup_id_to_mission from Utils import init_logging, is_windows
from worlds.sc2wol import SC2WoLWorld
from worlds.sc2wol.Items import lookup_id_to_name, item_table from worlds.sc2wol.Items import lookup_id_to_name, item_table
from worlds.sc2wol.Locations import SC2WOL_LOC_ID_OFFSET from worlds.sc2wol.Locations import SC2WOL_LOC_ID_OFFSET
from worlds.sc2wol import SC2WoLWorld from worlds.sc2wol.MissionTables import lookup_id_to_mission
from worlds.sc2wol.Regions import MissionInfo
from pathlib import Path
import re
from MultiServer import mark_raw
import ctypes
import sys
from Utils import init_logging, is_windows
if __name__ == "__main__": if __name__ == "__main__":
init_logging("SC2Client", exception_logger="Client") init_logging("SC2Client", exception_logger="Client")
@ -35,10 +35,12 @@ sc2_logger = logging.getLogger("Starcraft2")
import colorama import colorama
from NetUtils import * from NetUtils import ClientStatus, RawJSONtoTextParser
from CommonClient import CommonContext, server_loop, ClientCommandProcessor, gui_enabled, get_base_parser from CommonClient import CommonContext, server_loop, ClientCommandProcessor, gui_enabled, get_base_parser
nest_asyncio.apply() nest_asyncio.apply()
max_bonus: int = 8
victory_modulo: int = 100
class StarcraftClientProcessor(ClientCommandProcessor): class StarcraftClientProcessor(ClientCommandProcessor):
@ -98,13 +100,13 @@ class StarcraftClientProcessor(ClientCommandProcessor):
def _cmd_available(self) -> bool: def _cmd_available(self) -> bool:
"""Get what missions are currently available to play""" """Get what missions are currently available to play"""
request_available_missions(self.ctx.checked_locations, self.ctx.mission_req_table, self.ctx.ui) request_available_missions(self.ctx)
return True return True
def _cmd_unfinished(self) -> bool: def _cmd_unfinished(self) -> bool:
"""Get what missions are currently available to play and have not had all locations checked""" """Get what missions are currently available to play and have not had all locations checked"""
request_unfinished_missions(self.ctx.checked_locations, self.ctx.mission_req_table, self.ctx.ui, self.ctx) request_unfinished_missions(self.ctx)
return True return True
@mark_raw @mark_raw
@ -125,18 +127,19 @@ class SC2Context(CommonContext):
items_handling = 0b111 items_handling = 0b111
difficulty = -1 difficulty = -1
all_in_choice = 0 all_in_choice = 0
mission_req_table = None mission_req_table: typing.Dict[str, MissionInfo] = {}
items_rec_to_announce = [] announcements = queue.Queue()
rec_announce_pos = 0
items_sent_to_announce = []
sent_announce_pos = 0
announcements = []
announcement_pos = 0
sc2_run_task: typing.Optional[asyncio.Task] = None sc2_run_task: typing.Optional[asyncio.Task] = None
missions_unlocked = False missions_unlocked: bool = False # allow launching missions ignoring requirements
current_tooltip = None current_tooltip = None
last_loc_list = None last_loc_list = None
difficulty_override = -1 difficulty_override = -1
mission_id_to_location_ids: typing.Dict[int, typing.List[int]] = {}
raw_text_parser: RawJSONtoTextParser
def __init__(self, *args, **kwargs):
super(SC2Context, self).__init__(*args, **kwargs)
self.raw_text_parser = RawJSONtoTextParser(self)
async def server_auth(self, password_requested: bool = False): async def server_auth(self, password_requested: bool = False):
if password_requested and not self.password: if password_requested and not self.password:
@ -149,30 +152,32 @@ class SC2Context(CommonContext):
self.difficulty = args["slot_data"]["game_difficulty"] self.difficulty = args["slot_data"]["game_difficulty"]
self.all_in_choice = args["slot_data"]["all_in_map"] self.all_in_choice = args["slot_data"]["all_in_map"]
slot_req_table = args["slot_data"]["mission_req"] slot_req_table = args["slot_data"]["mission_req"]
self.mission_req_table = {} self.mission_req_table = {
# Compatibility for 0.3.2 server data. mission: MissionInfo(**slot_req_table[mission]) for mission in slot_req_table
if "category" not in next(iter(slot_req_table)): }
for i, mission_data in enumerate(slot_req_table.values()):
mission_data["category"] = wol_default_categories[i] self.build_location_to_mission_mapping()
for mission in slot_req_table:
self.mission_req_table[mission] = MissionInfo(**slot_req_table[mission])
# Look for and set SC2PATH. # Look for and set SC2PATH.
# check_game_install_path() returns True if and only if it finds + sets SC2PATH. # check_game_install_path() returns True if and only if it finds + sets SC2PATH.
if "SC2PATH" not in os.environ and check_game_install_path(): if "SC2PATH" not in os.environ and check_game_install_path():
check_mod_install() check_mod_install()
if cmd in {"PrintJSON"}: def on_print_json(self, args: dict):
if "receiving" in args: if "receiving" in args and self.slot_concerns_self(args["receiving"]):
if self.slot_concerns_self(args["receiving"]): relevant = True
self.announcements.append(args["data"]) elif "item" in args and self.slot_concerns_self(args["item"].player):
return relevant = True
if "item" in args: else:
if self.slot_concerns_self(args["item"].player): relevant = False
self.announcements.append(args["data"])
if relevant:
self.announcements.put(self.raw_text_parser(copy.deepcopy(args["data"])))
super(SC2Context, self).on_print_json(args)
def run_gui(self): def run_gui(self):
from kvui import GameManager, HoverBehavior, ServerToolTip, fade_in_animation from kvui import GameManager, HoverBehavior, ServerToolTip
from kivy.app import App from kivy.app import App
from kivy.clock import Clock from kivy.clock import Clock
from kivy.uix.tabbedpanel import TabbedPanelItem from kivy.uix.tabbedpanel import TabbedPanelItem
@ -190,6 +195,7 @@ class SC2Context(CommonContext):
class MissionButton(HoverableButton): class MissionButton(HoverableButton):
tooltip_text = StringProperty("Test") tooltip_text = StringProperty("Test")
ctx: SC2Context
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(HoverableButton, self).__init__(*args, **kwargs) super(HoverableButton, self).__init__(*args, **kwargs)
@ -210,10 +216,7 @@ class SC2Context(CommonContext):
self.ctx.current_tooltip = self.layout self.ctx.current_tooltip = self.layout
def on_leave(self): def on_leave(self):
if self.ctx.current_tooltip: self.ctx.ui.clear_tooltip()
App.get_running_app().root.remove_widget(self.ctx.current_tooltip)
self.ctx.current_tooltip = None
@property @property
def ctx(self) -> CommonContext: def ctx(self) -> CommonContext:
@ -235,13 +238,20 @@ class SC2Context(CommonContext):
mission_panel = None mission_panel = None
last_checked_locations = {} last_checked_locations = {}
mission_id_to_button = {} mission_id_to_button = {}
launching = False launching: typing.Union[bool, int] = False # if int -> mission ID
refresh_from_launching = True refresh_from_launching = True
first_check = True first_check = True
ctx: SC2Context
def __init__(self, ctx): def __init__(self, ctx):
super().__init__(ctx) super().__init__(ctx)
def clear_tooltip(self):
if self.ctx.current_tooltip:
App.get_running_app().root.remove_widget(self.ctx.current_tooltip)
self.ctx.current_tooltip = None
def build(self): def build(self):
container = super().build() container = super().build()
@ -267,12 +277,7 @@ class SC2Context(CommonContext):
self.mission_id_to_button = {} self.mission_id_to_button = {}
categories = {} categories = {}
available_missions = [] available_missions, unfinished_missions = calc_unfinished_missions(self.ctx)
unfinished_locations = initialize_blank_mission_dict(self.ctx.mission_req_table)
unfinished_missions = calc_unfinished_missions(self.ctx.checked_locations,
self.ctx.mission_req_table,
self.ctx, available_missions=available_missions,
unfinished_locations=unfinished_locations)
# separate missions into categories # separate missions into categories
for mission in self.ctx.mission_req_table: for mission in self.ctx.mission_req_table:
@ -283,7 +288,8 @@ class SC2Context(CommonContext):
for category in categories: for category in categories:
category_panel = MissionCategory() category_panel = MissionCategory()
category_panel.add_widget(Label(text=category, size_hint_y=None, height=50, outline_width=1)) category_panel.add_widget(
Label(text=category, size_hint_y=None, height=50, outline_width=1))
# Map is completed # Map is completed
for mission in categories[category]: for mission in categories[category]:
@ -295,7 +301,9 @@ class SC2Context(CommonContext):
text = f"[color=6495ED]{text}[/color]" text = f"[color=6495ED]{text}[/color]"
tooltip = f"Uncollected locations:\n" tooltip = f"Uncollected locations:\n"
tooltip += "\n".join(location for location in unfinished_locations[mission]) tooltip += "\n".join([self.ctx.location_names[loc] for loc in
self.ctx.locations_for_mission(mission)
if loc in self.ctx.missing_locations])
elif mission in available_missions: elif mission in available_missions:
text = f"[color=FFFFFF]{text}[/color]" text = f"[color=FFFFFF]{text}[/color]"
# Map requirements not met # Map requirements not met
@ -325,13 +333,17 @@ class SC2Context(CommonContext):
self.refresh_from_launching = False self.refresh_from_launching = False
self.mission_panel.clear_widgets() self.mission_panel.clear_widgets()
self.mission_panel.add_widget(Label(text="Launching Mission")) self.mission_panel.add_widget(Label(text="Launching Mission: " +
lookup_id_to_mission[self.launching]))
if self.ctx.ui:
self.ctx.ui.clear_tooltip()
def mission_callback(self, button): def mission_callback(self, button):
if not self.launching: if not self.launching:
self.ctx.play_mission(list(self.mission_id_to_button.keys()) mission_id: int = list(self.mission_id_to_button.values()).index(button)
[list(self.mission_id_to_button.values()).index(button)]) self.ctx.play_mission(list(self.mission_id_to_button)
self.launching = True [mission_id])
self.launching = mission_id
Clock.schedule_once(self.finish_launching, 10) Clock.schedule_once(self.finish_launching, 10)
def finish_launching(self, dt): def finish_launching(self, dt):
@ -349,7 +361,7 @@ class SC2Context(CommonContext):
def play_mission(self, mission_id): def play_mission(self, mission_id):
if self.missions_unlocked or \ if self.missions_unlocked or \
is_mission_available(mission_id, self.checked_locations, self.mission_req_table): is_mission_available(self, mission_id):
if self.sc2_run_task: if self.sc2_run_task:
if not self.sc2_run_task.done(): if not self.sc2_run_task.done():
sc2_logger.warning("Starcraft 2 Client is still running!") sc2_logger.warning("Starcraft 2 Client is still running!")
@ -364,6 +376,23 @@ class SC2Context(CommonContext):
f"{lookup_id_to_mission[mission_id]} is not currently unlocked. " f"{lookup_id_to_mission[mission_id]} is not currently unlocked. "
f"Use /unfinished or /available to see what is available.") f"Use /unfinished or /available to see what is available.")
def build_location_to_mission_mapping(self):
mission_id_to_location_ids: typing.Dict[int, typing.Set[int]] = {
mission_info.id: set() for mission_info in self.mission_req_table.values()
}
for loc in self.server_locations:
mission_id, objective = divmod(loc - SC2WOL_LOC_ID_OFFSET, victory_modulo)
mission_id_to_location_ids[mission_id].add(objective)
self.mission_id_to_location_ids = {mission_id: sorted(objectives) for mission_id, objectives in
mission_id_to_location_ids.items()}
def locations_for_mission(self, mission: str):
mission_id: int = self.mission_req_table[mission].id
objectives = self.mission_id_to_location_ids[self.mission_req_table[mission].id]
for objective in objectives:
yield SC2WOL_LOC_ID_OFFSET + mission_id * 100 + objective
async def main(): async def main():
multiprocessing.freeze_support() multiprocessing.freeze_support()
@ -459,11 +488,7 @@ def calc_difficulty(difficulty):
return 'X' return 'X'
async def starcraft_launch(ctx: SC2Context, mission_id): async def starcraft_launch(ctx: SC2Context, mission_id: int):
ctx.rec_announce_pos = len(ctx.items_rec_to_announce)
ctx.sent_announce_pos = len(ctx.items_sent_to_announce)
ctx.announcements_pos = len(ctx.announcements)
sc2_logger.info(f"Launching {lookup_id_to_mission[mission_id]}. If game does not launch check log file for errors.") sc2_logger.info(f"Launching {lookup_id_to_mission[mission_id]}. If game does not launch check log file for errors.")
with DllDirectory(None): with DllDirectory(None):
@ -472,32 +497,29 @@ async def starcraft_launch(ctx: SC2Context, mission_id):
class ArchipelagoBot(sc2.bot_ai.BotAI): class ArchipelagoBot(sc2.bot_ai.BotAI):
game_running = False game_running: bool = False
mission_completed = False mission_completed: bool = False
first_bonus = False boni: typing.List[bool]
second_bonus = False setup_done: bool
third_bonus = False ctx: SC2Context
fourth_bonus = False mission_id: int
fifth_bonus = False
sixth_bonus = False
seventh_bonus = False
eight_bonus = False
ctx: SC2Context = None
mission_id = 0
can_read_game = False can_read_game = False
last_received_update = 0 last_received_update: int = 0
def __init__(self, ctx: SC2Context, mission_id): def __init__(self, ctx: SC2Context, mission_id):
self.setup_done = False
self.ctx = ctx self.ctx = ctx
self.mission_id = mission_id self.mission_id = mission_id
self.boni = [False for _ in range(max_bonus)]
super(ArchipelagoBot, self).__init__() super(ArchipelagoBot, self).__init__()
async def on_step(self, iteration: int): async def on_step(self, iteration: int):
game_state = 0 game_state = 0
if iteration == 0: if not self.setup_done:
self.setup_done = True
start_items = calculate_items(self.ctx.items_received) start_items = calculate_items(self.ctx.items_received)
if self.ctx.difficulty_override >= 0: if self.ctx.difficulty_override >= 0:
difficulty = calc_difficulty(self.ctx.difficulty_override) difficulty = calc_difficulty(self.ctx.difficulty_override)
@ -511,36 +533,10 @@ class ArchipelagoBot(sc2.bot_ai.BotAI):
self.last_received_update = len(self.ctx.items_received) self.last_received_update = len(self.ctx.items_received)
else: else:
if self.ctx.announcement_pos < len(self.ctx.announcements): if not self.ctx.announcements.empty():
index = 0 message = self.ctx.announcements.get(timeout=1)
message = ""
while index < len(self.ctx.announcements[self.ctx.announcement_pos]):
message += self.ctx.announcements[self.ctx.announcement_pos][index]["text"]
index += 1
index = 0
start_rem_pos = -1
# Remove unneeded [Color] tags
while index < len(message):
if message[index] == '[':
start_rem_pos = index
index += 1
elif message[index] == ']' and start_rem_pos > -1:
temp_msg = ""
if start_rem_pos > 0:
temp_msg = message[:start_rem_pos]
if index < len(message) - 1:
temp_msg += message[index + 1:]
message = temp_msg
index += start_rem_pos - index
start_rem_pos = -1
else:
index += 1
await self.chat_send("SendMessage " + message) await self.chat_send("SendMessage " + message)
self.ctx.announcement_pos += 1 self.ctx.announcements.task_done()
# Archipelago reads the health # Archipelago reads the health
for unit in self.all_own_units(): for unit in self.all_own_units():
@ -568,169 +564,97 @@ class ArchipelagoBot(sc2.bot_ai.BotAI):
if game_state & (1 << 1) and not self.mission_completed: if game_state & (1 << 1) and not self.mission_completed:
if self.mission_id != 29: if self.mission_id != 29:
print("Mission Completed") print("Mission Completed")
await self.ctx.send_msgs([ await self.ctx.send_msgs(
{"cmd": 'LocationChecks', "locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id]}]) [{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + victory_modulo * self.mission_id]}])
self.mission_completed = True self.mission_completed = True
else: else:
print("Game Complete") print("Game Complete")
await self.ctx.send_msgs([{"cmd": 'StatusUpdate', "status": ClientStatus.CLIENT_GOAL}]) await self.ctx.send_msgs([{"cmd": 'StatusUpdate', "status": ClientStatus.CLIENT_GOAL}])
self.mission_completed = True self.mission_completed = True
if game_state & (1 << 2) and not self.first_bonus: for x, completed in enumerate(self.boni):
print("1st Bonus Collected") if not completed and game_state & (1 << (x + 2)):
await self.ctx.send_msgs( await self.ctx.send_msgs(
[{"cmd": 'LocationChecks', [{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 1]}]) "locations": [SC2WOL_LOC_ID_OFFSET + victory_modulo * self.mission_id + x + 1]}])
self.first_bonus = True self.boni[x] = True
if not self.second_bonus and game_state & (1 << 3):
print("2nd Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 2]}])
self.second_bonus = True
if not self.third_bonus and game_state & (1 << 4):
print("3rd Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 3]}])
self.third_bonus = True
if not self.fourth_bonus and game_state & (1 << 5):
print("4th Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 4]}])
self.fourth_bonus = True
if not self.fifth_bonus and game_state & (1 << 6):
print("5th Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 5]}])
self.fifth_bonus = True
if not self.sixth_bonus and game_state & (1 << 7):
print("6th Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 6]}])
self.sixth_bonus = True
if not self.seventh_bonus and game_state & (1 << 8):
print("6th Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 7]}])
self.seventh_bonus = True
if not self.eight_bonus and game_state & (1 << 9):
print("6th Bonus Collected")
await self.ctx.send_msgs(
[{"cmd": 'LocationChecks',
"locations": [SC2WOL_LOC_ID_OFFSET + 100 * self.mission_id + 8]}])
self.eight_bonus = True
else: else:
await self.chat_send("LostConnection - Lost connection to game.") await self.chat_send("LostConnection - Lost connection to game.")
def calc_objectives_completed(mission, missions_info, locations_done, unfinished_locations, ctx): def request_unfinished_missions(ctx: SC2Context):
objectives_complete = 0 if ctx.mission_req_table:
if missions_info[mission].extra_locations > 0:
for i in range(missions_info[mission].extra_locations):
if (missions_info[mission].id * 100 + SC2WOL_LOC_ID_OFFSET + i) in locations_done:
objectives_complete += 1
else:
unfinished_locations[mission].append(ctx.location_names[
missions_info[mission].id * 100 + SC2WOL_LOC_ID_OFFSET + i])
return objectives_complete
else:
return -1
def request_unfinished_missions(locations_done, location_table, ui, ctx):
if location_table:
message = "Unfinished Missions: " message = "Unfinished Missions: "
unlocks = initialize_blank_mission_dict(location_table) unlocks = initialize_blank_mission_dict(ctx.mission_req_table)
unfinished_locations = initialize_blank_mission_dict(location_table) unfinished_locations = initialize_blank_mission_dict(ctx.mission_req_table)
unfinished_missions = calc_unfinished_missions(locations_done, location_table, ctx, unlocks=unlocks, _, unfinished_missions = calc_unfinished_missions(ctx, unlocks=unlocks)
unfinished_locations=unfinished_locations)
message += ", ".join(f"{mark_up_mission_name(mission, location_table, ui,unlocks)}[{location_table[mission].id}] " + message += ", ".join(f"{mark_up_mission_name(ctx, mission, unlocks)}[{ctx.mission_req_table[mission].id}] " +
mark_up_objectives( mark_up_objectives(
f"[{unfinished_missions[mission]}/{location_table[mission].extra_locations}]", f"[{len(unfinished_missions[mission])}/"
f"{sum(1 for _ in ctx.locations_for_mission(mission))}]",
ctx, unfinished_locations, mission) ctx, unfinished_locations, mission)
for mission in unfinished_missions) for mission in unfinished_missions)
if ui: if ctx.ui:
ui.log_panels['All'].on_message_markup(message) ctx.ui.log_panels['All'].on_message_markup(message)
ui.log_panels['Starcraft2'].on_message_markup(message) ctx.ui.log_panels['Starcraft2'].on_message_markup(message)
else: else:
sc2_logger.info(message) sc2_logger.info(message)
else: else:
sc2_logger.warning("No mission table found, you are likely not connected to a server.") sc2_logger.warning("No mission table found, you are likely not connected to a server.")
def calc_unfinished_missions(locations_done, locations, ctx, unlocks=None, unfinished_locations=None, def calc_unfinished_missions(ctx: SC2Context, unlocks=None):
available_missions=[]):
unfinished_missions = [] unfinished_missions = []
locations_completed = [] locations_completed = []
if not unlocks: if not unlocks:
unlocks = initialize_blank_mission_dict(locations) unlocks = initialize_blank_mission_dict(ctx.mission_req_table)
if not unfinished_locations: available_missions = calc_available_missions(ctx, unlocks)
unfinished_locations = initialize_blank_mission_dict(locations)
if len(available_missions) > 0:
available_missions = []
available_missions.extend(calc_available_missions(locations_done, locations, unlocks))
for name in available_missions: for name in available_missions:
if not locations[name].extra_locations == -1: objectives = set(ctx.locations_for_mission(name))
objectives_completed = calc_objectives_completed(name, locations, locations_done, unfinished_locations, ctx) if objectives:
objectives_completed = ctx.checked_locations & objectives
if objectives_completed < locations[name].extra_locations: if len(objectives_completed) < len(objectives):
unfinished_missions.append(name) unfinished_missions.append(name)
locations_completed.append(objectives_completed) locations_completed.append(objectives_completed)
else: else: # infer that this is the final mission as it has no objectives
unfinished_missions.append(name) unfinished_missions.append(name)
locations_completed.append(-1) locations_completed.append(-1)
return {unfinished_missions[i]: locations_completed[i] for i in range(len(unfinished_missions))} return available_missions, dict(zip(unfinished_missions, locations_completed))
def is_mission_available(mission_id_to_check, locations_done, locations): def is_mission_available(ctx: SC2Context, mission_id_to_check):
unfinished_missions = calc_available_missions(locations_done, locations) unfinished_missions = calc_available_missions(ctx)
return any(mission_id_to_check == locations[mission].id for mission in unfinished_missions) return any(mission_id_to_check == ctx.mission_req_table[mission].id for mission in unfinished_missions)
def mark_up_mission_name(mission, location_table, ui, unlock_table): def mark_up_mission_name(ctx: SC2Context, mission, unlock_table):
"""Checks if the mission is required for game completion and adds '*' to the name to mark that.""" """Checks if the mission is required for game completion and adds '*' to the name to mark that."""
if location_table[mission].completion_critical: if ctx.mission_req_table[mission].completion_critical:
if ui: if ctx.ui:
message = "[color=AF99EF]" + mission + "[/color]" message = "[color=AF99EF]" + mission + "[/color]"
else: else:
message = "*" + mission + "*" message = "*" + mission + "*"
else: else:
message = mission message = mission
if ui: if ctx.ui:
unlocks = unlock_table[mission] unlocks = unlock_table[mission]
if len(unlocks) > 0: if len(unlocks) > 0:
pre_message = f"[ref={list(location_table).index(mission)}|Unlocks: " pre_message = f"[ref={list(ctx.mission_req_table).index(mission)}|Unlocks: "
pre_message += ", ".join(f"{unlock}({location_table[unlock].id})" for unlock in unlocks) pre_message += ", ".join(f"{unlock}({ctx.mission_req_table[unlock].id})" for unlock in unlocks)
pre_message += f"]" pre_message += f"]"
message = pre_message + message + "[/ref]" message = pre_message + message + "[/ref]"
@ -751,90 +675,91 @@ def mark_up_objectives(message, ctx, unfinished_locations, mission):
return formatted_message return formatted_message
def request_available_missions(locations_done, location_table, ui): def request_available_missions(ctx: SC2Context):
if location_table: if ctx.mission_req_table:
message = "Available Missions: " message = "Available Missions: "
# Initialize mission unlock table # Initialize mission unlock table
unlocks = initialize_blank_mission_dict(location_table) unlocks = initialize_blank_mission_dict(ctx.mission_req_table)
missions = calc_available_missions(locations_done, location_table, unlocks) missions = calc_available_missions(ctx, unlocks)
message += \ message += \
", ".join(f"{mark_up_mission_name(mission, location_table, ui, unlocks)}[{location_table[mission].id}]" ", ".join(f"{mark_up_mission_name(ctx, mission, unlocks)}"
f"[{ctx.mission_req_table[mission].id}]"
for mission in missions) for mission in missions)
if ui: if ctx.ui:
ui.log_panels['All'].on_message_markup(message) ctx.ui.log_panels['All'].on_message_markup(message)
ui.log_panels['Starcraft2'].on_message_markup(message) ctx.ui.log_panels['Starcraft2'].on_message_markup(message)
else: else:
sc2_logger.info(message) sc2_logger.info(message)
else: else:
sc2_logger.warning("No mission table found, you are likely not connected to a server.") sc2_logger.warning("No mission table found, you are likely not connected to a server.")
def calc_available_missions(locations_done, locations, unlocks=None): def calc_available_missions(ctx: SC2Context, unlocks=None):
available_missions = [] available_missions = []
missions_complete = 0 missions_complete = 0
# Get number of missions completed # Get number of missions completed
for loc in locations_done: for loc in ctx.checked_locations:
if loc % 100 == 0: if loc % victory_modulo == 0:
missions_complete += 1 missions_complete += 1
for name in locations: for name in ctx.mission_req_table:
# Go through the required missions for each mission and fill up unlock table used later for hover-over tooltips # Go through the required missions for each mission and fill up unlock table used later for hover-over tooltips
if unlocks: if unlocks:
for unlock in locations[name].required_world: for unlock in ctx.mission_req_table[name].required_world:
unlocks[list(locations)[unlock-1]].append(name) unlocks[list(ctx.mission_req_table)[unlock - 1]].append(name)
if mission_reqs_completed(name, missions_complete, locations_done, locations): if mission_reqs_completed(ctx, name, missions_complete):
available_missions.append(name) available_missions.append(name)
return available_missions return available_missions
def mission_reqs_completed(location_to_check, missions_complete, locations_done, locations): def mission_reqs_completed(ctx: SC2Context, mission_name: str, missions_complete):
"""Returns a bool signifying if the mission has all requirements complete and can be done """Returns a bool signifying if the mission has all requirements complete and can be done
Keyword arguments: Arguments:
ctx -- instance of SC2Context
locations_to_check -- the mission string name to check locations_to_check -- the mission string name to check
missions_complete -- an int of how many missions have been completed missions_complete -- an int of how many missions have been completed
locations_done -- a list of the location ids that have been complete """
locations -- a dict of MissionInfo for mission requirements for this world""" if len(ctx.mission_req_table[mission_name].required_world) >= 1:
if len(locations[location_to_check].required_world) >= 1:
# A check for when the requirements are being or'd # A check for when the requirements are being or'd
or_success = False or_success = False
# Loop through required missions # Loop through required missions
for req_mission in locations[location_to_check].required_world: for req_mission in ctx.mission_req_table[mission_name].required_world:
req_success = True req_success = True
# Check if required mission has been completed # Check if required mission has been completed
if not (locations[list(locations)[req_mission-1]].id * 100 + SC2WOL_LOC_ID_OFFSET) in locations_done: if not (ctx.mission_req_table[list(ctx.mission_req_table)[req_mission - 1]].id *
if not locations[location_to_check].or_requirements: victory_modulo + SC2WOL_LOC_ID_OFFSET) in ctx.checked_locations:
if not ctx.mission_req_table[mission_name].or_requirements:
return False return False
else: else:
req_success = False req_success = False
# Recursively check required mission to see if it's requirements are met, in case !collect has been done # Recursively check required mission to see if it's requirements are met, in case !collect has been done
if not mission_reqs_completed(list(locations)[req_mission-1], missions_complete, locations_done, if not mission_reqs_completed(ctx, list(ctx.mission_req_table)[req_mission - 1], missions_complete):
locations): if not ctx.mission_req_table[mission_name].or_requirements:
if not locations[location_to_check].or_requirements:
return False return False
else: else:
req_success = False req_success = False
# If requirement check succeeded mark or as satisfied # If requirement check succeeded mark or as satisfied
if locations[location_to_check].or_requirements and req_success: if ctx.mission_req_table[mission_name].or_requirements and req_success:
or_success = True or_success = True
if locations[location_to_check].or_requirements: if ctx.mission_req_table[mission_name].or_requirements:
# Return false if or requirements not met # Return false if or requirements not met
if not or_success: if not or_success:
return False return False
# Check number of missions # Check number of missions
if missions_complete >= locations[location_to_check].number: if missions_complete >= ctx.mission_req_table[mission_name].number:
return True return True
else: else:
return False return False
@ -929,7 +854,7 @@ class DllDirectory:
self.set(self._old) self.set(self._old)
@staticmethod @staticmethod
def get() -> str: def get() -> typing.Optional[str]:
if sys.platform == "win32": if sys.platform == "win32":
n = ctypes.windll.kernel32.GetDllDirectoryW(0, None) n = ctypes.windll.kernel32.GetDllDirectoryW(0, None)
buf = ctypes.create_unicode_buffer(n) buf = ctypes.create_unicode_buffer(n)

2
Utils.py
View File

@ -35,7 +35,7 @@ class Version(typing.NamedTuple):
build: int build: int
__version__ = "0.3.4" __version__ = "0.3.5"
version_tuple = tuplize_version(__version__) version_tuple = tuplize_version(__version__)
is_linux = sys.platform.startswith("linux") is_linux = sys.platform.startswith("linux")

4
worlds/sc2wol/MissionTables.py
View File

@ -69,8 +69,8 @@ vanilla_mission_req_table = {
"Zero Hour": MissionInfo(3, 4, [2], "Mar Sara", completion_critical=True), "Zero Hour": MissionInfo(3, 4, [2], "Mar Sara", completion_critical=True),
"Evacuation": MissionInfo(4, 4, [3], "Colonist"), "Evacuation": MissionInfo(4, 4, [3], "Colonist"),
"Outbreak": MissionInfo(5, 3, [4], "Colonist"), "Outbreak": MissionInfo(5, 3, [4], "Colonist"),
"Safe Haven": MissionInfo(6, 1, [5], "Colonist", number=7), "Safe Haven": MissionInfo(6, 4, [5], "Colonist", number=7),
"Haven's Fall": MissionInfo(7, 1, [5], "Colonist", number=7), "Haven's Fall": MissionInfo(7, 4, [5], "Colonist", number=7),
"Smash and Grab": MissionInfo(8, 5, [3], "Artifact", completion_critical=True), "Smash and Grab": MissionInfo(8, 5, [3], "Artifact", completion_critical=True),
"The Dig": MissionInfo(9, 4, [8], "Artifact", number=8, completion_critical=True), "The Dig": MissionInfo(9, 4, [8], "Artifact", number=8, completion_critical=True),
"The Moebius Factor": MissionInfo(10, 9, [9], "Artifact", number=11, completion_critical=True), "The Moebius Factor": MissionInfo(10, 9, [9], "Artifact", number=11, completion_critical=True),

1
worlds/sc2wol/__init__.py
View File

@ -43,6 +43,7 @@ class SC2WoLWorld(World):
locked_locations: typing.List[str] locked_locations: typing.List[str]
location_cache: typing.List[Location] location_cache: typing.List[Location]
mission_req_table = {} mission_req_table = {}
required_client_version = 0, 3, 5
def __init__(self, world: MultiWorld, player: int): def __init__(self, world: MultiWorld, player: int):
super(SC2WoLWorld, self).__init__(world, player) super(SC2WoLWorld, self).__init__(world, player)