Move multi-addr parser mod to `tractor`

piker/cli/__init__.py

@@ -114,7 +114,7 @@ def pikerd(
             )]
 
         from .. import service
-        from ..service._multiaddr import parse_addr
+        from tractor._multiaddr import parse_addr
 
         # transport-oriented endpoint multi-addresses
         eps: dict[

piker/service/_multiaddr.py

@@ -1,142 +0,0 @@
-# piker: trading gear for hackers
-# Copyright (C) 2018-present  Tyler Goodlet (in stewardship of pikers)
-
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-
-# You should have received a copy of the GNU Affero General Public License
-# along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-'''
-Multiaddress parser and utils according the spec(s) defined by
-`libp2p` and used in dependent project such as `ipfs`:
-
-- https://docs.libp2p.io/concepts/fundamentals/addressing/
-- https://github.com/libp2p/specs/blob/master/addressing/README.md
-
-'''
-from typing import Iterator
-
-from bidict import bidict
-
-# TODO: see if we can leverage libp2p ecosys projects instead of
-# rolling our own (parser) impls of the above addressing specs:
-# - https://github.com/libp2p/py-libp2p
-# - https://docs.libp2p.io/concepts/nat/circuit-relay/#relay-addresses
-# prots: bidict[int, str] = bidict({
-prots: bidict[int, str] = {
-    'ipv4': 3,
-    'ipv6': 3,
-    'wg': 3,
-
-    'tcp': 4,
-    'udp': 4,
-
-    # TODO: support the next-gen shite Bo
-    # 'quic': 4,
-    # 'ssh': 7,  # via rsyscall bootstrapping
-}
-
-prot_params: dict[str, tuple[str]] = {
-    'ipv4': ('addr',),
-    'ipv6': ('addr',),
-    'wg': ('addr', 'port', 'pubkey'),
-
-    'tcp': ('port',),
-    'udp': ('port',),
-
-    # 'quic': ('port',),
-    # 'ssh': ('port',),
-}
-
-
-def iter_prot_layers(
-    multiaddr: str,
-) -> Iterator[
-    tuple[
-        int,
-        list[str]
-    ]
-]:
-    '''
-    Unpack a libp2p style "multiaddress" into multiple "segments"
-    for each "layer" of the protocoll stack (in OSI terms).
-
-    '''
-    tokens: list[str] = multiaddr.split('/')
-    root, tokens = tokens[0], tokens[1:]
-    assert not root  # there is a root '/' on LHS
-    itokens = iter(tokens)
-
-    prot: str | None = None
-    params: list[str] = []
-    for token in itokens:
-        # every prot path should start with a known
-        # key-str.
-        if token in prots:
-            if prot is None:
-                prot: str = token
-            else:
-                yield prot, params
-                prot = token
-
-            params = []
-
-        elif token not in prots:
-            params.append(token)
-
-    else:
-        yield prot, params
-
-
-def parse_addr(
-    multiaddr: str,
-) -> dict[str, str | int | dict]:
-    '''
-    Parse a libp2p style "multiaddress" into it's distinct protocol
-    segments where each segment:
-
-        `../<protocol>/<param0>/<param1>/../<paramN>`
-
-    is loaded into a layers `dict[str, dict[str, Any]` which holds
-    each prot segment of the path as a separate entry sortable by
-    it's approx OSI "layer number".
-
-    Any `paramN` in the path must be distinctly defined in order
-    according to the (global) `prot_params` table in this module.
-
-    '''
-    layers: dict[str, str | int | dict] = {}
-    for (
-        prot_key,
-        params,
-    ) in iter_prot_layers(multiaddr):
-
-        layer: int = prots[prot_key]  # OSI layer used for sorting
-        ep: dict[str, int | str] = {'layer': layer}
-        layers[prot_key] = ep
-
-        # TODO; validation and resolving of names:
-        # - each param via a validator provided as part of the
-        #   prot_params def? (also see `"port"` case below..)
-        # - do a resolv step that will check addrs against
-        #   any loaded network.resolv: dict[str, str]
-        rparams: list = list(reversed(params))
-        for key in prot_params[prot_key]:
-            val: str | int = rparams.pop()
-
-            # TODO: UGHH, dunno what we should do for validation
-            # here, put it in the params spec somehow?
-            if key == 'port':
-                val = int(val)
-
-            ep[key] = val
-
-    return layers