config, set_version

modules/tile_gen/tile_gen.py

@@ -4,6 +4,7 @@ import click
 from tile_gen_lib.btrfs import make_btrfs
 from tile_gen_lib.planetiler import run_planetiler
 from tile_gen_lib.rclone import make_indexes_for_bucket, upload_area
+from tile_gen_lib.set_version import check_and_set_version
 
 
 @click.group()
@@ -48,5 +49,18 @@ def make_indexes():
         make_indexes_for_bucket(bucket)
 
 
+@cli.command()
+@click.argument('area', required=True)
+@click.option(
+    '--version', default='latest', help='Optional version string, like "20231227_043106_pt"'
+)
+def set_version(area, version):
+    """
+    Set versions for a given area
+    """
+
+    check_and_set_version(area, version)
+
+
 if __name__ == '__main__':
     cli()

modules/tile_gen/tile_gen_lib/set_version.py

@@ -0,0 +1,31 @@
+import subprocess
+
+from http_host_lib.config import config
+
+from tile_gen_lib.shared import get_versions_for_area
+
+
+def check_all_hosts(area, version):
+    pass
+
+
+def check_and_set_version(area, version):
+    if version == 'latest':
+        versions = get_versions_for_area(area)
+        version = versions[-1]
+
+    if not check_all_hosts(area, version):
+        return
+
+
+def set_version(area, version):
+    subprocess.run(
+        [
+            'rclone',
+            'rcat',
+            f'remote:ofm-assets/deployed_versions/{area}.txt',
+        ],
+        env=dict(RCLONE_CONFIG=config.rclone_config),
+        check=True,
+        input=version.strip().encode(),
+    )

modules/tile_gen/tile_gen_lib/shared.py

@@ -0,0 +1,92 @@
+import json
+from io import BytesIO
+from pathlib import Path
+
+import pycurl
+import requests
+
+
+def get_versions_for_area(area: str) -> list:
+    r = requests.get('https://btrfs.openfreemap.com/dirs.txt', timeout=30)
+    r.raise_for_status()
+
+    versions = [v.split('/')[2] for v in r.text.splitlines() if v.startswith(f'areas/{area}/')]
+    return sorted(versions)
+
+
+def check_host_version(domain, host_ip, area, version):
+    # check actual vector tile
+    url = f'https://{domain}/{area}/{version}/14/8529/5975.pbf'
+    assert pycurl_status(url, domain, host_ip) == 200
+
+    # check style
+    url = f'https://{domain}/styles/bright'
+    assert pycurl_status(url, domain, host_ip) == 200
+
+
+def check_host_latest(domain, host_ip, area, version):
+    # check TileJSON first
+    url = f'https://{domain}/{area}'
+    tilejson_str = pycurl_get(url, domain, host_ip)
+    tilejson = json.loads(tilejson_str)
+    tiles_url = tilejson['tiles'][0]
+    version_in_tilejson = tiles_url.split('/')[4]
+    assert version_in_tilejson == version
+
+    # check actual vector tile
+    url = f'https://{domain}/{area}/{version}/14/8529/5975.pbf'
+    assert pycurl_status(url, domain, host_ip) == 200
+
+    # check style
+    url = f'https://{domain}/styles/bright'
+    assert pycurl_status(url, domain, host_ip) == 200
+
+
+def pycurl_status(url, domain, host_ip):
+    """
+    Uses pycurl to make a HTTPS HEAD request using custom resolving,
+    checks if the status code is 200
+    """
+
+    c = pycurl.Curl()
+    c.setopt(c.URL, url)
+
+    # linux needs CA certs specified manually
+    if Path('/etc/ssl/certs/ca-certificates.crt').exists():
+        c.setopt(c.CAINFO, '/etc/ssl/certs/ca-certificates.crt')
+
+    c.setopt(c.RESOLVE, [f'{domain}:443:{host_ip}'])
+    c.setopt(c.NOBODY, True)
+    c.setopt(c.TIMEOUT, 5)
+    c.perform()
+    status_code = c.getinfo(c.RESPONSE_CODE)
+    c.close()
+
+    return status_code
+
+
+def pycurl_get(url, domain, host_ip):
+    """
+    Uses pycurl to make a HTTPS GET request using custom resolving,
+    checks if the status code is 200, and returns the content.
+    """
+
+    buffer = BytesIO()
+    c = pycurl.Curl()
+    c.setopt(c.URL, url)
+
+    # linux needs CA certs specified manually
+    if Path('/etc/ssl/certs/ca-certificates.crt').exists():
+        c.setopt(c.CAINFO, '/etc/ssl/certs/ca-certificates.crt')
+
+    c.setopt(c.RESOLVE, [f'{domain}:443:{host_ip}'])
+    c.setopt(c.WRITEDATA, buffer)
+    c.setopt(c.TIMEOUT, 5)
+    c.perform()
+    status_code = c.getinfo(c.RESPONSE_CODE)
+    c.close()
+
+    if status_code != 200:
+        raise ValueError(f'status code: {status_code}')
+
+    return buffer.getvalue().decode('utf8')