morgan/Grinch-AP
1
0
Fork 0
mirror of https://github.com/MarioSpore/Grinch-AP.git synced 2025-10-21 12:11:33 -06:00
Code Issues Packages Projects Releases Wiki Activity

Zillion: use "new" settings api and cleaning (#3903)

Browse Source 
* Zillion: use "new" settings api and cleaning

* python 3.10 typing update

* don't separate assignments of item link players
This commit is contained in:
Doug Hoskisson
2024-11-29 12:25:01 -08:00
committed by GitHub
parent b5343a36ff
commit 2fb59d39c9
10 changed files with 136 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
worlds/zillion/client.py
View File

@@ -3,7 +3,7 @@ import base64
import io
import pkgutil
import platform
from typing import Any, ClassVar, Coroutine, Dict, List, Optional, Protocol, Tuple, cast
from typing import Any, ClassVar, Coroutine, Protocol, cast
from CommonClient import CommonContext, server_loop, gui_enabled, \
ClientCommandProcessor, logger, get_base_parser
@@ -11,6 +11,7 @@ from NetUtils import ClientStatus
from Utils import async_start
import colorama
from typing_extensions import override
from zilliandomizer.zri.memory import Memory, RescueInfo
from zilliandomizer.zri import events
@@ -35,11 +36,11 @@ class ZillionCommandProcessor(ClientCommandProcessor):
class ToggleCallback(Protocol):
def __call__(self) -> None: ...
def __call__(self) -> object: ...
class SetRoomCallback(Protocol):
def __call__(self, rooms: List[List[int]]) -> None: ...
def __call__(self, rooms: list[list[int]]) -> object: ...
class ZillionContext(CommonContext):
@@ -47,7 +48,7 @@ class ZillionContext(CommonContext):
command_processor = ZillionCommandProcessor
items_handling = 1 # receive items from other players
known_name: Optional[str]
known_name: str | None
""" This is almost the same as `auth` except `auth` is reset to `None` when server disconnects, and this isn't. """
from_game: "asyncio.Queue[events.EventFromGame]"
@@ -56,11 +57,11 @@ class ZillionContext(CommonContext):
""" local checks watched by server """
next_item: int
""" index in `items_received` """
ap_id_to_name: Dict[int, str]
ap_id_to_zz_id: Dict[int, int]
ap_id_to_name: dict[int, str]
ap_id_to_zz_id: dict[int, int]
start_char: Chars = "JJ"
rescues: Dict[int, RescueInfo] = {}
loc_mem_to_id: Dict[int, int] = {}
rescues: dict[int, RescueInfo] = {}
loc_mem_to_id: dict[int, int] = {}
got_room_info: asyncio.Event
""" flag for connected to server """
got_slot_data: asyncio.Event
@@ -119,22 +120,22 @@ class ZillionContext(CommonContext):
self.finished_game = False
self.items_received.clear()
# override
def on_deathlink(self, data: Dict[str, Any]) -> None:
@override
def on_deathlink(self, data: dict[str, Any]) -> None:
self.to_game.put_nowait(events.DeathEventToGame())
return super().on_deathlink(data)
# override
@override
async def server_auth(self, password_requested: bool = False) -> None:
if password_requested and not self.password:
await super().server_auth(password_requested)
if not self.auth:
logger.info('waiting for connection to game...')
logger.info("waiting for connection to game...")
return
logger.info("logging in to server...")
await self.send_connect()
# override
@override
def run_gui(self) -> None:
from kvui import GameManager
from kivy.core.text import Label as CoreLabel
@@ -154,10 +155,10 @@ class ZillionContext(CommonContext):
MAP_WIDTH: ClassVar[int] = 281
map_background: CoreImage
_number_textures: List[Texture] = []
rooms: List[List[int]] = []
_number_textures: list[Texture] = []
rooms: list[list[int]] = []
def __init__(self, **kwargs: Any) -> None:
def __init__(self, **kwargs: Any) -> None: # noqa: ANN401
super().__init__(**kwargs)
FILE_NAME = "empty-zillion-map-row-col-labels-281.png"
@@ -183,7 +184,7 @@ class ZillionContext(CommonContext):
label.refresh()
self._number_textures.append(label.texture)
def update_map(self, *args: Any) -> None:
def update_map(self, *args: Any) -> None: # noqa: ANN401
self.canvas.clear()
with self.canvas:
@@ -203,6 +204,7 @@ class ZillionContext(CommonContext):
num_texture = self._number_textures[num]
Rectangle(texture=num_texture, size=num_texture.size, pos=pos)
@override
def build(self) -> Layout:
container = super().build()
self.map_widget = ZillionManager.MapPanel(size_hint_x=None, width=ZillionManager.MapPanel.MAP_WIDTH)
@@ -216,17 +218,18 @@ class ZillionContext(CommonContext):
self.map_widget.width = 0
self.container.do_layout()
def set_rooms(self, rooms: List[List[int]]) -> None:
def set_rooms(self, rooms: list[list[int]]) -> None:
self.map_widget.rooms = rooms
self.map_widget.update_map()
self.ui = ZillionManager(self)
self.ui_toggle_map = lambda: self.ui.toggle_map_width()
self.ui_set_rooms = lambda rooms: self.ui.set_rooms(rooms)
self.ui_toggle_map = lambda: isinstance(self.ui, ZillionManager) and self.ui.toggle_map_width()
self.ui_set_rooms = lambda rooms: isinstance(self.ui, ZillionManager) and self.ui.set_rooms(rooms)
run_co: Coroutine[Any, Any, None] = self.ui.async_run()
self.ui_task = asyncio.create_task(run_co, name="UI")
def on_package(self, cmd: str, args: Dict[str, Any]) -> None:
@override
def on_package(self, cmd: str, args: dict[str, Any]) -> None:
self.room_item_numbers_to_ui()
if cmd == "Connected":
logger.info("logged in to Archipelago server")
@@ -238,7 +241,7 @@ class ZillionContext(CommonContext):
if "start_char" not in slot_data:
logger.warning("invalid Zillion `Connected` packet, `slot_data` missing `start_char`")
return
self.start_char = slot_data['start_char']
self.start_char = slot_data["start_char"]
if self.start_char not in {"Apple", "Champ", "JJ"}:
logger.warning("invalid Zillion `Connected` packet, "
f"`slot_data` `start_char` has invalid value: {self.start_char}")
@@ -259,7 +262,7 @@ class ZillionContext(CommonContext):
self.rescues[0 if rescue_id == "0" else 1] = ri
if "loc_mem_to_id" not in slot_data:
logger.warn("invalid Zillion `Connected` packet, `slot_data` missing `loc_mem_to_id`")
logger.warning("invalid Zillion `Connected` packet, `slot_data` missing `loc_mem_to_id`")
return
loc_mem_to_id = slot_data["loc_mem_to_id"]
self.loc_mem_to_id = {}
@@ -286,7 +289,7 @@ class ZillionContext(CommonContext):
if "keys" not in args:
logger.warning(f"invalid Retrieved packet to ZillionClient: {args}")
return
keys = cast(Dict[str, Optional[str]], args["keys"])
keys = cast(dict[str, str | None], args["keys"])
doors_b64 = keys.get(f"zillion-{self.auth}-doors", None)
if doors_b64:
logger.info("received door data from server")
@@ -321,9 +324,9 @@ class ZillionContext(CommonContext):
if server_id in self.missing_locations:
self.ap_local_count += 1
n_locations = len(self.missing_locations) + len(self.checked_locations) - 1 # -1 to ignore win
logger.info(f'New Check: {loc_name} ({self.ap_local_count}/{n_locations})')
logger.info(f"New Check: {loc_name} ({self.ap_local_count}/{n_locations})")
async_start(self.send_msgs([
{"cmd": 'LocationChecks', "locations": [server_id]}
{"cmd": "LocationChecks", "locations": [server_id]}
]))
else:
# This will happen a lot in Zillion,
@@ -334,7 +337,7 @@ class ZillionContext(CommonContext):
elif isinstance(event_from_game, events.WinEventFromGame):
if not self.finished_game:
async_start(self.send_msgs([
{"cmd": 'LocationChecks', "locations": [loc_name_to_id["J-6 bottom far left"]]},
{"cmd": "LocationChecks", "locations": [loc_name_to_id["J-6 bottom far left"]]},
{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}
]))
self.finished_game = True
@@ -362,24 +365,24 @@ class ZillionContext(CommonContext):
ap_id = self.items_received[index].item
from_name = self.player_names[self.items_received[index].player]
# TODO: colors in this text, like sni client?
logger.info(f'received {self.ap_id_to_name[ap_id]} from {from_name}')
logger.info(f"received {self.ap_id_to_name[ap_id]} from {from_name}")
self.to_game.put_nowait(
events.ItemEventToGame(zz_item_ids)
)
self.next_item = len(self.items_received)
def name_seed_from_ram(data: bytes) -> Tuple[str, str]:
def name_seed_from_ram(data: bytes) -> tuple[str, str]:
""" returns player name, and end of seed string """
if len(data) == 0:
# no connection to game
return "", "xxx"
null_index = data.find(b'\x00')
null_index = data.find(b"\x00")
if null_index == -1:
logger.warning(f"invalid game id in rom {repr(data)}")
null_index = len(data)
name = data[:null_index].decode()
null_index_2 = data.find(b'\x00', null_index + 1)
null_index_2 = data.find(b"\x00", null_index + 1)
if null_index_2 == -1:
null_index_2 = len(data)
seed_name = data[null_index + 1:null_index_2].decode()
@@ -479,8 +482,8 @@ async def zillion_sync_task(ctx: ZillionContext) -> None:
async def main() -> None:
parser = get_base_parser()
parser.add_argument('diff_file', default="", type=str, nargs="?",
help='Path to a .apzl Archipelago Binary Patch file')
parser.add_argument("diff_file", default="", type=str, nargs="?",
help="Path to a .apzl Archipelago Binary Patch file")
# SNI parser.add_argument('--loglevel', default='info', choices=['debug', 'info', 'warning', 'error', 'critical'])
args = parser.parse_args()
print(args)