SC2: Python 3.11 compatibility (#1821)

Co-authored-by: Salzkorn

worlds/_sc2common/bot/client.py

@@ -0,0 +1,720 @@
+from __future__ import annotations
+
+from typing import Dict, Iterable, List, Optional, Set, Tuple, Union
+
+from worlds._sc2common.bot import logger
+
+from s2clientprotocol import debug_pb2 as debug_pb
+from s2clientprotocol import query_pb2 as query_pb
+from s2clientprotocol import raw_pb2 as raw_pb
+from s2clientprotocol import sc2api_pb2 as sc_pb
+from s2clientprotocol import spatial_pb2 as spatial_pb
+
+from .data import ActionResult, ChatChannel, Race, Result, Status
+from .game_data import AbilityData, GameData
+from .game_info import GameInfo
+from .position import Point2, Point3
+from .protocol import ConnectionAlreadyClosed, Protocol, ProtocolError
+from .renderer import Renderer
+from .unit import Unit
+from .units import Units
+
+
+# pylint: disable=R0904
+class Client(Protocol):
+
+    def __init__(self, ws, save_replay_path: str = None):
+        """
+        :param ws:
+        """
+        super().__init__(ws)
+        # How many frames will be waited between iterations before the next one is called
+        self.game_step: int = 4
+        self.save_replay_path: Optional[str] = save_replay_path
+        self._player_id = None
+        self._game_result = None
+        # Store a hash value of all the debug requests to prevent sending the same ones again if they haven't changed last frame
+        self._debug_hash_tuple_last_iteration: Tuple[int, int, int, int] = (0, 0, 0, 0)
+        self._debug_draw_last_frame = False
+        self._debug_texts = []
+        self._debug_lines = []
+        self._debug_boxes = []
+        self._debug_spheres = []
+
+        self._renderer = None
+        self.raw_affects_selection = False
+
+    @property
+    def in_game(self) -> bool:
+        return self._status in {Status.in_game, Status.in_replay}
+
+    async def join_game(self, name=None, race=None, observed_player_id=None, portconfig=None, rgb_render_config=None):
+        ifopts = sc_pb.InterfaceOptions(
+            raw=True,
+            score=True,
+            show_cloaked=True,
+            show_burrowed_shadows=True,
+            raw_affects_selection=self.raw_affects_selection,
+            raw_crop_to_playable_area=False,
+            show_placeholders=True,
+        )
+
+        if rgb_render_config:
+            assert isinstance(rgb_render_config, dict)
+            assert "window_size" in rgb_render_config and "minimap_size" in rgb_render_config
+            window_size = rgb_render_config["window_size"]
+            minimap_size = rgb_render_config["minimap_size"]
+            self._renderer = Renderer(self, window_size, minimap_size)
+            map_width, map_height = window_size
+            minimap_width, minimap_height = minimap_size
+
+            ifopts.render.resolution.x = map_width
+            ifopts.render.resolution.y = map_height
+            ifopts.render.minimap_resolution.x = minimap_width
+            ifopts.render.minimap_resolution.y = minimap_height
+
+        if race is None:
+            assert isinstance(observed_player_id, int), f"observed_player_id is of type {type(observed_player_id)}"
+            # join as observer
+            req = sc_pb.RequestJoinGame(observed_player_id=observed_player_id, options=ifopts)
+        else:
+            assert isinstance(race, Race)
+            req = sc_pb.RequestJoinGame(race=race.value, options=ifopts)
+
+        if portconfig:
+            req.server_ports.game_port = portconfig.server[0]
+            req.server_ports.base_port = portconfig.server[1]
+
+            for ppc in portconfig.players:
+                p = req.client_ports.add()
+                p.game_port = ppc[0]
+                p.base_port = ppc[1]
+
+        if name is not None:
+            assert isinstance(name, str), f"name is of type {type(name)}"
+            req.player_name = name
+
+        result = await self._execute(join_game=req)
+        self._game_result = None
+        self._player_id = result.join_game.player_id
+        return result.join_game.player_id
+
+    async def leave(self):
+        """ You can use 'await self.client.leave()' to surrender midst game. """
+        is_resign = self._game_result is None
+
+        if is_resign:
+            # For all clients that can leave, result of leaving the game either
+            # loss, or the client will ignore the result
+            self._game_result = {self._player_id: Result.Defeat}
+
+        try:
+            if self.save_replay_path is not None:
+                await self.save_replay(self.save_replay_path)
+                self.save_replay_path = None
+            await self._execute(leave_game=sc_pb.RequestLeaveGame())
+        except (ProtocolError, ConnectionAlreadyClosed):
+            if is_resign:
+                raise
+
+    async def save_replay(self, path):
+        logger.debug("Requesting replay from server")
+        result = await self._execute(save_replay=sc_pb.RequestSaveReplay())
+        with open(path, "wb") as f:
+            f.write(result.save_replay.data)
+        logger.info(f"Saved replay to {path}")
+
+    async def observation(self, game_loop: int = None):
+        if game_loop is not None:
+            result = await self._execute(observation=sc_pb.RequestObservation(game_loop=game_loop))
+        else:
+            result = await self._execute(observation=sc_pb.RequestObservation())
+        assert result.HasField("observation")
+
+        if not self.in_game or result.observation.player_result:
+            # Sometimes game ends one step before results are available
+            if not result.observation.player_result:
+                result = await self._execute(observation=sc_pb.RequestObservation())
+                assert result.observation.player_result
+
+            player_id_to_result = {}
+            for pr in result.observation.player_result:
+                player_id_to_result[pr.player_id] = Result(pr.result)
+            self._game_result = player_id_to_result
+        self._game_result = None
+
+        # if render_data is available, then RGB rendering was requested
+        if self._renderer and result.observation.observation.HasField("render_data"):
+            await self._renderer.render(result.observation)
+
+        return result
+
+    async def step(self, step_size: int = None):
+        """ EXPERIMENTAL: Change self._client.game_step during the step function to increase or decrease steps per second """
+        step_size = step_size or self.game_step
+        return await self._execute(step=sc_pb.RequestStep(count=step_size))
+
+    async def get_game_data(self) -> GameData:
+        result = await self._execute(
+            data=sc_pb.RequestData(ability_id=True, unit_type_id=True, upgrade_id=True, buff_id=True, effect_id=True)
+        )
+        return GameData(result.data)
+
+    async def dump_data(self, ability_id=True, unit_type_id=True, upgrade_id=True, buff_id=True, effect_id=True):
+        """
+        Dump the game data files
+        choose what data to dump in the keywords
+        this function writes to a text file
+        call it one time in on_step with:
+        await self._client.dump_data()
+        """
+        result = await self._execute(
+            data=sc_pb.RequestData(
+                ability_id=ability_id,
+                unit_type_id=unit_type_id,
+                upgrade_id=upgrade_id,
+                buff_id=buff_id,
+                effect_id=effect_id,
+            )
+        )
+        with open("data_dump.txt", "a") as file:
+            file.write(str(result.data))
+
+    async def get_game_info(self) -> GameInfo:
+        result = await self._execute(game_info=sc_pb.RequestGameInfo())
+        return GameInfo(result.game_info)
+
+    async def query_pathing(self, start: Union[Unit, Point2, Point3],
+                            end: Union[Point2, Point3]) -> Optional[Union[int, float]]:
+        """Caution: returns "None" when path not found
+        Try to combine queries with the function below because the pathing query is generally slow.
+
+        :param start:
+        :param end:"""
+        assert isinstance(start, (Point2, Unit))
+        assert isinstance(end, Point2)
+        if isinstance(start, Point2):
+            path = [query_pb.RequestQueryPathing(start_pos=start.as_Point2D, end_pos=end.as_Point2D)]
+        else:
+            path = [query_pb.RequestQueryPathing(unit_tag=start.tag, end_pos=end.as_Point2D)]
+        result = await self._execute(query=query_pb.RequestQuery(pathing=path))
+        distance = float(result.query.pathing[0].distance)
+        if distance <= 0.0:
+            return None
+        return distance
+
+    async def query_pathings(self, zipped_list: List[List[Union[Unit, Point2, Point3]]]) -> List[float]:
+        """Usage: await self.query_pathings([[unit1, target2], [unit2, target2]])
+        -> returns [distance1, distance2]
+        Caution: returns 0 when path not found
+
+        :param zipped_list:
+        """
+        assert zipped_list, "No zipped_list"
+        assert isinstance(zipped_list, list), f"{type(zipped_list)}"
+        assert isinstance(zipped_list[0], list), f"{type(zipped_list[0])}"
+        assert len(zipped_list[0]) == 2, f"{len(zipped_list[0])}"
+        assert isinstance(zipped_list[0][0], (Point2, Unit)), f"{type(zipped_list[0][0])}"
+        assert isinstance(zipped_list[0][1], Point2), f"{type(zipped_list[0][1])}"
+        if isinstance(zipped_list[0][0], Point2):
+            path = (
+                query_pb.RequestQueryPathing(start_pos=p1.as_Point2D, end_pos=p2.as_Point2D) for p1, p2 in zipped_list
+            )
+        else:
+            path = (query_pb.RequestQueryPathing(unit_tag=p1.tag, end_pos=p2.as_Point2D) for p1, p2 in zipped_list)
+        results = await self._execute(query=query_pb.RequestQuery(pathing=path))
+        return [float(d.distance) for d in results.query.pathing]
+
+    async def query_building_placement(
+        self,
+        ability: AbilityData,
+        positions: List[Union[Point2, Point3]],
+        ignore_resources: bool = True
+    ) -> List[ActionResult]:
+        """This function might be deleted in favor of the function above (_query_building_placement_fast).
+
+        :param ability:
+        :param positions:
+        :param ignore_resources:"""
+        assert isinstance(ability, AbilityData)
+        result = await self._execute(
+            query=query_pb.RequestQuery(
+                placements=(
+                    query_pb.RequestQueryBuildingPlacement(ability_id=ability.id.value, target_pos=position.as_Point2D)
+                    for position in positions
+                ),
+                ignore_resource_requirements=ignore_resources,
+            )
+        )
+        # Unnecessary converting to ActionResult?
+        return [ActionResult(p.result) for p in result.query.placements]
+
+    async def chat_send(self, message: str, team_only: bool):
+        """ Writes a message to the chat """
+        ch = ChatChannel.Team if team_only else ChatChannel.Broadcast
+        await self._execute(
+            action=sc_pb.RequestAction(
+                actions=[sc_pb.Action(action_chat=sc_pb.ActionChat(channel=ch.value, message=message))]
+            )
+        )
+
+    async def debug_kill_unit(self, unit_tags: Union[Unit, Units, List[int], Set[int]]):
+        """
+        :param unit_tags:
+        """
+        if isinstance(unit_tags, Units):
+            unit_tags = unit_tags.tags
+        if isinstance(unit_tags, Unit):
+            unit_tags = [unit_tags.tag]
+        assert unit_tags
+
+        await self._execute(
+            debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(kill_unit=debug_pb.DebugKillUnit(tag=unit_tags))])
+        )
+
+    async def move_camera(self, position: Union[Unit, Units, Point2, Point3]):
+        """Moves camera to the target position
+
+        :param position:"""
+        assert isinstance(position, (Unit, Units, Point2, Point3))
+        if isinstance(position, Units):
+            position = position.center
+        if isinstance(position, Unit):
+            position = position.position
+        await self._execute(
+            action=sc_pb.RequestAction(
+                actions=[
+                    sc_pb.Action(
+                        action_raw=raw_pb.ActionRaw(
+                            camera_move=raw_pb.ActionRawCameraMove(center_world_space=position.to3.as_Point)
+                        )
+                    )
+                ]
+            )
+        )
+
+    async def obs_move_camera(self, position: Union[Unit, Units, Point2, Point3]):
+        """Moves observer camera to the target position. Only works when observing (e.g. watching the replay).
+
+        :param position:"""
+        assert isinstance(position, (Unit, Units, Point2, Point3))
+        if isinstance(position, Units):
+            position = position.center
+        if isinstance(position, Unit):
+            position = position.position
+        await self._execute(
+            obs_action=sc_pb.RequestObserverAction(
+                actions=[
+                    sc_pb.ObserverAction(camera_move=sc_pb.ActionObserverCameraMove(world_pos=position.as_Point2D))
+                ]
+            )
+        )
+
+    async def move_camera_spatial(self, position: Union[Point2, Point3]):
+        """Moves camera to the target position using the spatial aciton interface
+
+        :param position:"""
+        assert isinstance(position, (Point2, Point3))
+        action = sc_pb.Action(
+            action_render=spatial_pb.ActionSpatial(
+                camera_move=spatial_pb.ActionSpatialCameraMove(center_minimap=position.as_PointI)
+            )
+        )
+        await self._execute(action=sc_pb.RequestAction(actions=[action]))
+
+    def debug_text_simple(self, text: str):
+        """ Draws a text in the top left corner of the screen (up to a max of 6 messages fit there). """
+        self._debug_texts.append(DrawItemScreenText(text=text, color=None, start_point=Point2((0, 0)), font_size=8))
+
+    def debug_text_screen(
+        self,
+        text: str,
+        pos: Union[Point2, Point3, tuple, list],
+        color: Union[tuple, list, Point3] = None,
+        size: int = 8,
+    ):
+        """
+        Draws a text on the screen (monitor / game window) with coordinates 0 <= x, y <= 1.
+
+        :param text:
+        :param pos:
+        :param color:
+        :param size:
+        """
+        assert len(pos) >= 2
+        assert 0 <= pos[0] <= 1
+        assert 0 <= pos[1] <= 1
+        pos = Point2((pos[0], pos[1]))
+        self._debug_texts.append(DrawItemScreenText(text=text, color=color, start_point=pos, font_size=size))
+
+    def debug_text_2d(
+        self,
+        text: str,
+        pos: Union[Point2, Point3, tuple, list],
+        color: Union[tuple, list, Point3] = None,
+        size: int = 8,
+    ):
+        return self.debug_text_screen(text, pos, color, size)
+
+    def debug_text_world(
+        self, text: str, pos: Union[Unit, Point3], color: Union[tuple, list, Point3] = None, size: int = 8
+    ):
+        """
+        Draws a text at Point3 position in the game world.
+        To grab a unit's 3d position, use unit.position3d
+        Usually the Z value of a Point3 is between 8 and 14 (except for flying units). Use self.get_terrain_z_height() from bot_ai.py to get the Z value (height) of the terrain at a 2D position.
+
+        :param text:
+        :param color:
+        :param size:
+        """
+        if isinstance(pos, Unit):
+            pos = pos.position3d
+        assert isinstance(pos, Point3)
+        self._debug_texts.append(DrawItemWorldText(text=text, color=color, start_point=pos, font_size=size))
+
+    def debug_text_3d(
+        self, text: str, pos: Union[Unit, Point3], color: Union[tuple, list, Point3] = None, size: int = 8
+    ):
+        return self.debug_text_world(text, pos, color, size)
+
+    def debug_line_out(
+        self, p0: Union[Unit, Point3], p1: Union[Unit, Point3], color: Union[tuple, list, Point3] = None
+    ):
+        """
+        Draws a line from p0 to p1.
+
+        :param p0:
+        :param p1:
+        :param color:
+        """
+        if isinstance(p0, Unit):
+            p0 = p0.position3d
+        assert isinstance(p0, Point3)
+        if isinstance(p1, Unit):
+            p1 = p1.position3d
+        assert isinstance(p1, Point3)
+        self._debug_lines.append(DrawItemLine(color=color, start_point=p0, end_point=p1))
+
+    def debug_box_out(
+        self,
+        p_min: Union[Unit, Point3],
+        p_max: Union[Unit, Point3],
+        color: Union[tuple, list, Point3] = None,
+    ):
+        """
+        Draws a box with p_min and p_max as corners of the box.
+
+        :param p_min:
+        :param p_max:
+        :param color:
+        """
+        if isinstance(p_min, Unit):
+            p_min = p_min.position3d
+        assert isinstance(p_min, Point3)
+        if isinstance(p_max, Unit):
+            p_max = p_max.position3d
+        assert isinstance(p_max, Point3)
+        self._debug_boxes.append(DrawItemBox(start_point=p_min, end_point=p_max, color=color))
+
+    def debug_box2_out(
+        self,
+        pos: Union[Unit, Point3],
+        half_vertex_length: float = 0.25,
+        color: Union[tuple, list, Point3] = None,
+    ):
+        """
+        Draws a box center at a position 'pos', with box side lengths (vertices) of two times 'half_vertex_length'.
+
+        :param pos:
+        :param half_vertex_length:
+        :param color:
+        """
+        if isinstance(pos, Unit):
+            pos = pos.position3d
+        assert isinstance(pos, Point3)
+        p0 = pos + Point3((-half_vertex_length, -half_vertex_length, -half_vertex_length))
+        p1 = pos + Point3((half_vertex_length, half_vertex_length, half_vertex_length))
+        self._debug_boxes.append(DrawItemBox(start_point=p0, end_point=p1, color=color))
+
+    def debug_sphere_out(self, p: Union[Unit, Point3], r: float, color: Union[tuple, list, Point3] = None):
+        """
+        Draws a sphere at point p with radius r.
+
+        :param p:
+        :param r:
+        :param color:
+        """
+        if isinstance(p, Unit):
+            p = p.position3d
+        assert isinstance(p, Point3)
+        self._debug_spheres.append(DrawItemSphere(start_point=p, radius=r, color=color))
+
+    async def _send_debug(self):
+        """Sends the debug draw execution. This is run by main.py now automatically, if there is any items in the list. You do not need to run this manually any longer.
+        Check examples/terran/ramp_wall.py for example drawing. Each draw request needs to be sent again in every single on_step iteration.
+        """
+        debug_hash = (
+            sum(hash(item) for item in self._debug_texts),
+            sum(hash(item) for item in self._debug_lines),
+            sum(hash(item) for item in self._debug_boxes),
+            sum(hash(item) for item in self._debug_spheres),
+        )
+        if debug_hash != (0, 0, 0, 0):
+            if debug_hash != self._debug_hash_tuple_last_iteration:
+                # Something has changed, either more or less is to be drawn, or a position of a drawing changed (e.g. when drawing on a moving unit)
+                self._debug_hash_tuple_last_iteration = debug_hash
+                try:
+                    await self._execute(
+                        debug=sc_pb.RequestDebug(
+                            debug=[
+                                debug_pb.DebugCommand(
+                                    draw=debug_pb.DebugDraw(
+                                        text=[text.to_proto()
+                                              for text in self._debug_texts] if self._debug_texts else None,
+                                        lines=[line.to_proto()
+                                               for line in self._debug_lines] if self._debug_lines else None,
+                                        boxes=[box.to_proto()
+                                               for box in self._debug_boxes] if self._debug_boxes else None,
+                                        spheres=[sphere.to_proto()
+                                                 for sphere in self._debug_spheres] if self._debug_spheres else None,
+                                    )
+                                )
+                            ]
+                        )
+                    )
+                except ProtocolError:
+                    return
+            self._debug_draw_last_frame = True
+            self._debug_texts.clear()
+            self._debug_lines.clear()
+            self._debug_boxes.clear()
+            self._debug_spheres.clear()
+        elif self._debug_draw_last_frame:
+            # Clear drawing if we drew last frame but nothing to draw this frame
+            self._debug_hash_tuple_last_iteration = (0, 0, 0, 0)
+            await self._execute(
+                debug=sc_pb.RequestDebug(
+                    debug=[
+                        debug_pb.DebugCommand(draw=debug_pb.DebugDraw(text=None, lines=None, boxes=None, spheres=None))
+                    ]
+                )
+            )
+            self._debug_draw_last_frame = False
+
+    async def debug_leave(self):
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(end_game=debug_pb.DebugEndGame())]))
+
+    async def debug_set_unit_value(self, unit_tags: Union[Iterable[int], Units, Unit], unit_value: int, value: float):
+        """Sets a "unit value" (Energy, Life or Shields) of the given units to the given value.
+        Can't set the life of a unit to 0, use "debug_kill_unit" for that. Also can't set the life above the unit's maximum.
+        The following example sets the health of all your workers to 1:
+        await self.debug_set_unit_value(self.workers, 2, value=1)"""
+        if isinstance(unit_tags, Units):
+            unit_tags = unit_tags.tags
+        if isinstance(unit_tags, Unit):
+            unit_tags = [unit_tags.tag]
+        assert hasattr(
+            unit_tags, "__iter__"
+        ), f"unit_tags argument needs to be an iterable (list, dict, set, Units), given argument is {type(unit_tags).__name__}"
+        assert (
+            1 <= unit_value <= 3
+        ), f"unit_value needs to be between 1 and 3 (1 for energy, 2 for life, 3 for shields), given argument is {unit_value}"
+        assert all(tag > 0 for tag in unit_tags), f"Unit tags have invalid value: {unit_tags}"
+        assert isinstance(value, (int, float)), "Value needs to be of type int or float"
+        assert value >= 0, "Value can't be negative"
+        await self._execute(
+            debug=sc_pb.RequestDebug(
+                debug=(
+                    debug_pb.DebugCommand(
+                        unit_value=debug_pb.
+                        DebugSetUnitValue(unit_value=unit_value, value=float(value), unit_tag=unit_tag)
+                    ) for unit_tag in unit_tags
+                )
+            )
+        )
+
+    async def debug_hang(self, delay_in_seconds: float):
+        """ Freezes the SC2 client. Not recommended to be used. """
+        delay_in_ms = int(round(delay_in_seconds * 1000))
+        await self._execute(
+            debug=sc_pb.RequestDebug(
+                debug=[debug_pb.DebugCommand(test_process=debug_pb.DebugTestProcess(test=1, delay_ms=delay_in_ms))]
+            )
+        )
+
+    async def debug_show_map(self):
+        """ Reveals the whole map for the bot. Using it a second time disables it again. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=1)]))
+
+    async def debug_control_enemy(self):
+        """ Allows control over enemy units and structures similar to team games control - does not allow the bot to spend the opponent's ressources. Using it a second time disables it again.  """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=2)]))
+
+    async def debug_food(self):
+        """ Should disable food usage (does not seem to work?). Using it a second time disables it again.  """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=3)]))
+
+    async def debug_free(self):
+        """ Units, structures and upgrades are free of mineral and gas cost. Using it a second time disables it again.  """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=4)]))
+
+    async def debug_all_resources(self):
+        """ Gives 5000 minerals and 5000 vespene to the bot. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=5)]))
+
+    async def debug_god(self):
+        """ Your units and structures no longer take any damage. Using it a second time disables it again. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=6)]))
+
+    async def debug_minerals(self):
+        """ Gives 5000 minerals to the bot. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=7)]))
+
+    async def debug_gas(self):
+        """ Gives 5000 vespene to the bot. This does not seem to be working. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=8)]))
+
+    async def debug_cooldown(self):
+        """ Disables cooldowns of unit abilities for the bot. Using it a second time disables it again. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=9)]))
+
+    async def debug_tech_tree(self):
+        """ Removes all tech requirements (e.g. can build a factory without having a barracks). Using it a second time disables it again. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=10)]))
+
+    async def debug_upgrade(self):
+        """ Researches all currently available upgrades. E.g. using it once unlocks combat shield, stimpack and 1-1. Using it a second time unlocks 2-2 and all other upgrades stay researched. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=11)]))
+
+    async def debug_fast_build(self):
+        """ Sets the build time of units and structures and upgrades to zero. Using it a second time disables it again. """
+        await self._execute(debug=sc_pb.RequestDebug(debug=[debug_pb.DebugCommand(game_state=12)]))
+
+    async def quick_save(self):
+        """Saves the current game state to an in-memory bookmark.
+        See: https://github.com/Blizzard/s2client-proto/blob/eeaf5efaea2259d7b70247211dff98da0a2685a2/s2clientprotocol/sc2api.proto#L93"""
+        await self._execute(quick_save=sc_pb.RequestQuickSave())
+
+    async def quick_load(self):
+        """Loads the game state from the previously stored in-memory bookmark.
+        Caution:
+            - The SC2 Client will crash if the game wasn't quicksaved
+            - The bot step iteration counter will not reset
+            - self.state.game_loop will be set to zero after the quickload, and self.time is dependant on it"""
+        await self._execute(quick_load=sc_pb.RequestQuickLoad())
+
+
+class DrawItem:
+
+    @staticmethod
+    def to_debug_color(color: Union[tuple, Point3]):
+        """ Helper function for color conversion """
+        if color is None:
+            return debug_pb.Color(r=255, g=255, b=255)
+        # Need to check if not of type Point3 because Point3 inherits from tuple
+        if isinstance(color, (tuple, list)) and not isinstance(color, Point3) and len(color) == 3:
+            return debug_pb.Color(r=color[0], g=color[1], b=color[2])
+        # In case color is of type Point3
+        r = getattr(color, "r", getattr(color, "x", 255))
+        g = getattr(color, "g", getattr(color, "y", 255))
+        b = getattr(color, "b", getattr(color, "z", 255))
+        if max(r, g, b) <= 1:
+            r *= 255
+            g *= 255
+            b *= 255
+
+        return debug_pb.Color(r=int(r), g=int(g), b=int(b))
+
+
+class DrawItemScreenText(DrawItem):
+
+    def __init__(self, start_point: Point2 = None, color: Point3 = None, text: str = "", font_size: int = 8):
+        self._start_point: Point2 = start_point
+        self._color: Point3 = color
+        self._text: str = text
+        self._font_size: int = font_size
+
+    def to_proto(self):
+        return debug_pb.DebugText(
+            color=self.to_debug_color(self._color),
+            text=self._text,
+            virtual_pos=self._start_point.to3.as_Point,
+            world_pos=None,
+            size=self._font_size,
+        )
+
+    def __hash__(self):
+        return hash((self._start_point, self._color, self._text, self._font_size))
+
+
+class DrawItemWorldText(DrawItem):
+
+    def __init__(self, start_point: Point3 = None, color: Point3 = None, text: str = "", font_size: int = 8):
+        self._start_point: Point3 = start_point
+        self._color: Point3 = color
+        self._text: str = text
+        self._font_size: int = font_size
+
+    def to_proto(self):
+        return debug_pb.DebugText(
+            color=self.to_debug_color(self._color),
+            text=self._text,
+            virtual_pos=None,
+            world_pos=self._start_point.as_Point,
+            size=self._font_size,
+        )
+
+    def __hash__(self):
+        return hash((self._start_point, self._text, self._font_size, self._color))
+
+
+class DrawItemLine(DrawItem):
+
+    def __init__(self, start_point: Point3 = None, end_point: Point3 = None, color: Point3 = None):
+        self._start_point: Point3 = start_point
+        self._end_point: Point3 = end_point
+        self._color: Point3 = color
+
+    def to_proto(self):
+        return debug_pb.DebugLine(
+            line=debug_pb.Line(p0=self._start_point.as_Point, p1=self._end_point.as_Point),
+            color=self.to_debug_color(self._color),
+        )
+
+    def __hash__(self):
+        return hash((self._start_point, self._end_point, self._color))
+
+
+class DrawItemBox(DrawItem):
+
+    def __init__(self, start_point: Point3 = None, end_point: Point3 = None, color: Point3 = None):
+        self._start_point: Point3 = start_point
+        self._end_point: Point3 = end_point
+        self._color: Point3 = color
+
+    def to_proto(self):
+        return debug_pb.DebugBox(
+            min=self._start_point.as_Point,
+            max=self._end_point.as_Point,
+            color=self.to_debug_color(self._color),
+        )
+
+    def __hash__(self):
+        return hash((self._start_point, self._end_point, self._color))
+
+
+class DrawItemSphere(DrawItem):
+
+    def __init__(self, start_point: Point3 = None, radius: float = None, color: Point3 = None):
+        self._start_point: Point3 = start_point
+        self._radius: float = radius
+        self._color: Point3 = color
+
+    def to_proto(self):
+        return debug_pb.DebugSphere(
+            p=self._start_point.as_Point, r=self._radius, color=self.to_debug_color(self._color)
+        )
+
+    def __hash__(self):
+        return hash((self._start_point, self._radius, self._color))