right-tree/backend/right_tree/api/resource_generation_utils.py
Matthew Northcott 3f9f816a7e [#40] Bulk PDF export
- backend changes
2023-02-22 15:08:30 +13:00

187 lines
6.4 KiB
Python

import csv
from io import StringIO
from os.path import splitext
from .filters import *
from .wms_utils import get_address_from_coordinates, get_point_from_coordinates
import pdfkit
import pandas as pd
from PyPDF2 import PdfFileMerger
from django.core.files.storage import get_storage_class
Storage = get_storage_class()
storage = Storage()
CSV_FILENAME = 'plants.csv'
PLANTING_GUIDE_PDF_FILENAME = 'planting_guide.pdf'
HEADER_FIELDS = ['Names', 'Growth Form / Max Height (m) / Spacing (m) / Forest Position', 'Moisture Preferences',
'Tolerances (Water / Drought / Frost / Salinity)', 'Ecosystem Services', 'Carbon Sequestration Rate', 'Planting Stage']
def get_location_filters(params):
""" Retrives the selected location data from the request.
"""
filter_rows = [['LOCATION FILTERS:', ' ']]
coordinates = params.get('coordinates')
if coordinates is not None:
eco_district_layer = ecological_district_coordinate_filter(
coordinates).first()
point = get_point_from_coordinates(coordinates)
address = get_address_from_coordinates(coordinates)
filter_rows.append(['Point coordinates:', point])
filter_rows.append(['Ecological region:', eco_district_layer.ecologic_1 or '' ])
filter_rows.append(['Ecological district:', eco_district_layer.ecologic_2 or ' '])
filter_rows.append(['Property address:', address['full_address'] if address is not None else ' '])
else:
filter_rows.append(["None specified", " "])
return filter_rows
def get_soil_filters(params):
""" Retrives the selected soil type data from the request params.
"""
filter_rows = [['SOIL FILTERS:', ' ']]
soil_variant = params.get('soilVariant')
coordinates = params.get('coordinates')
if soil_variant is not None and coordinates is not None:
soil_order_obj = soil_order_coordinate_filter(coordinates).first()
filter_rows.append(['Soil Order:', f"{soil_order_obj.name or ' '} ({soil_order_obj.code or ' '})"])
filter_rows.append(['Soil Variant:', soil_variant])
else:
filter_rows.append(["None specified", " "])
return filter_rows
def get_site_filters(params):
""" Retrives the selected site data from the request params
"""
filter_rows = [['SITE FILTERS:', ' ']]
habitat = params.get('habitat')
zone = params.get('zone')
if zone is not None and habitat is not None:
habitat_json = json.loads(habitat)
zone_json = json.loads(zone)
filter_rows.append(['Habitat:', habitat_json.get("name", " ")])
filter_rows.append(['Zone Name:', zone_json.get("name", " ")])
filter_rows.append(['Zone Variant:', zone_json.get("variant", " ")])
filter_rows.append(['Zone Refined Variant:', zone_json.get("refined_variant", " ")])
else:
filter_rows.append(["None specified", " "])
return filter_rows
def get_additional_region_info(params):
""" If the location coordinates fall within the CHCH or Auckland regions then return a description of where to find more information.
"""
coordinates = params.get('coordinates')
if coordinates is not None:
if is_in_christchurch(coordinates):
return [["Your location falls within the ecosystem type covered by the Christchurch Council ecosystem maps - further information can be obtained from Ōtautahi/Christchurch ecosystems map link to: https://ccc.govt.nz/environment/land/ecosystem-map", " "], [' ', ' ']]
elif is_in_auckland(coordinates):
return [["Your location falls within the ecosystem type covered by the Auckland Council Tiaki Tāmaki Makaurau Conservation map - further information can be obtained from tiaki Tāmaki Makaurau conservation Auckland - link to https://www.tiakitamakimakaurau.nz/conservation-map/", " "], [' ', ' ']]
return []
def get_filter_values(params):
""" Retrives all selected values/filters from the request parameters
"""
filter_rows = []
# Add all the location filters
filter_rows += get_location_filters(params)
filter_rows.append([' ', ' '])
# Add the soil filters
filter_rows += get_soil_filters(params)
filter_rows.append([' ', ' '])
# Add the project site filters
filter_rows += get_site_filters(params)
filter_rows.append([' ', ' '])
filter_rows += get_additional_region_info(params)
return filter_rows
def generate_csv(data: list[list[str]], output_filename: str) -> str:
with storage.open(output_filename, 'w') as f:
csv.writer(f).writerows(data)
return storage.path(output_filename)
def generate_pdf(data: list[list[str]], output_filename: str):
""" Generates a pdf from a csv given data and a csv generation method.
Requires an html file to be generated as an intermediate step.
"""
name, _ = splitext(output_filename)
csv_filepath = generate_csv(data, f"{name}.csv")
pdf_filepath = storage.path(output_filename)
with StringIO() as html_buf:
# Convert csv to html
pd.read_csv(csv_filepath).to_html(html_buf) # reading from buffer causes segfault :/
html_buf.seek(0)
# Convert html to pdf
pdfkit.from_file(html_buf, pdf_filepath)
return pdf_filepath
def merge_pdfs(pdfs: list[str], output_filename):
"""Merge a list of PDF filenames"""
output_filepath = storage.path(output_filename)
merger = PdfFileMerger()
for pdf in pdfs:
merger.append(pdf)
merger.write(output_filepath)
merger.close()
return output_filepath
def serialize_plants_queryset(plants_queryset):
return [HEADER_FIELDS] + [
[
plant.display_name,
plant.display_growth_form,
plant.moisture_preferences,
plant.plant_tolerances,
plant.ecosystem_services,
plant.carbon_sequestration,
plant.stage
] for plant in plants_queryset
]
def create_planting_guide_pdf(filter_data, plant_data, output_filename):
""" Creates a planting guide pdf document with a pre-generated planting guide with
filter and plant list tabular informtation appended.
"""
# TODO: space values appear as NaN... this should be fixed
return merge_pdfs(
[
storage.path(PLANTING_GUIDE_PDF_FILENAME),
generate_pdf(filter_data, f"{output_filename}.filters"),
generate_pdf(plant_data, f"{output_filename}.plants"),
],
output_filename,
)