Create Mastr.translate method #461

.pre-commit-config.yaml

@@ -3,4 +3,4 @@ repos:
     rev: 22.6.0
     hooks:
       - id: black
-        language_version: python3.10
+        language_version: python3.8

environment.yml

@@ -1,6 +1,6 @@
-name: py38_open_mastr
+name: py39_open_mastr
 channels:
     - conda-forge
     - defaults
 dependencies:
-    - python=3.8
+    - python=3.9

open_mastr/mastr.py

@@ -1,4 +1,5 @@
 import os
+from sqlalchemy import inspect, create_engine
 
 # import xml dependencies
 from open_mastr.xml_download.utils_download_bulk import download_xml_Mastr
@@ -27,14 +28,15 @@
     create_data_dir,
     get_data_version_dir,
     get_project_home_dir,
-    get_output_dir,
     setup_logger,
 )
 import open_mastr.utils.orm as orm
 
 # import initialize_database dependencies
 from open_mastr.utils.helpers import (
     create_database_engine,
+    rename_table,
+    create_translated_database_engine,
 )
 
 # constants
@@ -62,16 +64,27 @@ class Mastr:
     ------------
         engine: {'sqlite', sqlalchemy.engine.Engine}, optional
             Defines the engine of the database where the MaStR is mirrored to. Default is 'sqlite'.
+
+        connect_to_translated_db: boolean, optional
+            Allows connection to an existing translated database. Default is 'False'.
+            Only for 'sqlite'-type engines.
     """
 
-    def __init__(self, engine="sqlite") -> None:
+    def __init__(self, engine="sqlite", connect_to_translated_db=False) -> None:
+
         validate_parameter_format_for_mastr_init(engine)
-        self.output_dir = get_output_dir()
+
         self.home_directory = get_project_home_dir()
-        self._sqlite_folder_path = os.path.join(self.output_dir, "data", "sqlite")
+        self._sqlite_folder_path = os.path.join(self.home_directory, "data", "sqlite")
         os.makedirs(self._sqlite_folder_path, exist_ok=True)
 
-        self.engine = create_database_engine(engine, self.output_dir)
+        self.is_translated = connect_to_translated_db
+        if connect_to_translated_db:
+            self.engine = create_translated_database_engine(
+                engine, self._sqlite_folder_path
+            )
+        else:
+            self.engine = create_database_engine(engine, self.home_directory)
 
         print(
             f"Data will be written to the following database: {self.engine.url}\n"
@@ -142,7 +155,7 @@ def download(
             Either "today" or None if the newest data dump should be downloaded
             rom the MaStR website. If an already downloaded dump should be used,
             state the date of the download in the format
-            "yyyymmdd" or use the string "existing". Defaults to None.
+            "yyyymmdd". Defaults to None.
 
             For API method:
 
@@ -214,12 +227,19 @@ def download(
             method, data, api_data_types, api_location_types, **kwargs
         )
 
-        date = transform_date_parameter(self, method, date, **kwargs)
+        date = transform_date_parameter(method, date, **kwargs)
+
+        if self.is_translated:
+            raise TypeError(
+                "you are currently connected to a translated database\n"
+                "a translated database cannot be further altered\n"
+                "translate a new database to replace the current one"
+            )
 
         if method == "bulk":
             # Find the name of the zipped xml folder
             bulk_download_date = parse_date_string(date)
-            xml_folder_path = os.path.join(self.output_dir, "data", "xml_download")
+            xml_folder_path = os.path.join(self.home_directory, "data", "xml_download")
             os.makedirs(xml_folder_path, exist_ok=True)
             zipped_xml_file_path = os.path.join(
                 xml_folder_path,
@@ -365,3 +385,59 @@ def to_csv(
         # FIXME: Currently metadata is only created for technology data, Fix in #386
         # Configure and save data package metadata file along with data
         # save_metadata(data=technologies_to_export, engine=self.engine)
+
+    def translate(self) -> None:
+        """
+        A database can be translated only once.
+
+        Deletes translated versions of the currently connected database.
+
+        Translates currently connected database,renames it with '-translated'
+        suffix and updates self.engine's path accordingly.
+
+        !!! example
+            ```python
+
+            from open_mastr import Mastr
+            import pandas as pd
+
+            db = Mastr()
+            db.download(data='biomass')
+            db.translate()
+
+            df = pd.read_sql(table='biomass_extended', con=db.engine)
+            print(df.head(10))
+            ```
+
+        """
+
+        if "sqlite" not in self.engine.dialect.name:
+            raise ValueError("engine has to be of type 'sqlite'")
+        if self.is_translated:
+            raise TypeError("the currently connected database is already translated")
+
+        inspector = inspect(self.engine)
+        old_path = r"{}".format(self.engine.url.database)
+        new_path = old_path[:-3] + "-translated.db"
+
+        if os.path.exists(new_path):
+            try:
+                os.remove(new_path)
+            except Exception as e:
+                print(f"An error occurred: {e}")
+
+            print("Replacing previous version of the translated database...")
+
+        for table in inspector.get_table_names():
+            rename_table(table, inspector.get_columns(table), self.engine)
+
+        self.engine.dispose()
+
+        try:
+            os.rename(old_path, new_path)
+            print(f"Database '{old_path}' changed to '{new_path}'")
+        except Exception as e:
+            print(f"An error occurred: {e}")
+
+        self.engine = create_engine(f"sqlite:///{new_path}")
+        self.is_translated = True