# Copyright (c) 2026 LightSeek Foundation # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """Port, socket, and host networking helpers.""" from __future__ import annotations import logging import os import socket import warnings logger = logging.getLogger(__name__) def is_port_available(port): """Return whether a port is available.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: try: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(("", port)) s.listen(1) return True except OSError: return False except OverflowError: return False def get_free_port(): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("", 0)) return s.getsockname()[1] except OSError: with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: s.bind(("", 0)) return s.getsockname()[1] def set_uvicorn_logging_configs(): from uvicorn.config import LOGGING_CONFIG LOGGING_CONFIG["formatters"]["default"][ "fmt" ] = "[%(asctime)s] %(levelprefix)s %(message)s" LOGGING_CONFIG["formatters"]["default"]["datefmt"] = "%Y-%m-%d %H:%M:%S" LOGGING_CONFIG["formatters"]["access"][ "fmt" ] = '[%(asctime)s] %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s' LOGGING_CONFIG["formatters"]["access"]["datefmt"] = "%Y-%m-%d %H:%M:%S" def get_local_ip_by_remote() -> str | None: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except Exception: pass try: hostname = socket.gethostname() ip = socket.gethostbyname(hostname) if ip and ip != "127.0.0.1" and ip != "0.0.0.0": return ip except Exception: pass try: s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) s.connect(("2001:4860:4860::8888", 80)) return s.getsockname()[0] except Exception: raise ValueError("Can not get local ip") def get_ip() -> str: from tokenspeed.runtime.utils.env import envs host_ip = envs.TOKENSPEED_HOST_IP.get() or os.getenv("HOST_IP", "") if host_ip: return host_ip try: hostname = socket.gethostname() ip = socket.gethostbyname(hostname) if ip and ip != "127.0.0.1" and ip != "0.0.0.0": return ip except Exception: pass s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except Exception: pass try: s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) s.connect(("2001:4860:4860::8888", 80)) return s.getsockname()[0] except Exception: pass warnings.warn( "Failed to get the IP address, using 0.0.0.0 by default." "The value can be set by the environment variable" " TOKENSPEED_HOST_IP or HOST_IP.", stacklevel=2, ) return "0.0.0.0"