implement ping and speed test services

debian/changelog

@@ -1,4 +1,4 @@
-wlanpi-core (1.0.5-5) UNRELEASED; urgency=medium
+wlanpi-core (1.0.5-6) UNRELEASED; urgency=medium
 
   [ Michael Ketchel ]
   * Draft build of new wlan control features
@@ -8,8 +8,9 @@ wlanpi-core (1.0.5-5) UNRELEASED; urgency=medium
   * Version bump for build
   * Add automatic default gateway configuration for wlan api
   * Fix some auto gateway config quirks
+  * implement ping and speed test services
 
- -- Michael Ketchel <[email protected]>  Wed, 13 Nov 2024 21:22:34 +0000
+ -- Michael Ketchel <[email protected]>  Thu, 14 Nov 2024 05:09:35 +0000
 
 wlanpi-core (1.0.5-1) unstable; urgency=high
 

wlanpi_core/api/api_v1/endpoints/utils_api.py

@@ -17,7 +17,7 @@
     response_model=utils.ReachabilityTest,
     response_model_exclude_none=True,
 )
-async def reachability():
+async def check_reachability():
     """
     Runs the reachability test and returns the results
     """
@@ -97,7 +97,7 @@ async def usb_interfaces():
 
 
 @router.get("/ufw", response_model=utils.Ufw)
-async def usb_interfaces():
+async def ufw_information():
     """
     Returns the UFW information.
     """
@@ -119,3 +119,41 @@ async def usb_interfaces():
     except Exception as ex:
         log.error(ex)
         return Response(content=f"Internal Server Error", status_code=500)
+
+
+@router.post("/ping", response_model=utils.PingResult)
+async def execute_ping(request: utils.PingRequest):
+    """
+    Pings a target and returns the results
+    """
+
+    try:
+        result = await utils_service.ping(
+            request.destination,
+            request.count,
+            request.interval,
+            request.ttl,
+            request.interface,
+        )
+        return result
+
+    except ValidationError as ve:
+        return Response(content=ve.error_msg, status_code=ve.status_code)
+    except Exception as ex:
+        log.error(ex)
+        return Response(content=f"Internal Server Error: {ex}", status_code=500)
+
+
+@router.post("/iperf2/client", response_model=utils.Iperf2Result)
+async def execute_iperf(request: utils.IperfRequest):
+    """
+    Runs iperf against a target and returns the results
+    """
+
+    try:
+        return await utils_service.run_iperf2_client(**request.__dict__)
+    except ValidationError as ve:
+        return Response(content=ve.error_msg, status_code=ve.status_code)
+    except Exception as ex:
+        log.error(ex)
+        return Response(content=f"Internal Server Error: {ex}", status_code=500)

wlanpi_core/schemas/utils/__init__.py

@@ -1 +1,11 @@
-from .utils import ReachabilityTest, SpeedTest, Ufw, Usb
+from .utils import (
+    Iperf2Result,
+    IperfRequest,
+    PingRequest,
+    PingResponse,
+    PingResult,
+    ReachabilityTest,
+    SpeedTest,
+    Ufw,
+    Usb,
+)

wlanpi_core/schemas/utils/utils.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Extra, Field
 
 
 class ReachabilityTest(BaseModel):
@@ -40,3 +40,96 @@ class Usb(BaseModel):
 class Ufw(BaseModel):
     status: str = Field()
     ports: list = Field()
