speedtest/__init__.py

@@ -0,0 +1,5 @@
+from .speedtest import Speedtest
+
+
+async def setup(bot):
+    await bot.add_cog(Speedtest(bot))

speedtest/info.json

@@ -0,0 +1,14 @@
+{
+  "author" : ["SeaswimmerTheFsh (seasw.)"],
+  "install_msg" : "Thank you for installing Speedtest!\nYou can find the source code of this cog [here](https://coastalcommits.com/SeaswimmerTheFsh/SeaCogs).",
+  "name" : "Speedtest",
+  "short" : "A collection of useful utilities.",
+  "description" : "A collection of useful utilities.",
+  "end_user_data_statement" : "This cog does not store end user data.",
+  "hidden": true,
+  "disabled": false,
+  "min_bot_version": "3.5.0",
+  "min_python_version": [3, 10, 0],
+  "tags" : ["utility", "information"],
+  "requirements": ["pydantic"]
+}

speedtest/models.py

@@ -0,0 +1,76 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class Speedtest(BaseModel):
+    type: str
+    timestamp: datetime
+    ping: "Ping"
+    download: "Bandwidth"
+    upload: "Bandwidth"
+    isp: str
+    interface: "Interface"
+    server: "Server"
+    result: "Result"
+
+    @classmethod
+    def from_json(cls, data: dict) -> "Speedtest":
+        return cls(
+            type=data["type"],
+            timestamp=datetime.fromisoformat(data["timestamp"]),
+            ping=Ping(**data["ping"]),
+            download=Bandwidth(**data["download"]),
+            upload=Bandwidth(**data["upload"]),
+            isp=data["isp"],
+            interface=Interface(**data["interface"]),
+            server=Server(**data["server"]),
+            result=Result(**data["result"])
+        )
+
+class Bandwidth(BaseModel):
+    bandwidth: float
+    bytes: int
+    elapsed: int
+    latency: "Latency"
+
+    @property
+    def mbps(self) -> float:
+        return self.bandwidth / 1_000_000
+
+class Latency(BaseModel):
+    iqm: float
+    low: float
+    high: float
+    jitter: float
+
+class Interface(BaseModel):
+    internalIp: str
+    name: str
+    macAddr: str
+    isVpn: bool
+    externalIp: str
+
+class Ping(BaseModel):
+    jitter: float
+    latency: float
+    low: float
+    high: float
+
+class Server(BaseModel):
+    id: int
+    name: str
+    location: str
+    country: str
+    host: str
+    port: int
+    ip: str
+
+class Result(BaseModel):
+    id: str
+    url: str
+    persisted: bool
+
+    @property
+    def image(self) -> str:
+        return self.url + ".png"

speedtest/speedtest.py

@@ -0,0 +1,71 @@
+#   _____                         _
+#  / ____|                       (_)
+# | (___   ___  __ _ _____      ___ _ __ ___  _ __ ___   ___ _ __
+#  \___ \ / _ \/ _` / __\ \ /\ / / | '_ ` _ \| '_ ` _ \ / _ \ '__|
+#  ____) |  __/ (_| \__ \\ V  V /| | | | | | | | | | | |  __/ |
+# |_____/ \___|\__,_|___/ \_/\_/ |_|_| |_| |_|_| |_| |_|\___|_|
+
+import asyncio
+import json
+import subprocess
+
+import discord
+from redbot.core import commands
+from redbot.core.bot import Red
+from redbot.core.utils import chat_formatting as cf
+
+from .models import Speedtest as sp
+
+
+class Speedtest(commands.Cog):
+    """A collection of random utilities."""
+
+    __author__ = ["SeaswimmerTheFsh"]
+    __version__ = "1.0.0"
+
+    def __init__(self, bot: Red):
+        self.bot = bot
+
+    def format_help_for_context(self, ctx: commands.Context) -> str:
+        pre_processed = super().format_help_for_context(ctx) or ""
+        n = "\n" if "\n\n" not in pre_processed else ""
+        text = [
+            f"{pre_processed}{n}",
+            f"Cog Version: **{self.__version__}**",
+            f"Author: {cf.humanize_list(self.__author__)}"
+        ]
+        return "\n".join(text)
+
+    async def run_speedtest(self) -> str | sp:
+        try:
+            process = await asyncio.create_subprocess_exec(
+                "speedtest", "-f", "json", "--accept-license",
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE
+            )
+        except FileNotFoundError:
+            return "Speedtest CLI is not installed."
+        stdout, stderr = await process.communicate()
+        if process.returncode != 0:
+            return stderr.decode("utf-8")
+        return sp.from_json(json.loads(stdout.decode("utf-8")))
+
+    @commands.command()
+    @commands.is_owner()
+    async def speedtest(self, ctx: commands.Context) -> None:
+        """Run a speedtest."""
+        msg = await ctx.maybe_send_embed("Running speedtest...")
+        async with ctx.typing():
+            speedtest = await self.run_speedtest()
+            if await ctx.embed_requested():
+                if not isinstance(speedtest, sp):
+                    await msg.edit(embed=discord.Embed(description=f"An error occurred! {speedtest}", color=discord.Colour.red()))
+                    return
+                embed = discord.Embed(title="Speedtest Results", url=speedtest.result.url, color=await ctx.embed_color())
+                embed.set_image(url=speedtest.result.image)
+                await msg.edit(embed=embed)
+            else:
+                if not isinstance(speedtest, sp):
+                    await msg.edit(content=f"An error occurred! \n`{speedtest}`")
+                    return
+                await msg.edit(content=f"**[Result]({speedtest.result.url})**")