Add hotspot vouchers

aiounifi/controller.py

@@ -22,6 +22,7 @@
 from .interfaces.system_information import SystemInformationHandler
 from .interfaces.traffic_routes import TrafficRoutes
 from .interfaces.traffic_rules import TrafficRules
+from .interfaces.vouchers import Vouchers
 from .interfaces.wlans import Wlans
 from .models.configuration import Configuration
 
@@ -53,6 +54,7 @@ def __init__(self, config: Configuration) -> None:
         self.system_information = SystemInformationHandler(self)
         self.traffic_rules = TrafficRules(self)
         self.traffic_routes = TrafficRoutes(self)
+        self.vouchers = Vouchers(self)
         self.wlans = Wlans(self)
 
         self.update_handlers: tuple[Callable[[], Coroutine[Any, Any, None]], ...] = (
@@ -66,6 +68,7 @@ def __init__(self, config: Configuration) -> None:
             self.system_information.update,
             self.traffic_rules.update,
             self.traffic_routes.update,
+            self.vouchers.update,
             self.wlans.update,
         )
 

aiounifi/interfaces/vouchers.py

@@ -0,0 +1,46 @@
+"""Hotspot vouchers as part of a UniFi network."""
+
+from ..models.api import TypedApiResponse
+from ..models.message import MessageKey
+from ..models.voucher import (
+    Voucher,
+    VoucherCreateRequest,
+    VoucherDeleteRequest,
+    VoucherListRequest,
+)
+from .api_handlers import APIHandler
+
+
+class Vouchers(APIHandler[Voucher]):
+    """Represents Hotspot vouchers."""
+
+    obj_id_key = "_id"
+    item_cls = Voucher
+    process_messages = (MessageKey.VOUCHER_CREATED,)
+    remove_messages = (MessageKey.VOUCHER_DELETED,)
+    api_request = VoucherListRequest.create()
+
+    async def create(self, voucher: Voucher) -> TypedApiResponse:
+        """Create voucher on controller."""
+        return await self.controller.request(
+            VoucherCreateRequest.create(
+                number=1,
+                quota=voucher.quota,
+                expire_number=int(
+                    voucher.duration.total_seconds() / 60  # Get minutes.
+                ),
+                expire_unit=1,
+                usage_quota=voucher.qos_usage_quota,
+                rate_max_up=voucher.qos_rate_max_up,
+                rate_max_down=voucher.qos_rate_max_down,
+                note=voucher.note,
+            )
+        )
+
+    async def delete(self, voucher: Voucher) -> TypedApiResponse:
+        """Delete voucher from controller."""
+        return await self.controller.request(
+            VoucherDeleteRequest.create(
+                obj_id=voucher.id,
+            )
+        )

aiounifi/models/message.py

@@ -50,6 +50,8 @@ class MessageKey(enum.Enum):
     VPN_CONNECT = "vpn:connect"
     VPN_DISCONNECT = "vpn:disconnect"
     VPN_CONNECTION_UPDATED = "vpn-connection:sync"
+    VOUCHER_CREATED = "voucher:create"
+    VOUCHER_DELETED = "voucher:delete"
     WLAN_CONF_ADDED = "wlanconf:add"
     WLAN_CONF_UPDATED = "wlanconf:sync"
     WLAN_CONF_DELETED = "wlanconf:delete"

aiounifi/models/voucher.py

