speedtest/__init__.py

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

speedtest/info.json

@@ -1,14 +0,0 @@
-{
-  "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

@@ -1,76 +0,0 @@
-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

@@ -1,71 +0,0 @@
-#   _____                         _
-#  / ____|                       (_)
-# | (___   ___  __ _ _____      ___ _ __ ___  _ __ ___   ___ _ __
-#  \___ \ / _ \/ _` / __\ \ /\ / / | '_ ` _ \| '_ ` _ \ / _ \ '__|
-#  ____) |  __/ (_| \__ \\ 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", "--accept-gdpr",
-                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})**")