+
+
+class PingRequest(BaseModel):
+    destination: str = Field(examples=["google.com", "192.168.1.1"])
+    count: int = Field(
+        examples=[1, 10], description="How many packets to send.", default=1
+    )
+    interval: float = Field(
+        examples=[1], description="The interval between packets, in seconds", default=1
+    )
+    ttl: Optional[int] = Field(
+        examples=[20], description="The Time-to-Live of the ping attempt.", default=None
+    )
+    interface: Optional[str] = Field(
+        examples=["eth0"],
+        description="The interface the ping should originate from",
+        default=None,
+    )
+
+
+class PingResponse(BaseModel):
+    type: str = Field(examples=["reply"])
+    timestamp: float = Field(examples=[1731371899.060181])
+    bytes: int = Field(examples=[64])
+    response_ip: str = Field(examples=["142.250.190.142"])
+    icmp_seq: int = Field(examples=[1])
+    ttl: int = Field(examples=[55])
+    time_ms: float = Field(examples=[26.6])
+    duplicate: bool = Field(examples=[False])
+
+
+class PingResult(
+    BaseModel,
+):
+    destination_ip: str = Field(examples=["142.250.190.142"])
+    interface: Optional[str] = Field(
+        examples=["eth0"],
+        default=None,
+        description="The interface the user specified that the ping be issued from. It will be empty if there wasn't one specified.",
+    )
+    data_bytes: Optional[int] = Field(examples=[56], default=None)
+    pattern: Optional[str] = Field()
+    destination: str = Field(examples=["google.com"])
+    packets_transmitted: int = Field(examples=[10])
+    packets_received: int = Field(examples=[10])
+    packet_loss_percent: float = Field(examples=[0.0])
+    duplicates: int = Field(examples=[0])
+    time_ms: float = Field(examples=[9012.0])
+    round_trip_ms_min: float = Field(examples=[24.108])
+    round_trip_ms_avg: float = Field(examples=[29.318])
+    round_trip_ms_max: float = Field(examples=[37.001])
+    round_trip_ms_stddev: float = Field(examples=[4.496])
+    jitter: Optional[float] = Field(examples=[37.001], default=None)
+    responses: list[PingResponse] = Field()
+
+
+class IperfRequest(BaseModel):
+    host: str = Field(examples=["192.168.1.1"])
+    port: int = Field(examples=[5001], default=5001)
+    time: int = Field(examples=[10], default=10)
+    udp: bool = Field(default=False)
+    reverse: bool = Field(default=False)
+    compatibility: bool = Field(default=False)
+    interface: Optional[str] = Field(examples=["wlan0"], default=None)
+    # version: int = Field(examples=[2, 3], default=3)
+    # interface: Optional[str] = Field(examples=["eth0, wlan0"], default=None)
+    # bind_address: Optional[str] = Field(examples=["192.168.1.12"], default=None)
+    #
+    # @model_validator(mode="after")
+    # def check_dynamic_condition(self) -> Self:
+    #     # print(self)
+    #     if self.version not in [2, 3]:
+    #         raise ValueError("iPerf version can be 2 or 3.")
+    #     if self.bind_address is not None and self.interface is not None:
+    #         raise ValueError("Only interface or bind_address can be specified.")
+    #     return self
+
+
+class Iperf2Result(BaseModel, extra=Extra.allow):
+    timestamp: int = Field()
+    source_address: str = Field(examples=["192.168.1.5"])
+    source_port: int = Field(examples=[5001])
+    destination_address: str = Field(examples=["192.168.1.1"])
+    destination_port: int = Field(examples=[12345])
+    transfer_id: int = Field(examples=[3])
+    interval: list[float] = Field(examples=[0.0, 10.0])
+    transferred_bytes: int = Field()
+    transferred_mbytes: float = Field()
+    bps: int = Field()
+    mbps: float = Field()
+    jitter: Optional[float] = Field(default=None)
+    error_count: Optional[int] = Field(default=None)
+    datagrams: Optional[int] = Field(default=None)

wlanpi_core/services/utils_service.py

@@ -1,11 +1,13 @@
 import os
 import re
+from typing import Optional
 
 from wlanpi_core.constants import UFW_FILE
 
 from ..models.runcommand_error import RunCommandError
+from ..schemas.utils import PingResult
 from ..utils.general import run_command_async
-from ..utils.network import get_default_gateways
+from ..utils.network import get_default_gateways, get_ip_address
 
 
 async def show_reachability():
