Factor out device-to-court mapping

Author: madduck

pyproject.toml

@@ -13,7 +13,7 @@ write_to = "_version.py"
 
 [project]
 name = "tptools"
-version = "0.6.6"
+version = "0.6.7"
 authors = [{ name = "martin f. krafft", email = "[email protected]" }]
 description = "A set of tools to export data from with TournamentSoftware"
 readme = "README.md"

tests/test_devcourtmap.py

@@ -0,0 +1,67 @@
+from io import BytesIO
+
+import pytest
+
+from tptools.court import Court
+from tptools.devcourtmap import DeviceCourtMap
+
+
+@pytest.fixture
+def tomldevmap() -> BytesIO:
+    devmap = {
+        "192.0.2.1": "C1",
+        "192.0.2.2": "C07",
+    }
+    toml = "\n".join([f'{ip} = "{court}"' for ip, court in devmap.items()])
+    return BytesIO(toml.encode())
+
+
+def test_toml_reader(tomldevmap: BytesIO) -> None:
+    devmap = DeviceCourtMap.read_toml_devmap(tomldevmap)
+    assert "192.0.2.1" in devmap
+    assert devmap["192.0.2.2"] == "C07"
+
+
+@pytest.mark.parametrize(
+    "input, exp",
+    [
+        ("C1", (None, 1)),
+        ("2-C1", (2, 1)),
+        ("-3-C4", (3, 4)),
+        ("C01", (None, 1)),
+        ("C42", (None, 42)),
+        ("c2", (None, 2)),
+        ("court3", (None, 3)),
+        ("Court4", (None, 4)),
+        ("COURT5", (None, 5)),
+        ("Court 6", (None, 6)),
+        ("Court 08", (None, 8)),
+        ("C 7", None),
+        ("Cou7", None),
+    ],
+)
+def test_name_normaliser(input: str, exp: tuple[int | None, int] | None) -> None:
+    assert DeviceCourtMap.normalise_court_name_for_matching(input) == exp
+
+
+@pytest.fixture
+def devcourtmap(tomldevmap: BytesIO) -> DeviceCourtMap:
+    return DeviceCourtMap(tomldevmap)
+
+
+@pytest.mark.parametrize(
+    "ip, text", [("192.0.2.1", "C1"), ("192.0.2.2", "C07"), ("192.0.2.3", None)]
+)
+def test_find_match(devcourtmap: DeviceCourtMap, ip: str, text: str | None) -> None:
+    assert devcourtmap.find_match_for_ip(ip) == text
+
+
+@pytest.mark.parametrize(
+    "ip, court",
+    [("192.0.2.1", "court1"), ("192.0.2.2", "court2"), ("192.0.2.3", "court3")],
+)
+def test_find_court(
+    devcourtmap: DeviceCourtMap, ip: str, court: str, court1: Court, court2: Court
+) -> None:
+    cm = {"court1": court1, "court2": court2}
+    assert devcourtmap.find_court_for_ip(ip, cm.values()) == cm.get(court)

tptools/__init__.py

@@ -1,4 +1,5 @@
 from .court import Court, CourtSelectionParams
+from .devcourtmap import DeviceCourtMap
 from .draw import Draw, Event, Stage
 from .drawtype import DrawType
 from .entry import Entry, Player
