a5aff40a-534f-4c34-9317-84fb0c43fac8

Build # a5aff40a-534f-4c34-9317-84fb0c43fac8

Build Type

push

circleci

Committed by Sym Roe

Commit Message

Management command for adding missing territory codes

Run Details

2330 of 3423 relevant lines covered (68.07%)

0.68 hits per line

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

0.0

/every_election/apps/api/management/commands/export_boundaries.py

import os
import os.path
import subprocess
from datetime import datetime

import geojson

from django.conf import settings
from django.core.management.base import BaseCommand
from elections.models import Election

TOPOJSON_BIN = os.path.join(
    settings.BASE_DIR, "..", "node_modules", "topojson", "node_modules", ".bin"
)


def set_precision(coords, precision):
    result = []
    try:
        return round(coords, int(precision))
    except TypeError:
        for coord in coords:
            result.append(set_precision(coord, precision))
    return result


def parse_date(date_string):
    return datetime.strptime(date_string, "%Y-%m-%d").date()


class Command(BaseCommand):
    help = "Export static boundary GeoJSON files for each group of elections."

    def add_arguments(self, parser):
        output_dir = os.path.join(settings.BASE_DIR, "static", "exports")
        parser.add_argument(
            "--from",
            dest="from",
            help="Export elections from this date",
            type=parse_date,
        )
        parser.add_argument(
            "--to", dest="to", help="Export elections until this date", type=parse_date
        )
        parser.add_argument(
            "--output",
            dest="output",
            help="Output directory (default every_election/static/exports)",
            default=output_dir,
        )

    def handle(self, *args, **options):
        try:
            os.mkdir(options["output"])
        except FileExistsError:
            pass

        if not (options["from"] or options["to"]):
            elections = Election.public_objects.future().filter(group_type="election")
        else:
            elections = Election.public_objects.all().filter(group_type="election")

            if options["from"]:
                elections = elections.filter(poll_open_date__gte=options["from"])

            if options["to"]:
                elections = elections.filter(poll_open_date__lte=options["to"])

        for election in elections:
            self.stdout.write("Exporting elections for group %s" % election)
            data = self.export_election(election)

            gj_path = os.path.join(options["output"], "%s.json" % election.election_id)
            with open(gj_path, "w") as output_file:
                geojson.dump(data, output_file)

            tj_path = os.path.join(
                options["output"], "%s-topo.json" % election.election_id
            )
            self.topojson_convert(gj_path, tj_path)
            tj_simple_path = os.path.join(
                options["output"], "%s-topo-simplified.json" % election.election_id
            )
            self.topojson_simplify(tj_path, tj_simple_path)

    def topojson_convert(self, source, dest):
        "Convert GeoJSON to TopoJSON by calling out to the topojson package"
        subprocess.check_call(
            [os.path.join(TOPOJSON_BIN, "geo2topo"), "-o", dest, source]
        )

    def topojson_simplify(self, source, dest):
        "Simplify a TopoJSON file"
        # The toposimplify settings here were arrived at by trial and error to keep the
        # simplified 2018-05-03 local elections topojson below 2.5MB.
        subprocess.check_call(
            [
                os.path.join(TOPOJSON_BIN, "toposimplify"),
                "-S",
                "0.2",
                "-F",
                "-o",
                dest,
                source,
            ]
        )

    def export_election(self, parent):
        "Return GeoJSON containing all leaf elections below this parent"
        features = []
        elections = self.get_ballots(parent)
        for election in elections:
            if election.geography:
                gj = geojson.loads(election.geography.geography.json)

                # Round coordinates to 6 decimal places (~10cm) precision to reduce
                # output size. This is probably as good as the source data accuracy.
                gj["coordinates"] = set_precision(gj["coordinates"], 6)

            else:
                self.stderr.write("Election %s has no geography" % election)
                gj = None

            feat = geojson.Feature(
                geometry=gj,
                id=election.election_id,
                properties={
                    "name": election.election_title,
                    "division": election.division.name if election.division else None,
                    "organisation": election.organisation.official_name,
                },
            )
            features.append(feat)
        return geojson.FeatureCollection(features, election_group=parent.election_id)

    def get_ballots(self, group):
        "Return the ballots for a group of elections."
        to_visit = [group]
        leaf_nodes = []
        while len(to_visit) > 0:
            e = to_visit.pop()
            children = e.get_children("public_objects").all()
            if e.identifier_type == "ballot":
                leaf_nodes.append(e)
            else:
                to_visit += children

        return leaf_nodes

