Zork Grand Inquisitor: Implement New Game (#2539)
Adds Archipelago support for Zork Grand Inquisitor, the 1997 point-and-click PC adventure game. The client (based on `CommonClient`), on top of its regular Archipelago duties, fully handles the randomization of the game and the monitoring / modification of the game state. No game modding needed at all; the player is ready to play an Archipelago seed if they can play the vanilla game through ScummVM. The "reverse engineering" (there's likely a better term for this...) of the game is my own original work and I included an MIT license at the root of my world directory. A PopTracker pack was also created to help people learn the game: https://github.com/SerpentAI/ZorkGrandInquisitorAPTracker
This commit is contained in:
Nicholas Brochu
188
worlds/zork_grand_inquisitor/client.py
Normal file
188
worlds/zork_grand_inquisitor/client.py
Normal file
@@ -0,0 +1,188 @@
|
||||
import asyncio
|
||||
|
||||
import CommonClient
|
||||
import NetUtils
|
||||
import Utils
|
||||
|
||||
from typing import Any, Dict, List, Optional, Set, Tuple
|
||||
|
||||
from .data_funcs import item_names_to_id, location_names_to_id, id_to_items, id_to_locations, id_to_goals
|
||||
from .enums import ZorkGrandInquisitorItems, ZorkGrandInquisitorLocations
|
||||
from .game_controller import GameController
|
||||
|
||||
|
||||
class ZorkGrandInquisitorCommandProcessor(CommonClient.ClientCommandProcessor):
|
||||
def _cmd_zork(self) -> None:
|
||||
"""Attach to an open Zork Grand Inquisitor process."""
|
||||
result: bool = self.ctx.game_controller.open_process_handle()
|
||||
|
||||
if result:
|
||||
self.ctx.process_attached_at_least_once = True
|
||||
self.output("Successfully attached to Zork Grand Inquisitor process.")
|
||||
else:
|
||||
self.output("Failed to attach to Zork Grand Inquisitor process.")
|
||||
|
||||
def _cmd_brog(self) -> None:
|
||||
"""List received Brog items."""
|
||||
self.ctx.game_controller.list_received_brog_items()
|
||||
|
||||
def _cmd_griff(self) -> None:
|
||||
"""List received Griff items."""
|
||||
self.ctx.game_controller.list_received_griff_items()
|
||||
|
||||
def _cmd_lucy(self) -> None:
|
||||
"""List received Lucy items."""
|
||||
self.ctx.game_controller.list_received_lucy_items()
|
||||
|
||||
def _cmd_hotspots(self) -> None:
|
||||
"""List received Hotspots."""
|
||||
self.ctx.game_controller.list_received_hotspots()
|
||||
|
||||
|
||||
class ZorkGrandInquisitorContext(CommonClient.CommonContext):
|
||||
tags: Set[str] = {"AP"}
|
||||
game: str = "Zork Grand Inquisitor"
|
||||
command_processor: CommonClient.ClientCommandProcessor = ZorkGrandInquisitorCommandProcessor
|
||||
items_handling: int = 0b111
|
||||
want_slot_data: bool = True
|
||||
|
||||
item_name_to_id: Dict[str, int] = item_names_to_id()
|
||||
location_name_to_id: Dict[str, int] = location_names_to_id()
|
||||
|
||||
id_to_items: Dict[int, ZorkGrandInquisitorItems] = id_to_items()
|
||||
id_to_locations: Dict[int, ZorkGrandInquisitorLocations] = id_to_locations()
|
||||
|
||||
game_controller: GameController
|
||||
|
||||
controller_task: Optional[asyncio.Task]
|
||||
|
||||
process_attached_at_least_once: bool
|
||||
can_display_process_message: bool
|
||||
|
||||
def __init__(self, server_address: Optional[str], password: Optional[str]) -> None:
|
||||
super().__init__(server_address, password)
|
||||
|
||||
self.game_controller = GameController(logger=CommonClient.logger)
|
||||
|
||||
self.controller_task = None
|
||||
|
||||
self.process_attached_at_least_once = False
|
||||
self.can_display_process_message = True
|
||||
|
||||
def run_gui(self) -> None:
|
||||
from kvui import GameManager
|
||||
|
||||
class TextManager(GameManager):
|
||||
logging_pairs: List[Tuple[str, str]] = [("Client", "Archipelago")]
|
||||
base_title: str = "Archipelago Zork Grand Inquisitor Client"
|
||||
|
||||
self.ui = TextManager(self)
|
||||
self.ui_task = asyncio.create_task(self.ui.async_run(), name="UI")
|
||||
|
||||
async def server_auth(self, password_requested: bool = False):
|
||||
if password_requested and not self.password:
|
||||
await super().server_auth(password_requested)
|
||||
|
||||
await self.get_username()
|
||||
await self.send_connect()
|
||||
|
||||
def on_package(self, cmd: str, _args: Any) -> None:
|
||||
if cmd == "Connected":
|
||||
self.game = self.slot_info[self.slot].game
|
||||
|
||||
# Options
|
||||
self.game_controller.option_goal = id_to_goals()[_args["slot_data"]["goal"]]
|
||||
self.game_controller.option_deathsanity = _args["slot_data"]["deathsanity"] == 1
|
||||
|
||||
self.game_controller.option_grant_missable_location_checks = (
|
||||
_args["slot_data"]["grant_missable_location_checks"] == 1
|
||||
)
|
||||
|
||||
async def controller(self):
|
||||
while not self.exit_event.is_set():
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Enqueue Received Item Delta
|
||||
network_item: NetUtils.NetworkItem
|
||||
for network_item in self.items_received:
|
||||
item: ZorkGrandInquisitorItems = self.id_to_items[network_item.item]
|
||||
|
||||
if item not in self.game_controller.received_items:
|
||||
if item not in self.game_controller.received_items_queue:
|
||||
self.game_controller.received_items_queue.append(item)
|
||||
|
||||
# Game Controller Update
|
||||
if self.game_controller.is_process_running():
|
||||
self.game_controller.update()
|
||||
self.can_display_process_message = True
|
||||
else:
|
||||
process_message: str
|
||||
|
||||
if self.process_attached_at_least_once:
|
||||
process_message = (
|
||||
"Lost connection to Zork Grand Inquisitor process. Please restart the game and use the /zork "
|
||||
"command to reattach."
|
||||
)
|
||||
else:
|
||||
process_message = (
|
||||
"Please use the /zork command to attach to a running Zork Grand Inquisitor process."
|
||||
)
|
||||
|
||||
if self.can_display_process_message:
|
||||
CommonClient.logger.info(process_message)
|
||||
self.can_display_process_message = False
|
||||
|
||||
# Send Checked Locations
|
||||
checked_location_ids: List[int] = list()
|
||||
|
||||
while len(self.game_controller.completed_locations_queue) > 0:
|
||||
location: ZorkGrandInquisitorLocations = self.game_controller.completed_locations_queue.popleft()
|
||||
location_id: int = self.location_name_to_id[location.value]
|
||||
|
||||
checked_location_ids.append(location_id)
|
||||
|
||||
await self.send_msgs([
|
||||
{
|
||||
"cmd": "LocationChecks",
|
||||
"locations": checked_location_ids
|
||||
}
|
||||
])
|
||||
|
||||
# Check for Goal Completion
|
||||
if self.game_controller.goal_completed:
|
||||
await self.send_msgs([
|
||||
{
|
||||
"cmd": "StatusUpdate",
|
||||
"status": CommonClient.ClientStatus.CLIENT_GOAL
|
||||
}
|
||||
])
|
||||
|
||||
|
||||
def main() -> None:
|
||||
Utils.init_logging("ZorkGrandInquisitorClient", exception_logger="Client")
|
||||
|
||||
async def _main():
|
||||
ctx: ZorkGrandInquisitorContext = ZorkGrandInquisitorContext(None, None)
|
||||
|
||||
ctx.server_task = asyncio.create_task(CommonClient.server_loop(ctx), name="server loop")
|
||||
ctx.controller_task = asyncio.create_task(ctx.controller(), name="ZorkGrandInquisitorController")
|
||||
|
||||
if CommonClient.gui_enabled:
|
||||
ctx.run_gui()
|
||||
|
||||
ctx.run_cli()
|
||||
|
||||
await ctx.exit_event.wait()
|
||||
await ctx.shutdown()
|
||||
|
||||
import colorama
|
||||
|
||||
colorama.init()
|
||||
|
||||
asyncio.run(_main())
|
||||
|
||||
colorama.deinit()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user