f1338858-c0fa-44cb-ba00-f6491924f61b

Committed 19 Aug 2025 09:05AM UTC coverage: 72.061% (+0.3%) from 71.724%

Build # f1338858-c0fa-44cb-ba00-f6491924f61b

Build Type

Pull #2498

circleci

Committed by

awdem

Commit Message

add tippecanoe to CI

Pull Request Pull Request #2498: Feat/divset pmtiles

Run Details

98 of 110 new or added lines in 6 files covered. (89.09%)

11 existing lines in 1 file now uncovered.

3598 of 4993 relevant lines covered (72.06%)

0.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.1

/every_election/apps/organisations/pmtiles_creator.py

import subprocess

from django.db import connection
from organisations.models import DivisionGeography


class PMtilesCreator:
    def __init__(self, divset):
        self.divset = divset

    def create_pmtile(self, dest_dir):
        pmtiles_fp = f"{dest_dir}/{self.divset.pmtiles_file_name}"

        try:
            geojson_fp = self.extract_geojson(dest_dir)
            tippecanoe_command = f"tippecanoe -o {pmtiles_fp} -zg --drop-rate=2 --drop-densest-as-needed {geojson_fp} -l {self.divset.id}"
            subprocess.run(tippecanoe_command, shell=True, check=True)

            return pmtiles_fp
        except subprocess.CalledProcessError as e:
            # TODO: handle this error better
            print(f"Error creating pmtiles file: {e}")

    def extract_geojson(self, dest_dir):
        geojson_fp = f"{dest_dir}/{self.divset.id}.geojson"
        ogr_command = self.construct_ogr_command(geojson_fp)
        subprocess.run(ogr_command, shell=True, check=True)
        return geojson_fp

    def construct_ogr_command(self, geojson_fp):
        db_settings = connection.settings_dict
        sql_query = self.construct_sql_query(self.divset.id)

        db_connection_string = (
            f"dbname={db_settings['NAME']} "
            f"user={db_settings['USER']} "
            f"password={db_settings['PASSWORD']} "
            f"host={db_settings['HOST'] if db_settings['HOST'] else '127.0.0.1'} "
            f"port={db_settings['PORT'] if db_settings['PORT'] else '5432'}"
        )

        return (
            f'ogr2ogr -f "GeoJSON" {geojson_fp} '
            f'"PG:{db_connection_string}" '
            f'-sql "{sql_query}"'
        )

    def construct_sql_query(self, divisionset_id):
        sql, params = (
            DivisionGeography.objects.filter(
                division__divisionset_id=divisionset_id
            )
            .select_related("division")
            .values(
                "id",
                "geography",
                "source",
                "division_id",
                "division__name",
            )
            .query.sql_with_params()
        )
        sql = sql.replace("::bytea", "")  # Remove bytea typecast
        return sql % params  # Substitute parameters into the SQL query

1	import subprocess	1✔
2
3	from django.db import connection	1✔
4	from organisations.models import DivisionGeography	1✔
5
6
7	class PMtilesCreator:	1✔
8	def __init__(self, divset):	1✔
9	self.divset = divset	1✔
10
11	def create_pmtile(self, dest_dir):	1✔
12	pmtiles_fp = f"{dest_dir}/{self.divset.pmtiles_file_name}"	1✔
13
14	try:	1✔
15	geojson_fp = self.extract_geojson(dest_dir)	1✔
16	tippecanoe_command = f"tippecanoe -o {pmtiles_fp} -zg --drop-rate=2 --drop-densest-as-needed {geojson_fp} -l {self.divset.id}"	1✔
17	subprocess.run(tippecanoe_command, shell=True, check=True)	1✔
18
19	return pmtiles_fp	1✔
NEW 20	except subprocess.CalledProcessError as e:	×
21	# TODO: handle this error better
NEW 22	print(f"Error creating pmtiles file: {e}")	×
23
24	def extract_geojson(self, dest_dir):	1✔
25	geojson_fp = f"{dest_dir}/{self.divset.id}.geojson"	1✔
26	ogr_command = self.construct_ogr_command(geojson_fp)	1✔
27	subprocess.run(ogr_command, shell=True, check=True)	1✔
28	return geojson_fp	1✔
29
30	def construct_ogr_command(self, geojson_fp):	1✔
31	db_settings = connection.settings_dict	1✔
32	sql_query = self.construct_sql_query(self.divset.id)	1✔
33
34	db_connection_string = (	1✔
35	f"dbname={db_settings['NAME']} "
36	f"user={db_settings['USER']} "
37	f"password={db_settings['PASSWORD']} "
38	f"host={db_settings['HOST'] if db_settings['HOST'] else '127.0.0.1'} "
39	f"port={db_settings['PORT'] if db_settings['PORT'] else '5432'}"
40	)
41
42	return (	1✔
43	f'ogr2ogr -f "GeoJSON" {geojson_fp} '
44	f'"PG:{db_connection_string}" '
45	f'-sql "{sql_query}"'
46	)
47
48	def construct_sql_query(self, divisionset_id):	1✔
49	sql, params = (	1✔
50	DivisionGeography.objects.filter(
51	division__divisionset_id=divisionset_id
52	)
53	.select_related("division")
54	.values(
55	"id",
56	"geography",
57	"source",
58	"division_id",
59	"division__name",
60	)
61	.query.sql_with_params()
62	)
63	sql = sql.replace("::bytea", "") # Remove bytea typecast	1✔
64	return sql % params # Substitute parameters into the SQL query	1✔

DemocracyClub / EveryElection / f1338858-c0fa-44cb-ba00-f6491924f61b

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous