feat: add ip address support

mafic/ip.py

@@ -0,0 +1,122 @@
+# SPDX-License-Identifier: MIT
+
+from __future__ import annotations
+
+from abc import ABC
+from datetime import datetime
+from enum import Enum
+from typing import TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    from .typings import (
+        BalancingIPRouteDetails,
+        BaseDetails,
+        FailingIPAddress,
+        IPBlock as IPBlockPayload,
+        NanoIPRouteDetails,
+        RotatingIPRouteDetails,
+        RotatingNanoIPRouteDetails,
+    )
+
+__all__ = (
+    "IPRoutePlannerType",
+    "IPBlockType",
+    "IPBlock",
+    "FailingAddress",
+    "BaseIPRoutePlannerStatus",
+    "BalancingIPRoutePlannerStatus",
+    "NanoIPRoutePlannerStatus",
+    "RotatingIPRoutePlannerStatus",
+    "RotatingNanoIPRoutePlannerStatus",
+    "RoutePlannerStatus",
+)
+
+
+class IPRoutePlannerType(Enum):
+    ROTATING_IP = "RotatingIPRoutePlanner"
+    NANO_IP = "NanoIPRoutePlanner"
+    ROTATING_NANO_IP = "RotatingNanoIPRoutePlanner"
+    BALANCING_IP = "BalancingIPRoutePlanner"
+
+
+class IPBlockType(Enum):
+    V4 = "Inet4Address"
+    V6 = "Inet6Address"
+
+
+class IPBlock:
+    __slots__ = ("size", "type")
+
+    def __init__(self, data: IPBlockPayload) -> None:
+        self.size: int = int(data["size"])
+        self.type: IPBlockType = IPBlockType(data["type"])
+
+
+class FailingAddress:
+    __slots__ = ("address", "time")
+
+    def __init__(self, data: FailingIPAddress) -> None:
+        self.address: str = data["address"]
+        self.time: datetime = datetime.fromtimestamp(data["failingTimestamp"])
+
+
+class BaseIPRoutePlannerStatus(ABC):
+    __slots__ = ("failing_addresses", "ip_block")
+
+    type: IPRoutePlannerType
+
+    def __init__(self, data: BaseDetails) -> None:
+        self.ip_block = IPBlock(data["ipBlock"])
+        self.failing_addresses = [
+            FailingAddress(addr) for addr in data["failingAddresses"]
+        ]
+
+
+class RotatingIPRoutePlannerStatus(BaseIPRoutePlannerStatus):
+    __slots__ = ("current_address", "ip_index", "rotate_index")
+
+    type = IPRoutePlannerType.ROTATING_IP
+
+    def __init__(self, data: RotatingIPRouteDetails) -> None:
+        super().__init__(data)
+        self.rotate_index: int = int(data["rotateIndex"])
+        self.ip_index: int = int(data["ipIndex"])
+        self.current_address: str = data["currentAddress"]
+
+
+class NanoIPRoutePlannerStatus(BaseIPRoutePlannerStatus):
+    __slots__ = ("current_address_index",)
+
+    type = IPRoutePlannerType.NANO_IP
+
+    def __init__(self, data: NanoIPRouteDetails) -> None:
+        super().__init__(data)
+        self.current_address_index: int = int(data["currentAddressIndex"])
+
+
+class RotatingNanoIPRoutePlannerStatus(BaseIPRoutePlannerStatus):
+    __slots__ = ("block_index", "current_address_index")
+
+    type = IPRoutePlannerType.ROTATING_NANO_IP
+
+    def __init__(self, data: RotatingNanoIPRouteDetails) -> None:
+        super().__init__(data)
+        self.block_index: int = int(data["blockIndex"])
+        self.current_address_index: int = int(data["currentAddressIndex"])
+
+
+class BalancingIPRoutePlannerStatus(BaseIPRoutePlannerStatus):
+    __slots__ = ("current_address_index", "ip_index", "rotate_index")
+
+    type = IPRoutePlannerType.BALANCING_IP
+
+    def __init__(self, data: BalancingIPRouteDetails) -> None:
+        super().__init__(data)
+
+
+RoutePlannerStatus = Union[
+    RotatingIPRoutePlannerStatus,
+    NanoIPRoutePlannerStatus,
+    RotatingNanoIPRoutePlannerStatus,
+    BalancingIPRoutePlannerStatus,
+]

mafic/node.py

@@ -13,6 +13,12 @@
 
 from .__libraries import ExponentialBackoff, dumps, loads
 from .errors import TrackLoadException
+from .ip import (
+    BalancingIPRoutePlannerStatus,
+    NanoIPRoutePlannerStatus,
+    RotatingIPRoutePlannerStatus,
+    RotatingNanoIPRoutePlannerStatus,
+)
 from .playlist import Playlist
 from .plugin import Plugin
 from .track import Track
@@ -25,15 +31,21 @@
 
     from .__libraries import Client, VoiceServerUpdatePayload
     from .filter import Filter
