diff --git a/speedtest/__init__.py b/speedtest/__init__.py new file mode 100644 index 0000000..f91daf6 --- /dev/null +++ b/speedtest/__init__.py @@ -0,0 +1,5 @@ +from .speedtest import Speedtest + + +async def setup(bot): + await bot.add_cog(Speedtest(bot)) diff --git a/speedtest/info.json b/speedtest/info.json new file mode 100644 index 0000000..c36a53a --- /dev/null +++ b/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"] +} diff --git a/speedtest/models.py b/speedtest/models.py new file mode 100644 index 0000000..cd1ca07 --- /dev/null +++ b/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" diff --git a/speedtest/speedtest.py b/speedtest/speedtest.py new file mode 100644 index 0000000..e727b0f --- /dev/null +++ b/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})**")