Android-pyApi/android_pyapi/android/net_ethernet.py

"""
Android net 以太网组件 — EthernetManager, EthernetNetworkSpecifier,
EthernetConfig, IpConfiguration, StaticIpConfiguration
方法签名完全对齐 Android SDK
"""

from __future__ import annotations

from typing import Any, Dict, List, Optional, Tuple

from .._core import Stub, AndroidOnlyError

__all__ = [
    "EthernetManager",
    "EthernetNetworkSpecifier",
    "EthernetNetworkUpdate",
    "EthernetConfig",
    "IpConfiguration",
    "StaticIpConfiguration",
]


# ═══════════════════════════════════════════════════════════════
# 常量
# ═══════════════════════════════════════════════════════════════

ETH_STATE_DISCONNECTED = 0
ETH_STATE_CONNECTING = 1
ETH_STATE_CONNECTED = 2
ETH_STATE_SUSPENDED = 3

ETH_MODE_DHCP = 0
ETH_MODE_STATIC = 1

ETH_LINK_MODE_AUTO = 0
ETH_LINK_MODE_10_HALF = 1
ETH_LINK_MODE_10_FULL = 2
ETH_LINK_MODE_100_HALF = 3
ETH_LINK_MODE_100_FULL = 4
ETH_LINK_MODE_1000_HALF = 5
ETH_LINK_MODE_1000_FULL = 6
ETH_LINK_MODE_2500_FULL = 7
ETH_LINK_MODE_5000_FULL = 8
ETH_LINK_MODE_10000_FULL = 9

ETH_ERROR_NONE = 0
ETH_ERROR_INTERFACE_NOT_FOUND = -1
ETH_ERROR_CONFIG_FAILED = -2
ETH_ERROR_LINK_DOWN = -3
ETH_ERROR_NOT_SUPPORTED = -4
ETH_ERROR_PERMISSION_DENIED = -5

ETH_CAPABILITY_WOL = 1
ETH_CAPABILITY_VLAN = 2
ETH_CAPABILITY_JUMBO_FRAMES = 4
ETH_CAPABILITY_EEE = 8
ETH_CAPABILITY_PTP = 16

ETH_DUPLEX_HALF = 0
ETH_DUPLEX_FULL = 1
ETH_DUPLEX_AUTO = 2

ETH_AUTH_NONE = 0
ETH_AUTH_8021X = 1
ETH_AUTH_MAB = 2


# ═══════════════════════════════════════════════════════════════
# IpConfiguration
# ═══════════════════════════════════════════════════════════════

class IpConfiguration(Stub):
    """IpConfiguration — IP 配置"""

    class IpAssignment(Stub):
        """IpConfiguration.IpAssignment — IP 分配方式"""
        DHCP = 0
        STATIC = 1
        UNASSIGNED = 2

    class ProxySettings(Stub):
        """IpConfiguration.ProxySettings — 代理设置"""
        NONE = 0
        STATIC = 1
        PAC = 2
        UNASSIGNED = 3

    def __init__(self, ipAssignment: int = IpAssignment.DHCP,
                 proxySettings: int = ProxySettings.NONE,
                 staticConfig: Optional["StaticIpConfiguration"] = None):
        super().__init__()
        self._ip_assignment = ipAssignment
        self._proxy_settings = proxySettings
        self._static_config = staticConfig

    def getIpAssignment(self) -> int:
        return self._ip_assignment

    def setIpAssignment(self, assignment: int):
        self._ip_assignment = assignment

    def getProxySettings(self) -> int:
        return self._proxy_settings

    def setProxySettings(self, settings: int):
        self._proxy_settings = settings

    def getStaticIpConfiguration(self) -> Optional["StaticIpConfiguration"]:
        return self._static_config

    def setStaticIpConfiguration(self, config: "StaticIpConfiguration"):
        self._static_config = config

    def toString(self) -> str:
        assignments = {0: "DHCP", 1: "STATIC", 2: "UNASSIGNED"}
        return f"IpConfiguration(assignment={assignments.get(self._ip_assignment, str(self._ip_assignment))})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, IpConfiguration):
            return False
        return self._ip_assignment == other._ip_assignment

    def hashCode(self) -> int:
        return self._ip_assignment & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# StaticIpConfiguration