+    from .ip import RoutePlannerStatus
     from .player import Player
     from .typings import (
+        BalancingIPRouteDetails,
         Coro,
         EventPayload,
         GetTracks,
         IncomingMessage,
+        NanoIPRouteDetails,
         OutgoingMessage,
         PlayPayload,
         PluginData,
+        RotatingIPRouteDetails,
+        RotatingNanoIPRouteDetails,
+        RoutePlannerStatus as RoutePlannerStatusPayload,
         TrackInfo,
     )
 
@@ -486,6 +498,34 @@ async def fetch_plugins(self) -> list[Plugin]:
 
         return [Plugin(**plugins) for plugins in plugins]
 
-    # TODO: route planner status
-    # TODO: unmark failed address
-    # TODO: unmark all failed addresses
+    async def fetch_route_planner_status(self) -> RoutePlannerStatus | None:
+        data: RoutePlannerStatusPayload = await self.__request(
+            "GET", "/routeplanner/status"
+        )
+
+        if data["class"] == "RotatingIpRoutePlanner":
+            return RotatingIPRoutePlannerStatus(
+                cast(RotatingIPRouteDetails, data["details"])
+            )
+        elif data["class"] == "NanoIpRoutePlanner":
+            return NanoIPRoutePlannerStatus(cast(NanoIPRouteDetails, data["details"]))
+        elif data["class"] == "RotatingNanoIpRoutePlanner":
+            return RotatingNanoIPRoutePlannerStatus(
+                cast(RotatingNanoIPRouteDetails, data["details"])
+            )
+        elif data["class"] == "BalancingIpRoutePlanner":
+            return BalancingIPRoutePlannerStatus(
+                cast(BalancingIPRouteDetails, data["details"])
+            )
+        elif data["class"] is None:
+            return None
+        else:
+            raise RuntimeError(f"Unknown route planner class {data['class']}.")
+
+    async def unmark_failed_address(self, address: str) -> None:
+        await self.__request(
+            "POST", "/routeplanner/free/address", json={"address": address}
+        )
+
+    async def unmark_all_addresses(self) -> None:
+        await self.__request("POST", "/routeplanner/free/all")

mafic/typings/http.py

@@ -10,9 +10,22 @@
     from .misc import FriendlyException
 
 __all__ = (
+    "BalancingIPRouteDetails",
+    "BalancingIPRoutePlanner",
+    "BaseDetails",
+    "EmptyRoutePlanner",
+    "FailingIPAddress",
     "GetTracks",
+    "IPBlock",
+    "NanoIPRouteDetails",
+    "NanoIPRoutePlanner",
     "PlaylistInfo",
     "PluginData",
+    "RotatingIPRouteDetails",
+    "RotatingIPRoutePlanner",
+    "RotatingNanoIPRouteDetails",
+    "RotatingNanoIPRoutePlanner",
+    "RoutePlannerStatus",
     "Tracks",
     "TrackInfo",
     "TrackWithInfo",
@@ -61,3 +74,86 @@ class TracksFailed(TypedDict):
 class PluginData(TypedDict):
     name: str
     version: str
+
+
+class IPBlock(TypedDict):
+    type: Literal["Inet4Address", "Inet6Address"]
+    size: str
+
+
+class FailingIPAddress(TypedDict):
+    address: str
+    failingTimestamp: int
+    failingTime: str
+
+
+class BaseDetails(TypedDict):
+    ipBlock: IPBlock
+    failingAddresses: list[FailingIPAddress]
+
+
+class RotatingIPRouteDetails(BaseDetails):
+    rotateIndex: str
+    ipIndex: str
+    currentAddress: str
+
+
+# Fields are named class
+RotatingIPRoutePlanner = TypedDict(
+    "RotatingIPRoutePlanner",
+    {
+        "class": Literal["RotatingIpRoutePlanner"],
+        "details": RotatingIPRouteDetails,
+    },
+)
+
+
+class NanoIPRouteDetails(BaseDetails):
+    currentAddressIndex: str
+
+
+NanoIPRoutePlanner = TypedDict(
+    "NanoIPRoutePlanner",
+    {
+        "class": Literal["NanoIpRoutePlanner"],
+        "details": NanoIPRouteDetails,
+    },
+)
+
+
+class RotatingNanoIPRouteDetails(BaseDetails):
+    blockIndex: str
+    currentAddressIndex: str
+
+
+RotatingNanoIPRoutePlanner = TypedDict(
+    "RotatingNanoIPRoutePlanner",
+    {
+        "class": Literal["RotatingNanoIpRoutePlanner"],
+        "details": RotatingNanoIPRouteDetails,
+    },
+)
+
+
+class BalancingIPRouteDetails(BaseDetails):
+    ...
+
+
+BalancingIPRoutePlanner = TypedDict(
+    "BalancingIPRoutePlanner",
+    {
+        "class": Literal["BalancingIpRoutePlanner"],
+        "details": BalancingIPRouteDetails,
+    },
+)
+
+
+EmptyRoutePlanner = TypedDict("EmptyRoutePlanner", {"class": None, "details": None})
+
+RoutePlannerStatus = Union[
+    RotatingIPRoutePlanner,
+    NanoIPRoutePlanner,
+    RotatingNanoIPRoutePlanner,
+    BalancingIPRoutePlanner,
+    EmptyRoutePlanner,
+]