import re from ipaddress import IPv4Address from uuid import UUID from orchestrator.forms.validators import Choice from gso.products.product_blocks.router import RouterVendor from gso.products.product_types.router import Router from gso.services.crm import all_customers from gso.services.netbox_client import NetBoxClient def customer_selector() -> Choice: customers = {} for customer in all_customers(): customers[customer["id"]] = customer["name"] return Choice("Select a customer", zip(customers.keys(), customers.items())) # type: ignore[arg-type] def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None: """Return a list of available interfaces for a given router and speed. For Nokia routers, return a list of available interfaces. For Juniper routers, return a string. """ if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA: return None interfaces = { interface["name"]: f"{interface['name']} - {interface['module']['display']} - {interface['description']}" for interface in NetBoxClient().get_available_interfaces(router_id, speed) } return Choice("ae member", zip(interfaces.keys(), interfaces.items())) # type: ignore[arg-type] def available_lags_choices(router_id: UUID) -> Choice | None: """Return a list of available lags for a given router. For Nokia routers, return a list of available lags. For Juniper routers, return a string. """ if Router.from_subscription(router_id).router.router_vendor != RouterVendor.NOKIA: return None side_a_ae_iface_list = NetBoxClient().get_available_lags(router_id) return Choice("ae iface", zip(side_a_ae_iface_list, side_a_ae_iface_list)) # type: ignore[arg-type] def get_router_vendor(router_id: UUID) -> str: """Retrieve the vendor of a router. Args: ---- router_id (UUID): The {term}`UUID` of the router. Returns: ------- str: The vendor of the router. """ return Router.from_subscription(router_id).router.router_vendor def iso_from_ipv4(ipv4_address: IPv4Address) -> str: """Calculate an :term:`ISO` address, based on an IPv4 address. :param IPv4Address ipv4_address: The address that's to be converted :returns: An :term:`ISO`-formatted address. """ padded_octets = [f"{x:>03}" for x in str(ipv4_address).split(".")] joined_octets = "".join(padded_octets) re_split = ".".join(re.findall("....", joined_octets)) return ".".join(["49.51e5.0001", re_split, "00"])