from airflow.hooks.base import BaseHook
from config import Config
from util.base import BaseUtil
from typing import Literal, Optional, Dict, Tuple
[docs]
class TileUtil:
"""
Tiling and tileserver utility functions to interact with tippecanoe,
and TileserverGL servers: https://johto.ready.net/ https://hoenn.ready.net/ https://kanto.ready.net/
"""
[docs]
@staticmethod
def get_all_tiles_on_server(
server_alias: Literal["johto", "hoenn", "kanto"],
) -> list[str]:
"""
Get all the tiles on the server
:param server_alias: the alias to check
:return: the list of tiles
"""
import requests
from bs4 import BeautifulSoup
server_url = f"https://{server_alias}.ready.net"
folder_contents = requests.get(server_url)
soup = BeautifulSoup(folder_contents.text, "html.parser")
links = [link.get("href", "") for link in soup.find_all("a")]
filtered_links = filter(lambda x: x.endswith(".json"), links)
return [f"{server_url}{link}" for link in filtered_links]
[docs]
@staticmethod
def is_mbtile_valid(path: str) -> bool:
"""Check if the mbtile is valid
:param path: the path to the mbtile
:type path: str
:return: True if the mbtile is valid, False otherwise
:rtype: bool
"""
import sqlite3
con = None
try:
con = sqlite3.connect(path)
cur = con.cursor()
cur.execute("PRAGMA integrity_check")
except Exception as ex:
print("Error: ", ex)
return False
finally:
if con:
con.close()
print("healthy")
return True
[docs]
@staticmethod
def tileserver_is_healthy(server_alias: Literal["johto", "hoenn", "kanto"]):
"""Check if a tileset exists in the server
:param tile_id: The id of the tileset
:type tile_id: str
:param server_alias: The alias of the server
:type server_alias: str
:return: True if the tileset exists, False otherwise
:rtype: bool
"""
import requests
tileset_url = f"https://{server_alias}.ready.net/health"
print(f"Checking health of {tileset_url}")
return requests.get(tileset_url, verify=False).status_code == 200
[docs]
@staticmethod
def tileset_exist_on_server(
tile_id: str, server_alias: Literal["johto", "hoenn", "kanto"]
) -> bool:
"""Check if a tileset exists in the server
:param tile_id: The id of the tileset
:type tile_id: str
:param server_alias: The alias of the server
:type server_alias: str
:return: True if the tileset exists, False otherwise
:rtype: bool
"""
import requests
tileset_url = f"https://{server_alias}.ready.net/data/{tile_id}.json"
print(f"Checking if tileset {tile_id} exists in {tileset_url}")
return requests.get(tileset_url, verify=False).status_code == 200
[docs]
@staticmethod
def get_tileset_info(
env: Literal["prod", "staging", "dev", "johto", "hoenn", "kanto"],
title_filter: Optional[str] = None,
use_regex: bool = False,
) -> Dict[str, Tuple[str, dict]]:
"""
Get all the tileset information from the given environment's tile server
:param env: Environment name ("prod", "staging", "dev") or server alias ("kanto", "hoenn", "johto")
:param title_filter: The filter to apply to the tileset titles.
:param use_regex: Whether to use regex for the title filter.
:return: A dictionary of tileset information.
"""
import os
import re
import requests
import subprocess
from pathlib import Path
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# Map environment name to server alias if needed
server_alias = env
if env == "prod":
server_alias = "kanto"
elif env == "staging":
server_alias = "hoenn"
elif env == "dev":
server_alias = "johto"
url = f"https://{server_alias}.ready.net/index.json"
try:
# Disable SSL verification
tileset_json = requests.get(url, timeout=10, verify=False).json()
print(f"Warning: SSL verification disabled for {url}")
except requests.RequestException as e:
print(f"Error fetching tileset info from {url}: {e}")
return {}
pattern = re.compile(title_filter) if use_regex and title_filter else None
tileset_info_dict = {}
for item in tileset_json:
id_or_title = item.get("id")
if title_filter:
if (use_regex and not pattern.search(id_or_title)) or (
not use_regex and title_filter not in id_or_title
):
continue
layer_info = {
layer.get("layer"): layer.get("count")
for layer in item.get("tilestats", {}).get("layers", [])
}
tileset_info_dict[id_or_title] = ("unknown", layer_info)
TILES_DIR = "tiles"
# GET .mbtiles file size from filesystem
command = f"ssh {server_alias} \"find ~/{TILES_DIR} -type f -name '*.mbtiles' -exec ls -lh {{}} + | sort -k 5 -h\""
print(f"Executing command: {command}")
try:
result = subprocess.run(
command,
shell=True,
check=True,
capture_output=True,
text=True,
env=os.environ.copy(),
)
for line in result.stdout.strip().split("\n"):
if line.strip():
parts = line.split()
id_or_title = Path(parts[-1]).stem
size = parts[4]
if id_or_title in tileset_info_dict:
tileset_info_dict[id_or_title] = (
size,
tileset_info_dict[id_or_title][1],
)
except subprocess.CalledProcessError as e:
print(f"Error connecting to {server_alias}: {e.stderr}")
return tileset_info_dict