@@ -200,3 +202,110 @@ async def show_ufw():
     response = ufw_info
 
     return response
+
+
+async def ping(
+    target: str,
+    count: int = 1,
+    interval: float = 1,
+    ttl: Optional[int] = None,
+    interface: Optional[str] = None,
+) -> PingResult:
+    def calculate_jitter(values: list[float], precision: int = 3) -> float:
+        return round(
+            sum([abs(values[i + 1] - values[i]) for i in range(len(values) - 1)])
+            / len(values),
+            precision,
+        )
+
+    command: list[str] = "jc ping -D".split()
+    command.extend(["-c", str(count), "-i", str(interval)])
+    if ttl is not None:
+        command.extend(["-t", str(ttl)])
+    if interface is not None:
+        command.extend(["-I", str(interface)])
+    command.append(target)
+    res = await run_command_async(command)
+    result: dict = res.output_from_json()  # type: ignore
+    # Calculate jitter if we can
+    result["jitter"] = (
+        calculate_jitter([x["time_ms"] for x in result["responses"]])
+        if len(result["responses"]) > 1
+        else None
+    )
+    result["interface"] = interface
+    return PingResult(**result)
+
+
+async def run_iperf2_client(
+    host: str,
+    port: int = 5001,
+    time: int = 10,
+    reverse: bool = False,
+    bind: Optional[str] = None,
+    interface: Optional[str] = None,
+    udp=False,
+    compatibility=False,
+):
+    command: list[str] = "iperf -y C ".split()
+    command.extend(["-t", str(time), "-c", host, "-p", str(port)])
+
+    if reverse:
+        command.append("-R")
+
+    if compatibility:
+        command.append("-C")
+
+    if bind:
+        command.extend(["-B", bind])
+    elif interface:
+        command.extend(["-B", get_ip_address(interface)])
+
+    if udp:
+        command.append("-u")
+        result = await run_command_async(command)
+        if result.stdout == "" and result.stderr:
+            raise RunCommandError(result.stderr, -1)
+        res = result.stdout.split("\n")[0].split(",")
+        return {
+            "timestamp": int(res[0]),
+            "source_address": res[1],
+            "source_port": int(res[2]),
+            "destination_address": res[3],
+            "destination_port": int(res[4]),
+            "transfer_id": int(res[5]),
+            "interval": [float(x) for x in res[6].split("-")],
+            "transferred_bytes": int(res[7]),
+            "transferred_mbytes": round(float(res[7]) / 1024 / 1024, 3),
+            "bps": int(res[8]),
+            "mbps": round(float(res[8]) / 1024 / 1024, 3),
+            "jitter": float(res[9]) if res[9] != "" else None,
+            "error_count": int(res[10]) if res[10] != "" else None,
+            "datagrams": int(res[11]) if res[11] != "" else None,
+            "extra_1": res[12],
+            "extra_2": res[13],
+        }
+    else:
+        result = await run_command_async(command)
+        if result.stdout == "" and result.stderr:
+            raise RunCommandError(result.stderr, -1)
+        res = result.stdout.split("\n")[0].split(",")
+        return {
+            "timestamp": int(res[0]),
+            "source_address": res[1],
+            "source_port": int(res[2]),
+            "destination_address": res[3],
+            "destination_port": int(res[4]),
+            "transfer_id": int(res[5]),
+            "interval": [float(x) for x in res[6].split("-")],
+            "transferred_bytes": int(res[7]),
+            "transferred_mbytes": round(float(res[7]) / 1024 / 1024, 3),
+            "bps": int(res[8]),
+            "mbps": round(float(res[8]) / 1024 / 1024, 3),
+        }
+
+
+async def run_iperf3_client(host: str, time: int = 10, bind_host: Optional[str] = None):
+    command = ["iperf3", "--json", "-t", str(time), "-c", host]
+    res = await run_command_async(command)
+    return res.output_from_json()