# ═══════════════════════════════════════════════════════════════

class StaticIpConfiguration(Stub):
    """StaticIpConfiguration — 静态 IP 配置"""

    def __init__(self, ipAddress: str = "192.168.1.100",
                 gateway: str = "192.168.1.1",
                 prefixLength: int = 24,
                 dnsServers: List[str] = None,
                 domains: str = ""):
        super().__init__()
        self._ip_address = ipAddress
        self._gateway = gateway
        self._prefix_length = prefixLength
        self._dns_servers = dnsServers or ["8.8.8.8", "8.8.4.4"]
        self._domains = domains

    def getIpAddress(self) -> str:
        return self._ip_address

    def setIpAddress(self, ip: str):
        self._ip_address = ip

    def getGateway(self) -> str:
        return self._gateway

    def setGateway(self, gateway: str):
        self._gateway = gateway

    def getPrefixLength(self) -> int:
        return self._prefix_length

    def setPrefixLength(self, prefixLength: int):
        self._prefix_length = prefixLength

    def getDnsServers(self) -> List[str]:
        return list(self._dns_servers)

    def setDnsServers(self, dnsServers: List[str]):
        self._dns_servers = list(dnsServers)

    def addDnsServer(self, server: str):
        self._dns_servers.append(server)

    def getDomains(self) -> str:
        return self._domains

    def setDomains(self, domains: str):
        self._domains = domains

    class Builder(Stub):
        """StaticIpConfiguration.Builder — 静态 IP 构建器"""

        def __init__(self):
            super().__init__()
            self._ip_address = ""
            self._gateway = ""
            self._prefix_length = 24
            self._dns_servers: List[str] = []
            self._domains = ""

        def setIpAddress(self, ip: str) -> "StaticIpConfiguration.Builder":
            self._ip_address = ip
            return self

        def setGateway(self, gateway: str) -> "StaticIpConfiguration.Builder":
            self._gateway = gateway
            return self

        def setPrefixLength(self, prefixLength: int) -> "StaticIpConfiguration.Builder":
            self._prefix_length = prefixLength
            return self

        def setDnsServers(self, dnsServers: List[str]) -> "StaticIpConfiguration.Builder":
            self._dns_servers = list(dnsServers)
            return self

        def setDomains(self, domains: str) -> "StaticIpConfiguration.Builder":
            self._domains = domains
            return self

        def build(self) -> "StaticIpConfiguration":
            return StaticIpConfiguration(
                ipAddress=self._ip_address, gateway=self._gateway,
                prefixLength=self._prefix_length,
                dnsServers=self._dns_servers, domains=self._domains,
            )

    def toString(self) -> str:
        return f"StaticIpConfiguration(ip={self._ip_address}, gateway={self._gateway})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, StaticIpConfiguration):
            return False
        return self._ip_address == other._ip_address

    def hashCode(self) -> int:
        return hash(self._ip_address) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# EthernetNetworkSpecifier
# ═══════════════════════════════════════════════════════════════

class EthernetNetworkSpecifier(Stub):
    """EthernetNetworkSpecifier — 以太网网络指定器"""

    def __init__(self, interfaceName: str = ""):
        super().__init__()
        self._interface_name = interfaceName

    def getInterfaceName(self) -> str:
        return self._interface_name

    def satisfiedBy(self, other: Optional["EthernetNetworkSpecifier"]) -> bool:
        if other is None:
            return False
        return self._interface_name == other._interface_name

    def toString(self) -> str:
        return f"EthernetNetworkSpecifier(iface={self._interface_name})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, EthernetNetworkSpecifier):
            return False
        return self._interface_name == other._interface_name

    def hashCode(self) -> int:
        return hash(self._interface_name) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# EthernetNetworkUpdate
# ═══════════════════════════════════════════════════════════════