@@ -10,6 +11,7 @@
 __all__ = [
     "Court",
     "CourtSelectionParams",
+    "DeviceCourtMap",
     "Draw",
     "DrawType",
     "Entry",

tptools/devcourtmap.py

@@ -0,0 +1,85 @@
+import logging
+import re
+import tomllib
+from collections.abc import Iterable
+from typing import IO, Never
+
+from tptools.court import Court
+from tptools.util import flatten_dict
+
+logger = logging.getLogger(__name__)
+
+
+class DeviceCourtMap:
+    def __init__(self, tomlfile: IO[bytes] | None = None) -> None:
+        self._devmap: dict[str, int | str] = (
+            {} if tomlfile is None else self.read_toml_devmap(tomlfile)
+        )
+
+    @staticmethod
+    def read_toml_devmap(tomlfile: IO[bytes]) -> dict[str, int | str] | Never:
+        devmap: dict[str, int | str] = tomllib.load(tomlfile)
+
+        # TOML turns a key like 172.21.22.1 into a nested dict {"172":{"21":{"22"…}…}…}
+        # so for convenience, flatten this:
+        return flatten_dict(devmap)
+
+    @staticmethod
+    def normalise_court_name_for_matching(
+        courtname: str,
+    ) -> tuple[int | None, int] | None:
+        m = re.match(
+            r"(?:-?(?P<loc>\d+)-)?c(?:ourt *)?0*(?P<court>\d+)",
+            courtname,
+            flags=re.IGNORECASE,
+        )
+        if m:
+            return int(loc) if (loc := m["loc"]) is not None else None, int(m["court"])
+        else:
+            return None
+
+    def find_court_for_ip(
+        self, clientip: str, courts: Iterable[Court] | None = None
+    ) -> Court | None:
+        courtname = None
+        if (courtname := self._devmap.get(clientip)) is not None:
+            logger.debug(f"Device at IP {clientip} might be on {courtname}")
+            for court in courts or []:
+                # TODO: can we do better than this to identify the court when we are
+                # given a string that might not be what the current courtnamepolicy
+                # returns, or an ID?
+                if isinstance(courtname, int):
+                    if courtname == court.id:
+                        return court
+
+                else:
+                    term = self.normalise_court_name_for_matching(courtname)
+                    if term is not None:
+                        comps = (
+                            []
+                            if court.location is None
+                            else [
+                                f"{court.location.id}-{court.name}",
+                                f"{court.location.id}-{court}",
+                            ]
+                        )
+
+                        for comparename in (
+                            *comps,
+                            court.name,
+                            str(court),
+                        ):
+                            comp = self.normalise_court_name_for_matching(comparename)
+                            if (
+                                comp is not None
+                                and (term[0] is None and term[1] == comp[1])
+                                or (term == comp)
+                            ):
+                                return court
+
+        logger.debug(f"No court found in devmap for device with IP {clientip}")
+        return None
+
+    def find_match_for_ip(self, clientip: str) -> str | None:
+        text = self._devmap.get(clientip)
+        return str(text) if text is not None else None

tptools/tpsrv/squoresrv.py

@@ -31,6 +31,7 @@
 from tptools import (
     Court,
     CourtSelectionParams,
+    DeviceCourtMap,
     Draw,
     Entry,
     MatchSelectionParams,
@@ -59,7 +60,6 @@
 )
 from tptools.namepolicy.policybase import RegexpSubstTuple
 from tptools.util import (
-    flatten_dict,
     normalise_dict_values_for_query_string,
     silence_logger,
 )
@@ -413,93 +413,40 @@ def get_config(
 
 def get_dev_map(
     path: Annotated[pathlib.Path, Depends(get_devmap_path)],
-) -> dict[str, int | str] | Never:
-    devmap: dict[str, int | str] = {}
-
+) -> DeviceCourtMap:
     try:
         with open(path, "rb") as f:
-            devmap = tomllib.load(f)
+            return DeviceCourtMap(f)
 
     except FileNotFoundError:
         logger.warning(f"Squore device to court map not found at {path}, ignoring…")
 
-    # TOML turns a key like 172.21.22.1 into a nested dict {"172":{"21":{"22"…}
-    # so for convenience, flatten this:
-    return flatten_dict(devmap)
+    return DeviceCourtMap()
 
 
 # }}}
 
 # {{{ Court name to dev/mirror
 
 
-def _normalise_court_name_for_matching(courtname: str) -> tuple[int | None, str] | None:
-    m = re.match(
-        r"(?:-?(?P<loc>\d+)-)?c(?:ourt *)?0*(?P<court>\d+)",
-        courtname,
-        flags=re.IGNORECASE,
-    )
-    if m:
-        return int(loc) if (loc := m["loc"]) is not None else None, m["court"]
-    else:
-        return None
-
-
 def get_court_for_dev(
-    dev_map: Annotated[dict[str, int | str], Depends(get_dev_map)],
+    dev_map: Annotated[DeviceCourtMap, Depends(get_dev_map)],
     courts: Annotated[list[Court], Depends(get_courts)],
     clientip: Annotated[str, Depends(get_remote)],
 ) -> Court | None:
-    courtname = None
-    if (courtname := dev_map.get(clientip)) is not None:
-        logger.debug(f"Device at IP {clientip} might want feed for {courtname}")
-        for court in courts:
-            # TODO: can we do better than this to identify the court when we are
-            # given a string that might not be what the current courtnamepolicy
-            # returns, or an ID?
-            if isinstance(courtname, int):
-                if courtname == court.id:
-                    return court
-
-            else:
-                term = _normalise_court_name_for_matching(courtname)
-                if term is not None:
-                    comps = (
-                        []
-                        if court.location is None
-                        else [
-                            f"{court.location.id}-{court.name}",
-                            f"{court.location.id}-{court}",
-                        ]
-                    )
-
-                    for comparename in (
-                        *comps,
-                        court.name,
-                        str(court),
-                    ):
-                        comp = _normalise_court_name_for_matching(comparename)
-                        if (
-                            comp is not None
-                            and (term[0] is None and term[1] == comp[1])
-                            or (term == comp)
-                        ):
-                            return court
-
-    logger.debug(f"No court found in devmap for device with IP {clientip}")
-    return None
+    return dev_map.find_court_for_ip(clientip, courts=courts)
 
 
 def get_mirror_for_dev(
-    dev_map: Annotated[dict[str, int | str], Depends(get_dev_map)],
+    dev_map: Annotated[DeviceCourtMap, Depends(get_dev_map)],
     clientip: Annotated[str, Depends(get_remote)],
 ) -> str | None:
     othername = None
-    if (othername := dev_map.get(clientip)) is not None and re.fullmatch(
+    if (othername := dev_map.find_match_for_ip(clientip)) is not None and re.fullmatch(
         r"\w{6}-\d+-.+", othername := str(othername)
     ):
         logger.debug(f"Device at IP {clientip} wants to be mirror for {othername}")
-        return othername  # type: ignore[return-value]
+        return othername
 
     logger.debug(f"No mirror device found in devmap for device with IP {clientip}")
     return None