# Copyright (c) ETH Zurich, SIS ID and HVL D-ITET
#
""" """
import logging
import socket
from collections.abc import Sequence
from ipaddress import IPv4Address, IPv6Address, ip_address
from logging import Logger
import numpy as np
from hvl_ccb.utils.typing import Number
[docs]
def validate_number(
x_name: str,
x: object,
limits: tuple[Number | None, Number | None] | list[Number | None] | None = (
None,
None,
),
number_type: type[Number] | tuple[type[Number], ...] = (int, float),
*,
closed_interval: bool | tuple[bool, bool] | list[bool] = True,
logger: Logger | None = None,
) -> None:
"""
Validate if given input `x` is a number of given `number_type` type, with value
between given `limits[0]` and `limits[1]`, if not `None`.
For array-like objects (npt.NDArray, list, tuple, dict) it is checked if all
elements are within the limits and have the correct type.
:param x_name: string name of the validate input, use for the error message
:param x: an input object to validate as number of given type within given range
:param limits: two-item list/tuple with lower and upper limit. Use `None`
for an unbounded side. `-np.inf` is accepted as unbounded lower limit,
and `np.inf` is accepted as unbounded upper limit.
:param number_type: expected type or tuple of types of a number,
by default `(int, float)`
:param closed_interval: if `True`, both limits are inclusive; if `False`, both
are exclusive; alternatively, provide `(lower_closed, upper_closed)` to
define half-open intervals
:param logger: logger of the calling submodule
:raises TypeError: when the validated input does not have expected type
:raises ValueError: when the validated input has correct number type but is not
within given range or has wrong input limits
"""
if logger is None:
logger = logging.getLogger(__name__)
if limits is None:
limits = (None, None)
elif (
not isinstance(limits, list | tuple)
or len(limits) != 2
or not all(
limit is None
or (not isinstance(limit, bool) and isinstance(limit, int | float | Number))
for limit in limits
)
):
msg = (
"limits must be None or a list/tuple of two numbers or None values "
"(lower_limit, upper_limit)."
)
logger.error(msg)
raise TypeError(msg)
if (limits[0] is not None and np.isposinf(limits[0])) or (
limits[1] is not None and np.isneginf(limits[1])
):
msg = (
"limits can only use -inf as an unbounded lower limit and +inf "
"as an unbounded upper limit."
)
logger.error(msg)
raise ValueError(msg)
if limits[0] is None or np.isneginf(limits[0]):
limits = (None, limits[1])
if limits[1] is None or np.isposinf(limits[1]):
limits = (limits[0], None)
if limits[0] is not None and limits[1] is not None and limits[0] > limits[1]:
msg = (
f"Upper limit {limits[1]} should be greater than "
f"the lower limit {limits[0]}"
)
logger.error(msg)
raise ValueError(msg)
if isinstance(closed_interval, bool):
closed_interval = (closed_interval, closed_interval)
elif len(closed_interval) != 2 or not all(
isinstance(is_closed, bool) for is_closed in closed_interval
):
msg = (
"closed_interval must be a bool or a tuple of two bools "
"(lower_closed, upper_closed)."
)
logger.error(msg)
raise TypeError(msg)
data_type = type(x)
if not isinstance(number_type, Sequence):
number_type = (number_type,)
if isinstance(x, float | int):
if not isinstance(x, number_type):
msg = (
f"{x_name} = {x} has to be of type "
f"{' or '.join(nt.__name__ for nt in number_type)}"
)
logger.error(msg)
raise TypeError(msg)
elif isinstance(x, list | tuple | dict | np.ndarray):
if isinstance(x, dict):
x = np.asarray(list(x.values()))
x = np.asarray(x)
if x.dtype not in number_type:
msg = (
f"{x_name} = {x} needs to include only numbers type "
f"{' or '.join(nt.__name__ for nt in number_type)}"
)
logger.error(msg)
raise TypeError(msg)
else:
msg = (
f"{x_name} = {x} must be an Integer, a Float, a Tuple, a List, "
f"a Dictionary or a Numpy array, but the received type is {data_type}."
)
logger.error(msg)
raise TypeError(msg)
lower_in_bounds = limits[0] is None or np.all(
np.greater_equal(x, limits[0])
if closed_interval[0]
else np.greater(x, limits[0])
)
upper_in_bounds = limits[1] is None or np.all(
np.less_equal(x, limits[1]) if closed_interval[1] else np.less(x, limits[1])
)
if not lower_in_bounds or not upper_in_bounds:
if limits[0] is None:
suffix = (
f"less than {limits[1]}"
if not closed_interval[1]
else f"less or equal than {limits[1]}"
)
elif limits[1] is None:
suffix = (
f"greater than {limits[0]}"
if not closed_interval[0]
else f"greater or equal than {limits[0]}"
)
else:
left_bracket = "[" if closed_interval[0] else "("
right_bracket = "]" if closed_interval[1] else ")"
suffix = (
f"in interval {left_bracket}{limits[0]}, {limits[1]}{right_bracket}"
)
msg = f"{x_name} = {x} must be {suffix}"
logger.error(msg)
raise ValueError(msg)
[docs]
def validate_bool(x_name: str, x: object, logger: Logger | None = None) -> None:
"""
Validate if given input `x` is a `bool`.
:param x_name: string name of the validate input, use for the error message
:param x: an input object to validate as boolean
:param logger: logger of the calling submodule
:raises TypeError: when the validated input does not have boolean type
"""
if logger is None:
logger = logging.getLogger(__name__)
if not isinstance(x, bool):
msg = f"{x_name} = {x} has to of type bool"
logger.error(msg)
raise TypeError(msg)
[docs]
def validate_and_resolve_host(
host: str | IPv4Address | IPv6Address | None, logger: Logger | None = None
) -> str:
if logger is None:
logger = logging.getLogger(__name__)
if host is None:
msg = "A host has to be provided."
logger.error(msg)
raise AttributeError(msg)
if isinstance(host, IPv4Address | IPv6Address):
host = str(host)
else:
try:
host = str(ip_address(host))
except ValueError:
try:
host = socket.gethostbyname(host)
except (socket.gaierror, UnicodeError) as exc:
# UnicodeError: "192..168.01"
# socket.gaierror: "192.168.0.1000"
# socket.gaierror: "itet-hvl-01"
msg = (
"The value of host is neither an IPv4 nor an IPv6 address nor a "
f"hostname that can be resolved. The received value is: '{host}'."
)
logger.exception(msg, exc_info=exc)
raise ValueError(msg) from exc
return str(host)
[docs]
def validate_tcp_port(port: int | None, logger: Logger | None = None) -> None:
if logger is None:
logger = logging.getLogger(__name__)
if port is None:
msg = "A port has to be provided."
logger.error(msg)
raise AttributeError(msg)
# TCP ports are in the range between and including 0 and (2**16)-1, but 0 is
# reserved and shall not be used
validate_number("port", port, (1, 2**16 - 1), int, logger=logger)