1	import os	×
2	import os.path	×
3	import subprocess	×
4	from datetime import datetime	×
5
6	import geojson	×
7
8	from django.conf import settings	×
9	from django.core.management.base import BaseCommand	×
10	from elections.models import Election	×
11
12	TOPOJSON_BIN = os.path.join(	×
13	settings.BASE_DIR, "..", "node_modules", "topojson", "node_modules", ".bin"
14	)
15
16
17	def set_precision(coords, precision):	×
18	result = []	×
19	try:	×
20	return round(coords, int(precision))	×
21	except TypeError:	×
22	for coord in coords:	×
23	result.append(set_precision(coord, precision))	×
24	return result	×
25
26
27	def parse_date(date_string):	×
28	return datetime.strptime(date_string, "%Y-%m-%d").date()	×
29
30
31	class Command(BaseCommand):	×
32	help = "Export static boundary GeoJSON files for each group of elections."	×
33
34	def add_arguments(self, parser):	×
35	output_dir = os.path.join(settings.BASE_DIR, "static", "exports")	×
36	parser.add_argument(	×
37	"--from",
38	dest="from",
39	help="Export elections from this date",
40	type=parse_date,
41	)
42	parser.add_argument(	×
43	"--to", dest="to", help="Export elections until this date", type=parse_date
44	)
45	parser.add_argument(	×
46	"--output",
47	dest="output",
48	help="Output directory (default every_election/static/exports)",
49	default=output_dir,
50	)
51
52	def handle(self, args, *options):	×
53	try:	×
54	os.mkdir(options["output"])	×
55	except FileExistsError:	×
56	pass	×
57
58	if not (options["from"] or options["to"]):	×
59	elections = Election.public_objects.future().filter(group_type="election")	×
60	else:
61	elections = Election.public_objects.all().filter(group_type="election")	×
62
63	if options["from"]:	×
64	elections = elections.filter(poll_open_date__gte=options["from"])	×
65
66	if options["to"]:	×
67	elections = elections.filter(poll_open_date__lte=options["to"])	×
68
69	for election in elections:	×
70	self.stdout.write("Exporting elections for group %s" % election)	×
71	data = self.export_election(election)	×
72
73	gj_path = os.path.join(options["output"], "%s.json" % election.election_id)	×
74	with open(gj_path, "w") as output_file:	×
75	geojson.dump(data, output_file)	×
76
77	tj_path = os.path.join(	×
78	options["output"], "%s-topo.json" % election.election_id
79	)
80	self.topojson_convert(gj_path, tj_path)	×
81	tj_simple_path = os.path.join(	×
82	options["output"], "%s-topo-simplified.json" % election.election_id
83	)
84	self.topojson_simplify(tj_path, tj_simple_path)	×
85
86	def topojson_convert(self, source, dest):	×
87	"Convert GeoJSON to TopoJSON by calling out to the topojson package"
88	subprocess.check_call(	×
89	[os.path.join(TOPOJSON_BIN, "geo2topo"), "-o", dest, source]
90	)
91
92	def topojson_simplify(self, source, dest):	×
93	"Simplify a TopoJSON file"
94	# The toposimplify settings here were arrived at by trial and error to keep the
95	# simplified 2018-05-03 local elections topojson below 2.5MB.
96	subprocess.check_call(	×
97	[
98	os.path.join(TOPOJSON_BIN, "toposimplify"),
99	"-S",
100	"0.2",
101	"-F",
102	"-o",
103	dest,
104	source,
105	]
106	)
107
108	def export_election(self, parent):	×
109	"Return GeoJSON containing all leaf elections below this parent"
110	features = []	×
111	elections = self.get_ballots(parent)	×
112	for election in elections:	×
113	if election.geography:	×
114	gj = geojson.loads(election.geography.geography.json)	×
115
116	# Round coordinates to 6 decimal places (~10cm) precision to reduce
117	# output size. This is probably as good as the source data accuracy.
118	gj["coordinates"] = set_precision(gj["coordinates"], 6)	×
119
120	else:
121	self.stderr.write("Election %s has no geography" % election)	×
122	gj = None	×
123
124	feat = geojson.Feature(	×
125	geometry=gj,
126	id=election.election_id,
127	properties={
128	"name": election.election_title,
129	"division": election.division.name if election.division else None,
130	"organisation": election.organisation.official_name,
131	},
132	)
133	features.append(feat)	×
134	return geojson.FeatureCollection(features, election_group=parent.election_id)	×
135
136	def get_ballots(self, group):	×
137	"Return the ballots for a group of elections."
138	to_visit = [group]	×
139	leaf_nodes = []	×
140	while len(to_visit) > 0:	×
141	e = to_visit.pop()	×
142	children = e.get_children("public_objects").all()	×
143	if e.identifier_type == "ballot":	×
144	leaf_nodes.append(e)	×
145	else:
146	to_visit += children	×
147
148	return leaf_nodes	×

DemocracyClub / EveryElection / a5aff40a-534f-4c34-9317-84fb0c43fac8

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