class EthernetNetworkUpdate(Stub):
    """EthernetNetworkUpdate — 以太网网络更新"""

    def __init__(self, interfaceName: str = "",
                 state: int = ETH_STATE_DISCONNECTED,
                 config: Optional["EthernetConfig"] = None):
        super().__init__()
        self._interface_name = interfaceName
        self._state = state
        self._config = config

    def getInterfaceName(self) -> str:
        return self._interface_name

    def getState(self) -> int:
        return self._state

    def getConfig(self) -> Optional["EthernetConfig"]:
        return self._config

    def toString(self) -> str:
        return f"EthernetNetworkUpdate(iface={self._interface_name}, state={self._state})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, EthernetNetworkUpdate):
            return False
        return self._interface_name == other._interface_name

    def hashCode(self) -> int:
        return hash(self._interface_name) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# EthernetConfig
# ═══════════════════════════════════════════════════════════════

class EthernetConfig(Stub):
    """EthernetConfig — 以太网配置"""

    def __init__(self, ipConfig: Optional[IpConfiguration] = None,
                 linkMode: int = ETH_LINK_MODE_AUTO,
                 interfaceName: str = "",
                 mtu: int = 1500,
                 macAddress: str = ""):
        super().__init__()
        self._ip_config = ipConfig or IpConfiguration()
        self._link_mode = linkMode
        self._interface_name = interfaceName
        self._mtu = mtu
        self._mac_address = macAddress

    def getIpConfiguration(self) -> IpConfiguration:
        return self._ip_config

    def setIpConfiguration(self, config: IpConfiguration):
        self._ip_config = config

    def getLinkMode(self) -> int:
        return self._link_mode

    def setLinkMode(self, mode: int):
        self._link_mode = mode

    def getInterfaceName(self) -> str:
        return self._interface_name

    def setInterfaceName(self, name: str):
        self._interface_name = name

    def getMtu(self) -> int:
        return self._mtu

    def setMtu(self, mtu: int):
        self._mtu = mtu

    def getMacAddress(self) -> str:
        return self._mac_address

    def setMacAddress(self, mac: str):
        self._mac_address = mac

    class Builder(Stub):
        """EthernetConfig.Builder — 以太网配置构建器"""

        def __init__(self):
            super().__init__()
            self._ip_config = IpConfiguration()
            self._link_mode = ETH_LINK_MODE_AUTO
            self._interface_name = ""
            self._mtu = 1500
            self._mac_address = ""

        def setIpConfiguration(self, config: IpConfiguration) -> "EthernetConfig.Builder":
            self._ip_config = config
            return self

        def setLinkMode(self, mode: int) -> "EthernetConfig.Builder":
            self._link_mode = mode
            return self

        def setInterfaceName(self, name: str) -> "EthernetConfig.Builder":
            self._interface_name = name
            return self

        def setMtu(self, mtu: int) -> "EthernetConfig.Builder":
            self._mtu = mtu
            return self

        def setMacAddress(self, mac: str) -> "EthernetConfig.Builder":
            self._mac_address = mac
            return self

        def build(self) -> "EthernetConfig":
            return EthernetConfig(
                ipConfig=self._ip_config, linkMode=self._link_mode,
                interfaceName=self._interface_name, mtu=self._mtu,
                macAddress=self._mac_address,
            )

    def toString(self) -> str:
        return f"EthernetConfig(iface={self._interface_name}, linkMode={self._link_mode})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, EthernetConfig):
            return False
        return self._interface_name == other._interface_name

    def hashCode(self) -> int:
        return hash(self._interface_name) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# EthernetManager
# ═══════════════════════════════════════════════════════════════

