planheatproject.py

# -*- coding: utf-8 -*-
"""
/***************************************************************************
 PlanHeat
                                 A QGIS plugin
 The PlanHeat tool is the heart of the PLANHEAT project.
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
                              -------------------
        begin                : 2019-04-25
        git sha              : $Format:%H$
        copyright            : (C) 2019 by PlanHeat consortium
        email                : stefano.barberis@rina.org
 ***************************************************************************/
"""
import os
import shutil
import uuid
import datetime
import json
from qgis.core import QgsProject, QgsLayerTreeGroup, QgsMapLayer, QgsLayerTreeLayer
from . import config

class PlanHeatProject:
    """This object represents a planheat project with all its information."""

    WORKING_PATH = os.path.join(os.environ["LOCALAPPDATA"], "QGIS/QGIS3", config.WORKING_DIRECTORY_NAME)
    TEMP_WORKING_PATH = os.path.join(WORKING_PATH, config.TEMP_WORKING_DIRECTORY_NAME)
    MAPPING_CLEAN_DB_FILE_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              config.PROJECTS_CLEAN_DATA_DIR,
                                              "planheat_cmm_smm_clean.db")

    def __init__(self, infos: dict, empty=False):
        # Sub modules
        self.mapping_module = None
        self.planning_module = None
        # Infos
        if not empty:
            self.infos = infos
            self.set_up_additional_infos()
            # Working directory
            self.infos[config.PROJECT_ID_KEY] = str(uuid.uuid4())
            self.working_directory_name = self.infos[config.PROJECT_ID_KEY]
            self.create_project_directories()
            self.init_mapping()
            self.serialize()
        else:
            self.infos = None
            self.working_directory_name = None

    def get_name(self):
        return self.infos.get(config.PROJECT_NAME_KEY, config.PROJECT_NAME_DEFAULT)

    def get_mapping_directory_path(self):
        return os.path.join(self.get_full_working_path(),
                            config.MAPPING_DIRECTORY_NAME)

    def get_planning_directory_path(self):
        return os.path.join(self.get_full_working_path(),
                            config.PLANNING_DIRECTORY_NAME)

    def get_mapping_cmm_smm_db_path(self):
        return os.path.join(self.get_mapping_directory_path(),
                            config.MAPPING_CMM_SMM_DB_FILE_NAME)

    def set_up_additional_infos(self):
        now = datetime.datetime.now()
        self.infos[config.PROJECT_CREATION_DATE_KEY] = now.strftime(config.DATE_FMT)

    def get_full_working_path(self):
        project_full_path = os.path.join(PlanHeatProject.WORKING_PATH, self.working_directory_name)
        return project_full_path

    def create_project_directories(self):
        project_full_path = self.get_full_working_path()
        if not os.path.exists(project_full_path):
            os.mkdir(project_full_path)
        mapping_directory = os.path.join(project_full_path, config.MAPPING_DIRECTORY_NAME)
        if not os.path.exists(mapping_directory):
            os.mkdir(mapping_directory)
        input_directory = os.path.join(mapping_directory, config.INPUT_FOLDER)
        if not os.path.exists(input_directory):
            os.mkdir(input_directory)
        planning_directory = os.path.join(project_full_path, config.PLANNING_DIRECTORY_NAME)
        if not os.path.exists(planning_directory):
            os.mkdir(planning_directory)
        district_planning_directory = os.path.join(planning_directory, config.DISTRICT_FOLDER)
        if not os.path.exists(district_planning_directory):
            os.mkdir(district_planning_directory)
        city_planning_directory = os.path.join(planning_directory, config.CITY_FOLDER)
        if not os.path.exists(city_planning_directory):
            os.mkdir(city_planning_directory)
        for subfolder in [config.BASELINE_FOLDER, config.FUTURE_FOLDER, config.TOOLS_FOLDER, config.KPIS_FOLDER]:
            os.mkdir(os.path.join(district_planning_directory, subfolder))
        origin_TJulia = os.path.join(   os.path.dirname(os.path.realpath(__file__)),
                                        "planning_and_simulation_modules","Tjulia")
        new_TJulia = os.path.join(district_planning_directory, config.KPIS_FOLDER, "Tjulia")
        copy_TJulia(origin_TJulia, new_TJulia, extension=".csv")

    def init_mapping(self):
        project_full_path = self.get_full_working_path()
        mapping_dir = os.path.join(project_full_path, config.MAPPING_DIRECTORY_NAME)
        # Copy clean db
        new_db_path = os.path.join(mapping_dir, config.MAPPING_CMM_SMM_DB_FILE_NAME)
        shutil.copy(self.MAPPING_CLEAN_DB_FILE_PATH, new_db_path)

    def serialize(self):
        info_file = os.path.join(self.get_full_working_path(), config.PROJECT_INFO_JSON_NAME)
        with open(info_file, "w") as f:
            json.dump(self.infos, f)
        # TODO: serialize mapping if any specific objects that are not file managed
        # TODO: serialize planning

    @staticmethod
    def load(infos: dict):
        """Load a planheat project object from a property dict."""
        php = PlanHeatProject({}, empty=True)
        php.infos = infos.copy()
        php.working_directory_name = infos[config.PROJECT_ID_KEY]
        return php

    @staticmethod
    def deserialize(directory_path: str):
        info_file = os.path.join(directory_path, config.PROJECT_INFO_JSON_NAME)
        info = PlanHeatProject.read_info(info_file)
        if info is None or (not PlanHeatProject.info_is_valid(info)):
            return None
        php = PlanHeatProject(info)
        # Copy all the submodules files
        PlanHeatProject.copy_submodules_dir(directory_path, php.get_full_working_path())
        return php

    @staticmethod
    def read_info(file_path: str):
        try:
            with open(file_path, 'r') as f:
                infos = json.load(f)
            return infos
        except json.decoder.JSONDecodeError:
            return None

    def duplicate(self):
        """Duplicate the current project. The return project will have a new id."""
        # Create the new php
        new_php = PlanHeatProject(self.infos.copy())
        # Copy all the submodules files
        new_working_dir = new_php.get_full_working_path()
        old_working_dir = self.get_full_working_path()
        PlanHeatProject.copy_submodules_dir(old_working_dir, new_working_dir)
        return new_php

    @staticmethod
    def copy_submodules_dir(source_dir, target_dir):
        for sub_dir in [config.MAPPING_DIRECTORY_NAME, config.PLANNING_DIRECTORY_NAME]:
            new_dir = os.path.join(target_dir, sub_dir)
            if os.path.exists(new_dir):
                shutil.rmtree(new_dir)
            old_dir = os.path.join(source_dir, sub_dir)
            shutil.copytree(old_dir , new_dir)

    def tear_down(self):
        """Tear down a project with its path."""
        layers_to_remove = self.get_current_project_involved_layers()
        QgsProject.instance().removeMapLayers([c.layer().id() for c in layers_to_remove])
        QgsProject.instance().layerTreeRoot().removeChildrenGroupWithoutLayers()
        shutil.rmtree(self.get_full_working_path())

    def get_current_project_involved_layers(self):
        """Return all the layer that have a data source located in the project directory"""
        involved_layers = []
        tree = QgsProject.instance().layerTreeRoot()
        children_stack = tree.children()
        while len(children_stack) > 0:
            c = children_stack.pop()
            if isinstance(c, QgsLayerTreeGroup):  # Explore sub children
                children_stack.extend(c.children())
            elif isinstance(c, QgsLayerTreeLayer): # Check if layer is a project layer
                l = c.layer()
                if l.type() == QgsMapLayer.VectorLayer or l.type() == QgsMapLayer.RasterLayer:
                    source_uri = l.dataProvider().dataSourceUri()
                    if self.infos[config.PROJECT_ID_KEY] in source_uri:
                        involved_layers.append(c)
        return involved_layers

    @staticmethod
    def info_is_valid(info: dict) -> bool:
        """Check that all mandatory properties are present."""
        if config.PROJECT_NAME_KEY not in info:
            return False
        if config.PROJECT_ID_KEY not in info:
            return False
        return True

    @staticmethod
    def directory_is_valid(dir: str) -> bool:
        """Check that all mandatory files are present."""
        if not os.path.exists(dir):
            return False
        info_path = os.path.join(dir, config.PROJECT_INFO_JSON_NAME)
        if not os.path.exists(info_path):
            return False
        mapping_db_path = os.path.join(dir, config.MAPPING_DIRECTORY_NAME, config.MAPPING_CMM_SMM_DB_FILE_NAME)
        if not os.path.exists(mapping_db_path):
            return False
        return True


    @staticmethod
    def get_creation_date(info):
        creation_date_str = info.get(config.PROJECT_CREATION_DATE_KEY, None)
        if creation_date_str is None:
            return datetime.datetime.min
        try:
            creation_date = datetime.datetime.strptime(creation_date_str, config.DATE_FMT)
        except ValueError:
            creation_date = None
        if creation_date is None:
            return datetime.datetime.min
        return creation_date



def copy_TJulia(from_dir, to_dir, extension=".csv", nav_path=""):
    os.makedirs(os.path.join(to_dir, nav_path), exist_ok=True)
    for f in os.listdir(os.path.join(from_dir, nav_path)):
        if os.path.isdir(os.path.join(from_dir, nav_path, f)):
            copy_TJulia(from_dir, to_dir, nav_path=os.path.join(nav_path, f))
        elif f.endswith(extension):
            shutil.copy2(   os.path.join(from_dir, nav_path, f), 
                            os.path.join(to_dir, nav_path, f))