@@ -0,0 +1,219 @@
+"""Hotspot vouchers as part of a UniFi network."""
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import NotRequired, Self, TypedDict
+
+from .api import ApiItem, ApiRequest
+
+
+class TypedVoucher(TypedDict):
+    """Voucher type definition."""
+
+    _id: str
+    site_id: str
+    note: NotRequired[str]
+    code: str
+    quota: int
+    duration: float
+    qos_overwrite: NotRequired[bool]
+    qos_usage_quota: NotRequired[str]
+    qos_rate_max_up: NotRequired[int]
+    qos_rate_max_down: NotRequired[int]
+    used: int
+    create_time: float
+    start_time: NotRequired[float]
+    end_time: NotRequired[float]
+    for_hotspot: NotRequired[bool]
+    admin_name: str
+    status: str
+    status_expires: float
+
+
+@dataclass
+class VoucherListRequest(ApiRequest):
+    """Request object for voucher list."""
+
+    @classmethod
+    def create(cls) -> Self:
+        """Create voucher list request."""
+        return cls(
+            method="get",
+            path="/stat/voucher",
+        )
+
+
+@dataclass
+class VoucherCreateRequest(ApiRequest):
+    """Request object for voucher create."""
+
+    @classmethod
+    def create(
+        cls,
+        number: int,
+        quota: int,
+        expire_number: int,
+        expire_unit: int = 1,
+        usage_quota: int | None = None,
+        rate_max_up: int | None = None,
+        rate_max_down: int | None = None,
+        note: str | None = None,
+    ) -> Self:
+        """Create voucher create request.
+
+        :param number: number of vouchers
+        :param quota: number of using; 0 = unlimited
+        :param expire_number: expiration of voucher per expire_unit
+        :param expire_unit: scale of expire_number, 1 = minute, 60 = hour, 3600 = day
+        :param usage_quota: quantity of bytes allowed in MB
+        :param rate_max_up: up speed allowed in kbps
+        :param rate_max_down: down speed allowed in kbps
+        :param note: description
+        """
+        data = {
+            "cmd": "create-voucher",
+            "n": number,
+            "quota": quota,
+            "expire_number": expire_number,
+            "expire_unit": expire_unit,
+        }
+        if usage_quota:
+            data["bytes"] = usage_quota
+        if rate_max_up:
+            data["up"] = rate_max_up
+        if rate_max_down:
+            data["down"] = rate_max_down
+        if note:
+            data["note"] = note
+
+        return cls(
+            method="post",
+            path="/cmd/hotspot",
+            data=data,
+        )
+
+
+@dataclass
+class VoucherDeleteRequest(ApiRequest):
+    """Request object for voucher delete."""
+
+    @classmethod
+    def create(
+        cls,
+        obj_id: str,
+    ) -> Self:
+        """Create voucher delete request."""
+        data = {
+            "cmd": "delete-voucher",
+            "_id": obj_id,
+        }
+        return cls(
+            method="post",
+            path="/cmd/hotspot",
+            data=data,
+        )
+
+
+class Voucher(ApiItem):
+    """Represents a voucher."""
+
+    raw: TypedVoucher
+
+    @property
+    def id(self) -> str:
+        """ID of voucher."""
+        return self.raw["_id"]
+
+    @property
+    def site_id(self) -> str:
+        """Site ID."""
+        return self.raw["site_id"]
+
+    @property
+    def note(self) -> str:
+        """Note describing voucher."""
+        return self.raw.get("note") or ""
+
+    @property
+    def code(self) -> str:
+        """Code in known format 00000-00000."""
+        if len(c := self.raw.get("code", "")) > 5:
+            # API returns the code without a hyphen. But this is necessary. Separate the API string after the fifth digit.
+            return f"{c[:5]}-{c[5:]}"
+        return c
+
+    @property
+    def quota(self) -> int:
+        """Allowed usages (0 = unlimited)."""
+        return self.raw["quota"]
+
+    @property
+    def duration(self) -> timedelta:
+        """Expiration of voucher."""
+        return timedelta(minutes=self.raw["duration"])
+
+    @property
+    def qos_overwrite(self) -> bool:
+        """QoS defaults overwritten."""
+        return self.raw.get("qos_overwrite", False)
+
+    @property
+    def qos_usage_quota(self) -> int:
+        """Quantity of bytes allowed in MB."""
+        return int(self.raw.get("qos_usage_quota", 0))
+
+    @property
+    def qos_rate_max_up(self) -> int:
+        """Up speed allowed in kbps."""
+        return self.raw.get("qos_rate_max_up", 0)
+
+    @property
+    def qos_rate_max_down(self) -> int:
+        """Down speed allowed in kbps."""
+        return self.raw.get("qos_rate_max_down", 0)
+
+    @property
+    def used(self) -> int:
+        """Number of usages."""
+        return self.raw["used"]
+
+    @property
+    def create_time(self) -> datetime:
+        """Create datetime."""
+        return datetime.fromtimestamp(self.raw["create_time"])
+
+    @property
+    def start_time(self) -> datetime | None:
+        """Start datetime of first usage."""
+        if "start_time" in self.raw:
+            return datetime.fromtimestamp(self.raw["start_time"])
+        return None
+
+    @property
+    def end_time(self) -> datetime | None:
+        """End datetime of latest usage."""
+        if "end_time" in self.raw:
+            return datetime.fromtimestamp(self.raw["end_time"])
+        return None
+
+    @property
+    def for_hotspot(self) -> bool:
+        """For hotspot."""
+        return self.raw.get("for_hotspot", False)
+
+    @property
+    def admin_name(self) -> str:
+        """Admin name."""
+        return self.raw["admin_name"]
+
+    @property
+    def status(self) -> str:
+        """Status."""
+        return self.raw["status"]
+
+    @property
+    def status_expires(self) -> timedelta | None:
+        """Status expires in seconds."""
+        if (status_expiry := self.raw["status_expires"]) > 0:
+            return timedelta(seconds=status_expiry)
+        return None