class EthernetManager(Stub):
    """EthernetManager — 以太网管理器"""

    ACTION_ETHERNET_STATE_CHANGED = "android.net.ethernet.ETHERNET_STATE_CHANGED"
    EXTRA_ETHERNET_STATE = "ethernet_state"
    EXTRA_INTERFACE_NAME = "interface_name"
    EXTRA_LINK_MODE = "link_mode"

    STATE_DISABLED = 0
    STATE_ENABLED = 1

    def __init__(self):
        super().__init__()
        self._enabled = False
        self._interfaces: Dict[str, EthernetConfig] = {}
        self._state = EthernetManager.STATE_DISABLED
        self._listeners: Dict[int, Callable] = {}

    def isEnabled(self) -> bool:
        return self._enabled

    def setEnabled(self, enabled: bool):
        self._enabled = enabled
        self._state = EthernetManager.STATE_ENABLED if enabled else EthernetManager.STATE_DISABLED

    def getState(self) -> int:
        return self._state

    def getConfiguration(self, interfaceName: str) -> Optional["EthernetConfig"]:
        return self._interfaces.get(interfaceName)

    def setConfiguration(self, interfaceName: str, config: "EthernetConfig"):
        self._interfaces[interfaceName] = config

    def removeInterface(self, interfaceName: str):
        self._interfaces.pop(interfaceName, None)

    def getInterfaceNames(self) -> List[str]:
        return list(self._interfaces.keys())

    def addInterface(self, interfaceName: str, config: Optional[EthernetConfig] = None):
        self._interfaces[interfaceName] = config or EthernetConfig(interfaceName=interfaceName)

    def setLinkMode(self, interfaceName: str, mode: int):
        config = self._interfaces.get(interfaceName)
        if config:
            config.setLinkMode(mode)

    def updateConfiguration(self, update: EthernetNetworkUpdate) -> bool:
        iface = update.getInterfaceName()
        config = update.getConfig()
        if config:
            self._interfaces[iface] = config
        elif iface in self._interfaces:
            pass
        return True

    def registerEthernetEventListener(self, listener: Callable):
        lid = id(listener)
        self._listeners[lid] = listener

    def unregisterEthernetEventListener(self, listener: Callable):
        lid = id(listener)
        self._listeners.pop(lid, None)

    def getAvailableInterfaces(self) -> List[str]:
        import socket
        import netifaces
        interfaces = []
        try:
            import netifaces
            for iface in netifaces.interfaces():
                if iface.startswith(("eth", "en", "em")):
                    interfaces.append(iface)
        except ImportError:
            import psutil
            try:
                import psutil
                stats = psutil.net_if_stats()
                for name in stats:
                    if name.startswith(("eth", "en", "em")):
                        interfaces.append(name)
            except ImportError:
                pass
        return interfaces + list(self._interfaces.keys())

    def isInterfaceAvailable(self, interfaceName: str) -> bool:
        return interfaceName in self._interfaces

    def toString(self) -> str:
        return f"EthernetManager(enabled={self._enabled}, interfaces={len(self._interfaces)})"

    def equals(self, other: Any) -> bool:
        return self is other

    def hashCode(self) -> int:
        return id(self) & 0x7FFFFFFF

    def getInterfaceCount(self) -> int:
        return len(self._interfaces)

    def hasInterface(self, interfaceName: str) -> bool:
        return interfaceName in self._interfaces

    def getInterfaceNames(self) -> List[str]:
        return list(self._interfaces.keys())

    def getEnabledInterfaces(self) -> List[str]:
        return [name for name, cfg in self._interfaces.items()
                if cfg.getLinkMode() != ETH_STATE_DISCONNECTED]

    def getInterfaceConfig(self, interfaceName: str) -> Optional[EthernetConfig]:
        return self._interfaces.get(interfaceName)

    def setInterfaceUp(self, interfaceName: str) -> bool:
        config = self._interfaces.get(interfaceName)
        if config:
            config.setLinkMode(ETH_LINK_MODE_AUTO)
            return True
        return False

    def setInterfaceDown(self, interfaceName: str) -> bool:
        config = self._interfaces.get(interfaceName)
        if config:
            config.setLinkMode(ETH_STATE_DISCONNECTED)
            return True
        return False

    def getLinkMode(self, interfaceName: str) -> int:
        config = self._interfaces.get(interfaceName)
        return config.getLinkMode() if config else ETH_STATE_DISCONNECTED

    def getMtu(self, interfaceName: str) -> int:
        config = self._interfaces.get(interfaceName)
        return config.getMtu() if config else 0

    def setMtu(self, interfaceName: str, mtu: int) -> bool:
        config = self._interfaces.get(interfaceName)
        if config:
            config.setMtu(mtu)
            return True
        return False

    def getMacAddress(self, interfaceName: str) -> str:
        config = self._interfaces.get(interfaceName)
        return config.getMacAddress() if config else ""

    def setMacAddress(self, interfaceName: str, mac: str) -> bool:
        config = self._interfaces.get(interfaceName)
        if config:
            config.setMacAddress(mac)
            return True
        return False