nonsensebot/app/modules/latency.py

import urllib.parse

import niobot
import time
import httpx


class LatencyModule(niobot.Module):
    @niobot.command("latency")
    async def latency(self, ctx: niobot.Context, homeserver: str = None):
        """See the bot's latency."""
        latency = ctx.latency

        homeserver = homeserver or await niobot.resolve_homeserver(ctx.message.sender.split(":")[1])
        if not homeserver.startswith("http"):
            homeserver = f"https://{homeserver}"

        parsed = urllib.parse.urlparse(homeserver)
        loc = "https://" + parsed.netloc
        async with httpx.AsyncClient(headers={"User-Agent": niobot.__user_agent__}) as client:
            timings = []
            for rnd in range(5):
                start = time.time()
                await client.get(f"{loc}/_matrix/client/v1/ping")
                end = time.time()
                timings.append(end - start)
            fed_latency = sum(timings) / len(timings) * 1000

        hs_target = parsed.hostname
        return await ctx.respond(
            "Latency: {:,.2f}ms (federation latency to {!r}: {:,.2f}ms)".format(latency, hs_target, fed_latency)
        )