78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
"""
|
|
This module simply returns the latency between a message being sent and the bot receiving it.
|
|
|
|
It also supports a rudimentary federation latency check, by sending a request to the target homeserver.
|
|
"""
|
|
|
|
import urllib.parse
|
|
|
|
import niobot
|
|
import time
|
|
import httpx
|
|
|
|
|
|
class LatencyModule(niobot.Module):
|
|
@niobot.command("latency", aliases=["ping"])
|
|
async def latency(self, ctx: niobot.Context, homeserver: str = None):
|
|
"""
|
|
See the bot's latency.
|
|
|
|
If you supply a homeserver, it will also show the federation latency to that homeserver.
|
|
"""
|
|
latency = ctx.latency
|
|
|
|
response = await ctx.respond(
|
|
"Latency: {:,.2f}ms{}".format(
|
|
latency, " (pinging homeserver...)" if homeserver else ""
|
|
)
|
|
)
|
|
if homeserver:
|
|
if not homeserver.startswith("http"):
|
|
homeserver = f"https://{homeserver}"
|
|
else:
|
|
return
|
|
|
|
parsed = urllib.parse.urlparse(homeserver)
|
|
loc = "https://" + parsed.netloc
|
|
try:
|
|
start_resolv = time.perf_counter()
|
|
homeserver = await niobot.resolve_homeserver(parsed.netloc)
|
|
end_resolv = time.perf_counter()
|
|
resolv = round(end_resolv - start_resolv, 2)
|
|
except ValueError as e:
|
|
await ctx.respond(
|
|
"\N{WARNING SIGN} Failed to resolve homeserver: " + str(e)
|
|
)
|
|
homeserver = loc
|
|
resolv = -1
|
|
|
|
async with httpx.AsyncClient(
|
|
headers={"User-Agent": niobot.__user_agent__}
|
|
) as client:
|
|
timings = []
|
|
for rnd in range(5):
|
|
start = time.perf_counter()
|
|
try:
|
|
res = await client.get(
|
|
f"{homeserver}/_matrix/federation/v1/version"
|
|
)
|
|
res.raise_for_status()
|
|
except httpx.HTTPError:
|
|
timings.append(-1)
|
|
else:
|
|
end = time.perf_counter()
|
|
timings.append(end - start)
|
|
if sum(timings) == -5:
|
|
return await response.edit(
|
|
"Latency: {:,.2f}ms (failed to connect to homeserver)".format(
|
|
latency
|
|
)
|
|
)
|
|
fed_latency = sum(timings) / len(timings) * 1000
|
|
|
|
hs_target = homeserver[8:]
|
|
return await ctx.respond(
|
|
"Latency: {:,.2f}ms ({:,.2f}ms to resolve homeserver, federation latency to {!r}: {:,.2f}ms)".format(
|
|
latency, resolv * 1000, hs_target, fed_latency
|
|
)
|
|
)
|