cambot/spoutnik.py

import random
import re
import requests
from nextcord import File
from unidecode import unidecode
from datetime import datetime
from settings import *
import geopandas as gpd
from shapely.geometry import Point
from geopy.distance import geodesic
from geopy.geocoders import Nominatim
from math import sinh, atan, pi, log, tan, cos, ceil

geolocator = Nominatim(user_agent="discord-bot-spoutnik")

def tile_to_wgs84(x, y, z):
    return atan(sinh(pi * (1 - 2 * y / 2 ** z))) * 180 / pi, (x / 2 ** z - 0.5) * 360


def wgs84_to_tile(lat, lon, z):
    return int((lon + 180) / 360 * 2 ** z), int((1 - (log(tan(lat * pi / 180) + 1 / cos(lat * pi / 180))) / pi) / 2 * 2 ** z)


def random_point(filename):
    shape = gpd.read_file(filename).union_all()

    min_x, min_y, max_x, max_y = shape.bounds

    while True:
        random_x = random.uniform(min_x, max_x)
        random_y = random.uniform(min_y, max_y)
        point = Point(random_x, random_y)

        if shape.contains(point):
            return point.coords[0][::-1]

def clean_address(address):
    output = address
    output = re.sub(r", Lausanne, District de Lausanne, Vaud, \d+, Suisse", "", output)
    return output

class Game:
    def __init__(self, channel):
        self.channel = channel
        self.target = None
        self.preset = None
        self.file = f"/tmp/spoutnik_{channel}.jpg"
        self.winner = None
        self.last_player = None
        self.last_datetime = None
        self.resetters = set()
        self.jump_url = None

    async def reset(self, preset, force=False):
        if not force and not self.winner and self.last_datetime and (datetime.now() - self.last_datetime).total_seconds() < SPOUTNIK_EXPIRY_TIMEOUT:
            return

        output = ""
        self.preset = preset
        zoom = self.preset["zoom"]
        if self.target and not self.winner:
            output = f":expressionless: Le lieu précédent était [ici](<https://www.google.ch/maps/@{self.target[0]},{self.target[1]},{zoom}z>).\n\n"
        random_lat, random_lon = random_point(self.preset["geojson_file"])
        tile_x, tile_y = wgs84_to_tile(random_lat, random_lon, zoom)
        req = requests.get(f"https://khms2.google.com/kh/v=1000?x={tile_x}&y={tile_y}&z={zoom}")
        with open(self.file, "wb") as f:
            f.write(req.content)
        self.target = tile_to_wgs84(tile_x + 0.5, tile_y + 0.5, zoom)
        self.winner = None
        self.last_player = None
        self.last_datetime = None
        self.resetters = set()
        output += f":satellite_orbital: Il y a un nouveau lieu à trouver ! Le terrain de jeu est {self.preset["label"]}."
        output += f"\n\nDonnez-moi les coordonnées GPS du centre de l'image, à {self.preset["max_distance"]} m près."
        output += f"\n\n*Sur le site Google Maps, clic droit puis clic sur les coordonnées.*"
        output += f"\n*Sur l'app Google Maps, appui long puis appui sur les coordonnées.*"
        self.jump_url = (await self.channel.send(output, file=File(self.file))).jump_url

    async def parse(self, message):
        guess = unidecode(message.content.strip()).upper()

        # if it's the special keyword 'reset', consider resetting
        if guess == "RESET" and SPOUTNIK_RESETTERS_NEEDED != 0:
            if message.author in self.resetters:
                return
            self.resetters.add(message.author)
            await self.channel.send(f"Réinitialisation : {len(self.resetters)}/{SPOUTNIK_RESETTERS_NEEDED}")
            if len(self.resetters) >= SPOUTNIK_RESETTERS_NEEDED:
                await self.reset(preset=self.preset, force=True)
                return

        # if somebody has already won, return silently
        if self.winner:
            return

        # same if the game was never initialized
        if not self.target:
            return

        # same if the message is not WGS 84 coordinates
        if not (regex := re.match(r"^(\d{1,2}[.,]\d+) *[ ,.;:] *(\d{1,3}[.,]\d+)$", guess)):
            return

        # check for errors and react accordingly
        error = False
        if SPOUTNIK_FORCE_ALTERNATION and message.author == self.last_player and (elapsed := (datetime.now() - self.last_datetime).total_seconds()) < SPOUTNIK_ALTERNATION_TIMEOUT:
            await message.add_reaction("\N{BUSTS IN SILHOUETTE}")
            error = True
            remaining = ceil(SPOUTNIK_ALTERNATION_TIMEOUT - elapsed)
            await self.channel.send(f"{message.author.mention} : tu pourras rejouer dans {remaining} seconde{'s' if remaining > 1 else ''}")
        if error:
            return

        # everything is OK, validate the proposal
        self.last_player = message.author
        self.last_datetime = datetime.now()

        lat = float(regex.group(1).replace(",", "."))
        lon = float(regex.group(2).replace(",", "."))

        distance = geodesic((lat, lon), self.target).meters
        blurred_distance = f"{ceil(distance / 1000)} km" if distance > 1000 else f"{ceil(distance / 100) * 100} m"
        zoom = self.preset["zoom"]
        if distance <= self.preset["max_distance"]:
            self.winner = message.author
            self.resetters = set()
            output = f":trophy: YOUPI ! {message.author.mention} a trouvé le lieu ! :trophy:"
            output += f"\n\n[Lieu exact](<https://www.google.ch/maps/@{self.target[0]},{self.target[1]},{zoom}z>)"
            await self.channel.send(output)
        else:
            try:
                address = clean_address(geolocator.reverse((lat, lon), language="fr").address)
            except ValueError:
                address = "un lieu apparemment invalide"
            output = f"{message.author.mention} propose [un point](<https://www.google.ch/maps/@{lat},{lon},{zoom}z>) proche de {address}."
            output += f"\n\nC'est pas ça chef, mais tu es à moins de {blurred_distance} de [la cible](<{self.jump_url}>) !"
            await self.channel.send(output)