Source code for pingparsing._pingtransmitter

.. codeauthor:: Tsuyoshi Hombashi <>

import ipaddress
import platform
from collections import namedtuple
from typing import Optional, cast

import humanreadable as hr
import subprocrunner
import typepy
from subprocrunner.typing import Command
from typepy import Integer, StrictLevel, String, TypeConversionError

from ._cmd_maker import LinuxPingCmdMaker, MacosPingCmdMaker, WindowsPingCmdMaker
from ._logger import logger
from ._typing import PingAddOpts, TimeArg


[docs]class PingResult(namedtuple("PingResult", "stdout stderr returncode")): """ Data class to store ``ping`` command execution result. .. py:attribute:: stdout :type: Optional[str] Standard output of ``ping`` command execution result. .. py:attribute:: stderr :type: Optional[str] Standard error of ``ping`` command execution result. .. py:attribute:: returncode :type: int Return code of ``ping`` command execution result. """
[docs]class PingTransmitter: """ Transmitter class to send ICMP packets by using the OS built-in ``ping`` command. .. py:attribute:: count :type: Optional[int] :value: None Number of sending ICMP packets. This attribute ignored if the value is |None|. .. py:attribute:: packet_size :type: Optional[int] :value: None Specifies the number of data bytes to be sent. .. py:attribute:: ttl :type: Optional[int] :value: None Specifies the Time to Live. .. py:attribute:: ping_option :type: Union[str, Sequence[str]] :value: "" Additional ``ping`` command option. .. py:attribute:: interface :type: Optional[str] :value: None Interface name or zone-id. The attribute required when :py:attr:`~.destination` is IPv6 link-local scope address. .. py:attribute:: timestamp :type: bool :value: False [Only for Linux environment] If |True|, add timestamp for each ping result. Defaults to ``False``. .. py:attribute:: auto_codepage :type: bool :value: True [Only for Windows environment] Automatically change code page if ``True``. Defaults to ``True``. """ @property def destination(self) -> str: """ Hostname or IP-address (IPv4/IPv6) to sending ICMP packets. """ return self.__destination @destination.setter def destination(self, value: str) -> None: if not String(value, strict_level=StrictLevel.MAX).is_type(): raise ValueError("empty destination") self.__destination = value @property def destination_host(self) -> str: """ Alias to :py:attr:`~.destination` """ return self.destination @destination_host.setter def destination_host(self, value: str) -> None: self.destination = value @property def timeout(self) -> Optional[hr.Time]: """ Time to wait for a response per packet. You can specify either a number or a string (e.g. ``"1sec"``). If only a number is specified and a unit not found, the unit will be considered as seconds. +------------+----------------------------------------------------------+ | Unit | Available specifiers (str) | +============+==========================================================+ |days |``d``/``day``/``days`` | +------------+----------------------------------------------------------+ |hours |``h``/``hour``/``hours`` | +------------+----------------------------------------------------------+ |minutes |``m``/``min``/``mins``/``minute``/``minutes`` | +------------+----------------------------------------------------------+ |seconds |``s``/``sec``/``secs``/``second``/``seconds`` | +------------+----------------------------------------------------------+ |milliseconds|``ms``/``msec``/``msecs``/``millisecond``/``milliseconds``| +------------+----------------------------------------------------------+ |microseconds|``us``/``usec``/``usecs``/``microsecond``/``microseconds``| +------------+----------------------------------------------------------+ Use system default timeout if the value is |None|. Defaults to |None|. If the system does not support timeout in milliseconds, round up as seconds. Returns: humanreadable.Time: timeout """ return self.__timeout @timeout.setter def timeout(self, value: TimeArg) -> None: if value is None: self.__timeout: Optional[hr.Time] = value return if isinstance(value, hr.Time): timeout = cast(hr.Time, value) else: timeout = hr.Time(str(value), default_unit=hr.Time.Unit.MILLISECOND) if timeout.milliseconds <= 0: raise ValueError("timeout must be greater than zero") self.__timeout = cast(hr.Time, timeout) @property def deadline(self) -> Optional[hr.Time]: """ Timeout before ping exits. You can specify either a number or a string (e.g. ``"1sec"``). If only a number is specified and a unit not found, the unit will be considered as seconds. +------------+----------------------------------------------------------+ | Unit | Available specifiers (str) | +============+==========================================================+ |days |``d``/``day``/``days`` | +------------+----------------------------------------------------------+ |hours |``h``/``hour``/``hours`` | +------------+----------------------------------------------------------+ |minutes |``m``/``min``/``mins``/``minute``/``minutes`` | +------------+----------------------------------------------------------+ |seconds |``s``/``sec``/``secs``/``second``/``seconds`` | +------------+----------------------------------------------------------+ |milliseconds|``ms``/``msec``/``msecs``/``millisecond``/``milliseconds``| +------------+----------------------------------------------------------+ |microseconds|``us``/``usec``/``usecs``/``microsecond``/``microseconds``| +------------+----------------------------------------------------------+ If both :py:attr:`~.deadline` and :py:attr:`~.count` are |None|, :py:attr:`~.deadline` is automatically set to the default value (``3 seconds``). Defaults to |None|. Returns: humanreadable.Time: deadline """ return self.__deadline @deadline.setter def deadline(self, value: TimeArg) -> None: if value is None: self.__deadline = value return if isinstance(value, hr.Time): deadline = cast(hr.Time, value) else: deadline = hr.Time(str(value), default_unit=hr.Time.Unit.SECOND) if deadline.milliseconds <= 0: raise ValueError("deadline must be greater than zero") self.__deadline = deadline def __init__(self) -> None: self.__destination = "" self.count: Optional[int] = None self.packet_size: Optional[int] = None self.ttl: Optional[int] = None self.ping_option: PingAddOpts = [] self.is_quiet = False self.interface: Optional[str] = None self.auto_codepage = True self.timeout = None self.deadline = None self.timestamp = False
[docs] def ping(self) -> PingResult: """ Sending ICMP packets. :return: ``ping`` command execution result. :rtype: :py:class:`.PingResult` :raises ValueError: If parameters not valid. """ self.__validate_ping_param() ping_runner = subprocrunner.SubprocessRunner(self.__make_ping_command()) return PingResult(ping_runner.stdout, ping_runner.stderr, ping_runner.returncode)
@staticmethod def __is_linux(): return platform.system() == "Linux" @staticmethod def __is_macos(): return platform.system() == "Darwin" @staticmethod def __is_windows(): return platform.system() == "Windows" def __is_ipv6(self): try: network = ipaddress.ip_address(str(self.destination)) except ValueError as e: logger.debug(e) return False logger.debug(f"IP address: version={network.version}, address={self.destination}") return network.version == 6 def __validate_ping_param(self) -> None: self.__validate_count() self.__validate_interface() def __validate_count(self) -> None: if self.count is None: return try: count = Integer(self.count).convert() except TypeConversionError as e: raise ValueError(f"count must be an integer: {e}") if count <= 0: raise ValueError("count must be greater than zero") def __validate_interface(self) -> None: if not self.__is_ipv6(): return if not ipaddress.ip_network(str(self.destination)).is_link_local: return if typepy.is_null_string(self.interface): raise ValueError("interface required to ping to IPv6 link local address") def __make_ping_command(self) -> Command: from typing import Any # noqa maker_class: Any = None if self.__is_linux(): maker_class = LinuxPingCmdMaker elif self.__is_macos(): maker_class = MacosPingCmdMaker elif self.__is_windows(): maker_class = WindowsPingCmdMaker else: raise RuntimeError(f"not supported platform: {platform.system()}") return maker_class( count=self.count, packet_size=self.packet_size, ttl=self.ttl, deadline=self.deadline, timeout=self.timeout, interface=self.interface, is_ipv6=self.__is_ipv6(), timestamp=self.timestamp, auto_codepage=self.auto_codepage, ping_option=self.ping_option, ).make_cmd(destination=self.destination)