Source code for aequilibrae.project.project

import logging
import os
import shutil
import sqlite3
import warnings

from aequilibrae import global_logger
from aequilibrae.log import Log
from aequilibrae.parameters import Parameters
from aequilibrae.project.about import About
from aequilibrae.project.data import Matrices
from aequilibrae.project.database_connection import database_connection
from aequilibrae.context import activate_project, get_active_project
from aequilibrae.project.network import Network
from aequilibrae.project.zoning import Zoning
from aequilibrae.reference_files import spatialite_database
from aequilibrae.log import get_log_handler
from aequilibrae.project.project_cleaning import clean
from aequilibrae.project.project_creation import initialize_tables
from aequilibrae.transit.transit import Transit




class Project:
    """AequilibraE project class

    .. code-block:: python
        :caption: Create Project

        >>> new_project = Project()
        >>> new_project.new(project_path)

    .. code-block:: python
        :caption: Open Project

        >>> existing_project = Project()
        >>> existing_project.open(project_path)
    """



    def __init__(self):
        self.path_to_file: str = None
        self.project_base_path = ""
        self.source: str = None
        self.conn: sqlite3.Connection = None
        self.network: Network = None
        self.about: About = None
        self.logger: logging.Logger = None
        self.transit: Transit = None




    @classmethod
    def from_path(cls, project_folder):
        project = Project()
        project.open(project_folder)
        return project




    def open(self, project_path: str) -> None:
        """
        Loads project from disk

        :Arguments:
            **project_path** (:obj:`str`): Full path to the project data folder. If the project inside does
            not exist, it will fail.
        """

        file_name = os.path.join(project_path, "project_database.sqlite")
        if not os.path.isfile(file_name):
            raise FileNotFoundError("Model does not exist. Check your path and try again")
        self.project_base_path = project_path
        self.path_to_file = file_name
        self.source = self.path_to_file
        self.__setup_logger()
        self.activate()

        self.conn = self.connect()

        self.__load_objects()
        global_logger.info(f"Opened project on {self.project_base_path}")
        clean(self)




    def new(self, project_path: str) -> None:
        """Creates a new project

        :Arguments:
            **project_path** (:obj:`str`): Full path to the project data folder. If folder exists, it will fail
        """

        self.project_base_path = project_path
        self.path_to_file = os.path.join(self.project_base_path, "project_database.sqlite")
        self.source = self.path_to_file

        if os.path.isdir(project_path):
            raise FileExistsError("Location already exists. Choose a different name or remove the existing directory")

        # We create the project folder and create the base file
        os.mkdir(self.project_base_path)

        self.__setup_logger()
        self.activate()

        self.__create_empty_network()
        self.__load_objects()
        self.about.create()
        global_logger.info(f"Created project on {self.project_base_path}")




    def close(self) -> None:
        """Safely closes the project"""
        if not self.project_base_path:
            global_logger.warning("This Aequilibrae project is not opened")
            return

        try:
            self.conn.commit()
            clean(self)
            self.conn.close()
            for obj in [self.parameters, self.network]:
                del obj

            del self.network.link_types
            del self.network.modes

            global_logger.info(f"Closed project on {self.project_base_path}")

        except (sqlite3.ProgrammingError, AttributeError):
            global_logger.warning(f"This project at {self.project_base_path} is already closed")

        finally:
            self.deactivate()




    def connect(self):
        return database_connection("network", self.project_base_path)




    def activate(self):
        activate_project(self)




    def deactivate(self):
        if get_active_project(must_exist=False) is self:
            activate_project(None)




    def log(self) -> Log:
        """Returns a log object

        allows the user to read the log or clear it"""
        return Log(self.project_base_path)


    def __load_objects(self):
        matrix_folder = os.path.join(self.project_base_path, "matrices")
        if not os.path.isdir(matrix_folder):
            os.mkdir(matrix_folder)

        self.network = Network(self)
        self.about = About(self)
        self.matrices = Matrices(self)

    @property
    def project_parameters(self) -> Parameters:
        return Parameters(self)

    @property
    def parameters(self) -> dict:
        return self.project_parameters.parameters



    def check_file_indices(self) -> None:
        """Makes results_database.sqlite and the matrices folder compatible with project database"""
        raise NotImplementedError


    @property
    def zoning(self):
        return Zoning(self.network)

    def __create_empty_network(self):
        shutil.copyfile(spatialite_database, self.path_to_file)

        self.conn = self.connect()

        # Write parameters to the project folder
        p = self.project_parameters
        p.parameters["system"]["logging_directory"] = self.project_base_path
        p.write_back()

        # Create actual tables
        self.conn.execute("PRAGMA foreign_keys = ON;")
        self.conn.commit()
        initialize_tables(self, "network")

    def __setup_logger(self):
        self.logger = logging.getLogger(f"aequilibrae.{self.project_base_path}")
        self.logger.propagate = False
        self.logger.setLevel(logging.DEBUG)

        par = self.parameters or self.project_parameters._default
        do_log = par["system"]["logging"]

        if do_log:
            log_file = os.path.join(self.project_base_path, "aequilibrae.log")
            self.logger.addHandler(get_log_handler(log_file))