outils_OSM/osm_vc63/utils.py

#!/usr/bin/env python3


# Copyright 2021 Olav63, SebF
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.


"""Collections de méthodes utilitaires"""

import json
import logging
from collections import OrderedDict
import requests
from pyexcel_ods3 import save_data
from osm_vc63 import errors


class Utils:
    """Classe de méthodes utilitaires"""

    overpass_url: str
    geo_api_url: str
    dossier_sauvegarde: str

    def __init__(self, overpass_url, geo_api_url, dossier_sauvegarde):
        self.overpass_url = overpass_url
        self.geo_api_url = geo_api_url
        self.dossier_sauvegarde = dossier_sauvegarde

    def save_as_ods(self, fields, data, nom_req):
        """Sauvegarde de data dans un classeur ods"""

        ods_data_sheet = OrderedDict()
        ods_data = []
        ods_data.append(fields.keys())
        index_line = 2

        for element in data["elements"]:
            line = []
            index_col = 0

            for field in fields.keys():
                if field in element["tags"]:
                    if field == "capacity":
                        val = element["tags"][field]
                        line.append(int(val) if val.isdigit() else val)
                    else:
                        line.append(element["tags"][field])
                else:
                    line.append("")
                index_col = index_col + 1

            ods_data.append(line)
            index_line = index_line + 1

        ods_data_sheet.update({"resultats": ods_data})

        save_data(self.dossier_sauvegarde + nom_req + ".ods", ods_data_sheet)

        logging.info("Sauvegarde résultats format ODS")

    def save_as_json(self, export_json, nom_req):
        """Enregistrement du JSON"""

        json_file = open(self.dossier_sauvegarde + nom_req + ".json", "w")
        json_file.write(json.dumps(export_json))
        json_file.close()

        logging.info("Sauvegarde résultat format JSON/OSM")

    def nettoyage_json_pour_umap(self, data, overpass_query_fields):
        """Sélection uniquement des champs export_json == oui"""

        export_json = {
            "version": data["version"],
            "generator": data["generator"] + " and ETALAB API",
            "osm3s": data["osm3s"],
            "elements": [],
        }

        index_line = 0

        for element in data["elements"]:
            export_json["elements"].append(
                {"type": element["type"], "id": element["id"]}
            )

            # positionnement des éléments
            if element["type"] == "node":  # noeuds
                export_json["elements"][index_line]["lat"] = element["lat"]
                export_json["elements"][index_line]["lon"] = element["lon"]
            else:  # ways et relations
                export_json["elements"][index_line]["center"] = element["center"]
                export_json["elements"][index_line]["nodes"] = element["nodes"]

            # filtrage des tags
            description = ""
            for tag in overpass_query_fields.keys():
                if overpass_query_fields[tag]["export_json"] == "Oui":
                    if tag in element["tags"]:
                        if overpass_query_fields[tag]["FR"] != "":
                            description = (
                                description + overpass_query_fields[tag]["FR"] + " : "
                            )

                        description = description + str(element["tags"][tag]) + "\n"
            export_json["elements"][index_line]["tags"] = {"description": description}

            index_line = index_line + 1

        return export_json

    def run_overpass_query(self, critere, aire_de_recherche):
        """Envoie la requête Overpass et retourne la réponse JSON."""

        overpass_query = (
            """[out:json];
        (
        """
            + critere
            + """
        );
        out center;
        """
        )
        overpass_query = overpass_query.replace("aire_de_recherche", aire_de_recherche)

        response = requests.get(self.overpass_url, params={"data": overpass_query})

        if response.status_code != 200:
            raise errors.OverpassError(response.status_code)

        return response.json()

    def run_reverse_geocoding(self, lat, lon):
        """Retourne une adresse JSON à partir d'une position GPS."""

        url = self.geo_api_url + "/reverse/"

        response = requests.get(url, params={"lon": str(lon), "lat": str(lat)})

        if response.status_code != 200:
            raise errors.GeoApiError(response.status_code)

        return response.json()

    def geocodage(self, data):
        """Renseigne une adresse pour chaque élément de data"""

        for element in data["elements"]:

            if element["type"] == "node":
                rev_geocode = self.run_reverse_geocoding(element["lat"], element["lon"])
            else:
                rev_geocode = self.run_reverse_geocoding(
                    element["center"]["lat"], element["center"]["lon"]
                )

            api_adresse = rev_geocode["features"][0]

            element["tags"]["api_adresse:geometry:coordinates:lon"] = api_adresse[
                "geometry"
            ]["coordinates"][0]
            element["tags"]["api_adresse:geometry:coordinates:lat"] = api_adresse[
                "geometry"
            ]["coordinates"][1]

            element["tags"]["api_adresse:properties:label"] = api_adresse["properties"][
                "label"
            ]
            element["tags"]["api_adresse:properties:score"] = api_adresse["properties"][
                "score"
            ]

            if "housenumber" in api_adresse["properties"]:
                element["tags"]["api_adresse:properties:housenumber"] = api_adresse[
                    "properties"
                ]["housenumber"]

            element["tags"]["api_adresse:properties:type"] = api_adresse["properties"][
                "type"
            ]

            element["tags"]["api_adresse:properties:name"] = api_adresse["properties"][
                "name"
            ]
            element["tags"]["api_adresse:properties:postcode"] = api_adresse[
                "properties"
            ]["postcode"]
            element["tags"]["api_adresse:properties:citycode"] = api_adresse[
                "properties"
            ]["citycode"]
            element["tags"]["api_adresse:properties:city"] = api_adresse["properties"][
                "city"
            ]

            if "street" in api_adresse["properties"]:
                element["tags"]["api_adresse:properties:street"] = api_adresse[
                    "properties"
                ]["street"]

            element["tags"]["api_adresse:properties:attribution"] = rev_geocode[
                "attribution"
            ]
            element["tags"]["api_adresse:properties:licence"] = rev_geocode["licence"]

        logging.info("Géocodage inversé terminé")

        return data

    def traduction(self, tag, dictionnaire, data):
        """Traduit le champ tag des éléments de data avec dict"""

        for element in data["elements"]:
            if tag in element["tags"]:
                element["tags"][tag] = dictionnaire[element["tags"][tag]]

        return data