right-tree/backend/right_tree/api/resource_generation_utils.py

import csv
from io import StringIO
from os.path import splitext

from .filters import is_in_christchurch, is_in_auckland
from .models import EcologicalDistrictLayer, SoilOrder, Questionnaire, ActivationKey
from .wms_utils import get_address_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(questionnaire):
    """ Retrives the selected location data from the request.
    """
    filter_rows = [['LOCATION FILTERS:', ' ']]

    eco_district_layer = EcologicalDistrictLayer.objects.filter(geom__intersects=questionnaire.location).first()
    address = get_address_from_coordinates(questionnaire.location)

    filter_rows.append(['Point coordinates:', questionnaire.location])
    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 ' '])

    return filter_rows


def get_soil_filters(questionnaire):
    """ Retrives the selected soil type data from the request params.
    """
    filter_rows = [['SOIL FILTERS:', ' ']]

    soil_order = SoilOrder.objects.filter(soillayer__geom__intersects=questionnaire.location).first()

    filter_rows.append(['Soil Order:', f"{soil_order.name or ' '} ({soil_order.code or ' '})"])
    filter_rows.append(['Soil Variant:', questionnaire.soil_variant])

    return filter_rows


def get_site_filters(questionnaire):
    """ Retrives the selected site data from the request params
    """
    filter_rows = [['SITE FILTERS:', ' ']]

    zone = questionnaire.zone
    habitat = zone.habitat

    filter_rows.append(['Habitat:', habitat.name])
    filter_rows.append(['Zone Name:', zone.name])
    filter_rows.append(['Zone Variant:', zone.variant])
    filter_rows.append(['Zone Refined Variant:', zone.refined_variant])

    return filter_rows


def get_additional_region_info(questionnaire):
    """ If the location coordinates fall within the CHCH or Auckland regions then return a description of where to find more information.
    """
    if is_in_christchurch(questionnaire.location):
        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(questionnaire.location):
        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 = []

    try:
        ak = ActivationKey.objects.get(key=params.get("key", ""))
        q = Questionnaire.objects.get(key=ak)
    except (ActivationKey.DoesNotExist, Questionnaire.DoesNotExist):
        return filter_rows

    # Add all the location filters
    filter_rows += get_location_filters(q)
    filter_rows.append([' ', ' '])

    # Add the soil filters
    filter_rows += get_soil_filters(q)
    filter_rows.append([' ', ' '])

    # Add the project site filters
    filter_rows += get_site_filters(q)
    filter_rows.append([' ', ' '])

    filter_rows += get_additional_region_info(q)

    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,
    )