mirror of
https://github.com/ArchipelagoMW/Archipelago.git
synced 2026-03-29 14:43:22 -07:00
119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
from enum import Enum
|
|
from logging import Logger
|
|
from typing import List
|
|
|
|
from .Items import CivVIItemData
|
|
from .TunerClient import TunerClient, TunerConnectionException, TunerTimeoutException
|
|
|
|
class ConnectionState(Enum):
|
|
DISCONNECTED = 0
|
|
IN_GAME = 1
|
|
IN_MENU = 2
|
|
|
|
|
|
|
|
class CivVIInterface:
|
|
logger: Logger
|
|
tuner: TunerClient
|
|
last_error: str = None
|
|
|
|
def __init__(self, logger: Logger):
|
|
self.logger = logger
|
|
self.tuner = TunerClient(logger)
|
|
|
|
async def is_in_game(self) -> ConnectionState:
|
|
command = "IsInGame()"
|
|
try:
|
|
result = await self.tuner.send_game_command(command)
|
|
if result == "false":
|
|
return ConnectionState.IN_MENU
|
|
self.last_error = None
|
|
return ConnectionState.IN_GAME
|
|
except TunerTimeoutException:
|
|
self.print_connection_error(
|
|
"Not connected to game, waiting for connection to be available")
|
|
return ConnectionState.DISCONNECTED
|
|
except TunerConnectionException as e:
|
|
if "The remote computer refused the network connection" in str(e):
|
|
self.print_connection_error(
|
|
"Unable to connect to game. Verify that the tuner is enabled. Attempting to reconnect")
|
|
else:
|
|
self.print_connection_error(
|
|
"Not connected to game, waiting for connection to be available")
|
|
return ConnectionState.DISCONNECTED
|
|
except Exception as e:
|
|
if "attempt to index a nil valuestack traceback" in str(e) \
|
|
or ".. is not supported for string .. nilstack traceback" in str(e):
|
|
return ConnectionState.IN_MENU
|
|
|
|
def print_connection_error(self, error: str) -> None:
|
|
if error != self.last_error:
|
|
self.last_error = error
|
|
self.logger.info(error)
|
|
|
|
async def give_item_to_player(self, item: CivVIItemData, sender: str = "", amount: int = 1) -> None:
|
|
if isinstance(item.civ_vi_id, str):
|
|
item_id = f'"{item.civ_vi_id}"'
|
|
else:
|
|
item_id = item.civ_vi_id
|
|
|
|
command = f"HandleReceiveItem({item_id}, \"{item.name}\", \"{item.item_type.value}\", \"{sender}\", {amount})"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def resync(self) -> None:
|
|
"""Has the client resend all the checked locations"""
|
|
command = "Resync()"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def check_victory(self) -> bool:
|
|
command = "ClientGetVictory()"
|
|
result = await self.tuner.send_game_command(command)
|
|
return result == "true"
|
|
|
|
async def get_checked_locations(self) -> List[str]:
|
|
command = "GetUnsentCheckedLocations()"
|
|
result = await self.tuner.send_game_command(command, 2048 * 4)
|
|
return result.split(",")
|
|
|
|
async def get_deathlink(self) -> str:
|
|
"""returns either "false" or the name of the unit that killed the player's unit"""
|
|
command = "ClientGetDeathLink()"
|
|
result = await self.tuner.send_game_command(command)
|
|
return result
|
|
|
|
async def kill_unit(self, message: str) -> None:
|
|
command = f"KillUnit(\"{message}\")"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def get_last_received_index(self) -> int:
|
|
command = "ClientGetLastReceivedIndex()"
|
|
result = await self.tuner.send_game_command(command)
|
|
return int(result)
|
|
|
|
async def send_notification(self, item: CivVIItemData, sender="someone") -> None:
|
|
command = f"GameCore.NotificationManager:SendNotification(GameCore.NotificationTypes.USER_DEFINED_2, \"{item.name} Received\", \"You have received {item.name} from \" .. \"{sender}\", 0, {item.civ_vi_id})"
|
|
await self.tuner.send_command(command)
|
|
|
|
async def decrease_gold_by_percent(self, percent: int, message: str) -> None:
|
|
command = f"DecreaseGoldByPercent({percent}, \"{message}\")"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def decrease_faith_by_percent(self, percent: int, message: str) -> None:
|
|
command = f"DecreaseFaithByPercent({percent}, \"{message}\")"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def decrease_era_score_by_amount(self, amount: int, message: str) -> None:
|
|
command = f"DecreaseEraScoreByAmount({amount}, \"{message}\")"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def set_max_allowed_era(self, count: int) -> None:
|
|
command = f"SetMaxAllowedEra(\"{count}\")"
|
|
await self.tuner.send_game_command(command)
|
|
|
|
async def get_max_allowed_era(self) -> int:
|
|
command = "ClientGetMaxAllowedEra()"
|
|
result = await self.tuner.send_game_command(command)
|
|
if result == "":
|
|
return -1
|
|
return int(result)
|