BizHawkClient: Add BizHawkClient (#1978)

Adds a generic client that can communicate with BizHawk. Similar to SNIClient, but for arbitrary systems and doesn't have an intermediary application like SNI.

worlds/_bizhawk/client.py

@@ -0,0 +1,87 @@
+"""
+A module containing the BizHawkClient base class and metaclass
+"""
+
+
+from __future__ import annotations
+
+import abc
+from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Tuple, Union
+
+from worlds.LauncherComponents import Component, SuffixIdentifier, Type, components, launch_subprocess
+
+if TYPE_CHECKING:
+    from .context import BizHawkClientContext
+else:
+    BizHawkClientContext = object
+
+
+class AutoBizHawkClientRegister(abc.ABCMeta):
+    game_handlers: ClassVar[Dict[Tuple[str, ...], Dict[str, BizHawkClient]]] = {}
+
+    def __new__(cls, name: str, bases: Tuple[type, ...], namespace: Dict[str, Any]) -> AutoBizHawkClientRegister:
+        new_class = super().__new__(cls, name, bases, namespace)
+
+        if "system" in namespace:
+            systems = (namespace["system"],) if type(namespace["system"]) is str else tuple(sorted(namespace["system"]))
+            if systems not in AutoBizHawkClientRegister.game_handlers:
+                AutoBizHawkClientRegister.game_handlers[systems] = {}
+
+            if "game" in namespace:
+                AutoBizHawkClientRegister.game_handlers[systems][namespace["game"]] = new_class()
+
+        return new_class
+
+    @staticmethod
+    async def get_handler(ctx: BizHawkClientContext, system: str) -> Optional[BizHawkClient]:
+        for systems, handlers in AutoBizHawkClientRegister.game_handlers.items():
+            if system in systems:
+                for handler in handlers.values():
+                    if await handler.validate_rom(ctx):
+                        return handler
+
+        return None
+
+
+class BizHawkClient(abc.ABC, metaclass=AutoBizHawkClientRegister):
+    system: ClassVar[Union[str, Tuple[str, ...]]]
+    """The system that the game this client is for runs on"""
+
+    game: ClassVar[str]
+    """The game this client is for"""
+
+    @abc.abstractmethod
+    async def validate_rom(self, ctx: BizHawkClientContext) -> bool:
+        """Should return whether the currently loaded ROM should be handled by this client. You might read the game name
+        from the ROM header, for example. This function will only be asked to validate ROMs from the system set by the
+        client class, so you do not need to check the system yourself.
+
+        Once this function has determined that the ROM should be handled by this client, it should also modify `ctx`
+        as necessary (such as setting `ctx.game = self.game`, modifying `ctx.items_handling`, etc...)."""
+        ...
+
+    async def set_auth(self, ctx: BizHawkClientContext) -> None:
+        """Should set ctx.auth in anticipation of sending a `Connected` packet. You may override this if you store slot
+        name in your patched ROM. If ctx.auth is not set after calling, the player will be prompted to enter their
+        username."""
+        pass
+
+    @abc.abstractmethod
+    async def game_watcher(self, ctx: BizHawkClientContext) -> None:
+        """Runs on a loop with the approximate interval `ctx.watcher_timeout`. The currently loaded ROM is guaranteed
+        to have passed your validator when this function is called, and the emulator is very likely to be connected."""
+        ...
+
+    def on_package(self, ctx: BizHawkClientContext, cmd: str, args: dict) -> None:
+        """For handling packages from the server. Called from `BizHawkClientContext.on_package`."""
+        pass
+
+
+def launch_client(*args) -> None:
+    from .context import launch
+    launch_subprocess(launch, name="BizHawkClient")
+
+
+if not any(component.script_name == "BizHawkClient" for component in components):
+    components.append(Component("BizHawk Client", "BizHawkClient", component_type=Type.CLIENT, func=launch_client,
+                                file_identifier=SuffixIdentifier()))