tests/conftest.py

@@ -103,6 +103,7 @@ def _endpoint_fixture(
     traffic_rule_payload: list[dict[str, Any]],
     traffic_rule_status: int,
     wlan_payload: list[dict[str, Any]],
+    voucher_payload: list[dict[str, Any]],
 ) -> None:
     """Use fixtures to mock all endpoints."""
 
@@ -177,6 +178,11 @@ def mock_get_request(
         "/proxy/network/api/s/default/rest/wlanconf",
         wlan_payload,
     )
+    mock_get_request(
+        "/api/s/default/stat/voucher",
+        "/proxy/network/api/s/default/stat/voucher",
+        voucher_payload,
+    )
 
 
 @pytest.fixture(name="response_payload")
@@ -255,3 +261,9 @@ def traffic_rule_status_fixture() -> int:
 def wlan_data_fixture() -> list[dict[str, Any]]:
     """WLAN data."""
     return []
+
+
+@pytest.fixture(name="voucher_payload")
+def voucher_data_fixture() -> list[dict[str, Any]]:
+    """Vouchers data."""
+    return []

tests/fixtures.py

@@ -4110,6 +4110,44 @@
     },
 ]
 
+VOUCHERS = [
+    {
+        "_id": "657e370a4543a555901865c7",
+        "admin_name": "Admin",
+        "code": "7470075124",
+        "create_time": 1638342818,
+        "duration": 5184000,
+        "end_time": 1949382832,
+        "for_hotspot": False,
+        "note": "auto-generated",
+        "qos_overwrite": False,
+        "quota": 0,
+        "site_id": "5a32aa4ee4b0412345678910",
+        "start_time": 1638342832,
+        "status": "USED_MULTIPLE",
+        "status_expires": 244679302,
+        "used": 2,
+    },
+    {
+        "_id": "61facea3873fdb075ce28d71",
+        "admin_name": "Admin",
+        "code": "44703",
+        "create_time": 1643826851,
+        "duration": 480,
+        "for_hotspot": False,
+        "note": None,
+        "qos_overwrite": True,
+        "qos_rate_max_down": 5000,
+        "qos_rate_max_up": 2000,
+        "qos_usage_quota": 1000,
+        "quota": 1,
+        "site_id": "5a32aa4ee4b0412345678910",
+        "status": "VALID_ONE",
+        "status_expires": 0,
+        "used": 0,
+    },
+]
+
 WLANS = [
     {
         "_id": "012345678910111213141516",