bench_executor.morphkgc

  1#!/usr/bin/env python3
  2
  3"""
  4Morph-KGC is an engine that constructs RDF and RDF-star knowledge graphs
  5from heterogeneous data sources with the R2RML, RML and RML-star mapping
  6languages.
  7
  8**Website**: https://morph-kgc.readthedocs.io/<br>
  9**Repository**: https://github.com/oeg-upm/morph-kgc
 10"""
 11
 12import os
 13import configparser
 14from timeout_decorator import timeout, TimeoutError  # type: ignore
 15from typing import Optional
 16from bench_executor.container import Container
 17from bench_executor.logger import Logger
 18
 19VERSION = '2.2.0'
 20TIMEOUT = 6 * 3600  # 6 hours
 21
 22
 23class MorphKGC(Container):
 24    """Morph-KGC container for executing R2RML, RML, and RML-star mappings."""
 25    def __init__(self, data_path: str, config_path: str, directory: str,
 26                 verbose: bool):
 27        """Creates an instance of the MorphKGC class.
 28
 29        Parameters
 30        ----------
 31        data_path : str
 32            Path to the data directory of the case.
 33        config_path : str
 34            Path to the config directory of the case.
 35        directory : str
 36            Path to the directory to store logs.
 37        verbose : bool
 38            Enable verbose logs.
 39        """
 40        self._data_path = os.path.abspath(data_path)
 41        self._config_path = os.path.abspath(config_path)
 42        self._logger = Logger(__name__, directory, verbose)
 43        os.umask(0)
 44        os.makedirs(os.path.join(self._data_path, 'morphkgc'), exist_ok=True)
 45        super().__init__(f'blindreviewing/morph-kgc:v{VERSION}', 'Morph-KGC',
 46                         self._logger,
 47                         volumes=[f'{self._data_path}/morphkgc:/data',
 48                                  f'{self._data_path}/shared:/data/shared'])
 49
 50    @property
 51    def root_mount_directory(self) -> str:
 52        """Subdirectory in the root directory of the case for Morph-KGC.
 53
 54        Returns
 55        -------
 56        subdirectory : str
 57            Subdirectory of the root directory for Morph-KGC.
 58
 59        """
 60        return __name__.lower()
 61
 62    @timeout(TIMEOUT)
 63    def _execute_with_timeout(self, arguments) -> bool:
 64        """Execute a mapping with a provided timeout.
 65
 66        Returns
 67        -------
 68        success : bool
 69            Whether the execution was successfull or not.
 70        """
 71        cmd = 'python3 -m morph_kgc /data/config_morphkgc.ini'
 72        success = self.run_and_wait_for_exit(cmd)
 73
 74        return success
 75
 76    def execute(self, arguments: list) -> bool:
 77        """Execute Morph-KGC with given arguments.
 78
 79        Parameters
 80        ----------
 81        arguments : list
 82            Arguments to supply to Morph-KGC.
 83
 84        Returns
 85        -------
 86        success : bool
 87            Whether the execution succeeded or not.
 88        """
 89        try:
 90            return self._execute_with_timeout(arguments)
 91        except TimeoutError:
 92            msg = f'Timeout ({TIMEOUT}s) reached for Morph-KGC'
 93            self._logger.warning(msg)
 94
 95        return False
 96
 97    def execute_mapping(self,
 98                        mapping_file: str,
 99                        output_file: str,
100                        serialization: str,
101                        rdb_username: Optional[str] = None,
102                        rdb_password: Optional[str] = None,
103                        rdb_host: Optional[str] = None,
104                        rdb_port: Optional[int] = None,
105                        rdb_name: Optional[str] = None,
106                        rdb_type: Optional[str] = None,
107                        multiple_files: bool = False) -> bool:
108        """Execute a mapping file with Morph-KGC.
109
110        Morph-KGC can transform SQL relational databases (MySQL, PostgreSQL,
111        Oracle, Microsoft SQL Server, MariaDB, SQLite), tabular (CSV, TSV,
112        Excel, Parquet, Feather, ORC, Stata, SAS, SPSS, ODS) and
113        hierarchical files (JSON, XML).
114
115        Morph-KGC currently only supports N-Quads and N-Triples in RDF and
116        RDF-Star. RDF and RDF-Star output is done automatically, you can use
117        the same serialization format (`ntriples`, `nquads`) for both RDF
118        and RDF-Star output.
119
120        Morph-KGC can generate all triples in a single file or spread it
121        among multiple files.
122
123        Parameters
124        ----------
125        mapping_file : str
126            Path to the mapping file to execute.
127        output_file : str
128            Name of the output file to store the triples in.
129        serialization : str
130            Serialization format to use.
131        rdb_username : Optional[str]
132            Username for the database, required when a database is used as
133            source.
134        rdb_password : Optional[str]
135            Password for the database, required when a database is used as
136            source.
137        rdb_host : Optional[str]
138            Hostname for the database, required when a database is used as
139            source.
140        rdb_port : Optional[int]
141            Port for the database, required when a database is used as source.
142        rdb_name : Optional[str]
143            Database name for the database, required when a database is used as
144            source.
145        rdb_type : Optional[str]
146            Database type, required when a database is used as source.
147        multiple_files : bool
148            If the generated triples must be stored in multiple files, default
149            a single file.
150
151        Returns
152        -------
153        success : bool
154            Whether the execution was successfull or not.
155        """
156
157        if serialization == 'nquads':
158            serialization = 'N-QUADS'
159        elif serialization == 'ntriples':
160            serialization = 'N-TRIPLES'
161        else:
162            raise NotImplementedError('Unsupported serialization:'
163                                      f'"{serialization}"')
164
165        # Generate INI configuration file since no CLI is available
166        config = configparser.ConfigParser()
167        config['CONFIGURATION'] = {
168            'output_format': serialization
169        }
170        config['DataSource0'] = {
171            'mappings': f'/data/shared/{mapping_file}'
172        }
173
174        # Morph-KGC can keep the mapping partition results separate, provide
175        # this option, default OFF
176        if multiple_files:
177            config['CONFIGURATION']['output_dir'] = '/data/shared/'
178        else:
179            config['CONFIGURATION']['output_file'] = \
180                    f'/data/shared/{output_file}'
181
182        if rdb_username is not None and rdb_password is not None \
183                and rdb_host is not None and rdb_port is not None \
184                and rdb_name is not None and rdb_type is not None:
185            if rdb_type == 'MySQL':
186                protocol = 'mysql+pymysql'
187            elif rdb_type == 'PostgreSQL':
188                protocol = 'postgresql+psycopg2'
189            else:
190                raise ValueError(f'Unknown RDB type: "{rdb_type}"')
191            rdb_dsn = f'{protocol}://{rdb_username}:{rdb_password}' + \
192                      f'@{rdb_host}:{rdb_port}/{rdb_name}'
193            config['DataSource0']['db_url'] = rdb_dsn
194
195        os.umask(0)
196        os.makedirs(os.path.join(self._data_path, 'morphkgc'), exist_ok=True)
197        path = os.path.join(self._data_path, 'morphkgc', 'config_morphkgc.ini')
198        with open(path, 'w') as f:
199            config.write(f)
200
201        return self.execute([])