old_rule.py

#!/usr/bin/python3.10
# -*-coding:Utf-8 -*

#==============================================================================
# TENET: Transduction Rule
#------------------------------------------------------------------------------
# Class to define a Compositional Transduction Rule (CTR)
#==============================================================================

#==============================================================================
# Importing required modules
#==============================================================================

from utility.timer import timer_return


#==============================================================================
# Global Variable
#==============================================================================

PREFIX_PATTERN = 'PREFIX {0}: {1}'

QUERY_PATTERN = """{0}
    CONSTRUCT {{
        {1}
    }}
    WHERE {{
        {2}
        {3}
    }}
    """

RULE_STRING = """ *** Compositional Transduction Rule (CTR) ***
-- label: {0}
-- comment: {1}
-- prefix_list: {2}
-- clauses: {3}
-- binding: {4}
-- construction: {5}
    """


#==============================================================================
# Class
#==============================================================================

class OldRule:
    """ Class to define a Compositional Transduction Rule (CTR).
    """
    
    #--------------------------------------------------------------------------
    # Constructor(s)
    #--------------------------------------------------------------------------
           
    def __init__(self, label='', comment='', prefix_list=[], 
                 clause='', binding='', construction=''):
        self.label = label
        self.comment = comment
        self.prefix_list = prefix_list
        self.clause = clause
        self.binding = binding
        self.construction = construction


    #--------------------------------------------------------------------------
    # Loading Method(s)
    #--------------------------------------------------------------------------
        
    def load_dict(self, rule_def):
        for attr in self.__dict__:
            if attr in rule_def.keys():
                self.__dict__[attr] = rule_def[attr] 
                
    def load_prefix_list(self, prefix_list):
        self.prefix_list = prefix_list
        
    
    #--------------------------------------------------------------------------
    # Accessor(s)
    #--------------------------------------------------------------------------    
    
    # NA

    
    #--------------------------------------------------------------------------
    # Base Method(s)
    #--------------------------------------------------------------------------

    def __str__(self):
        return RULE_STRING.format(self.label,
                                  self.comment,
                                  self.prefix_list,
                                  self.clause,
                                  self.binding,
                                  self.construction
                                  )
    
    
    #--------------------------------------------------------------------------
    # Method(s) to get the SPARQL query corresponding to the rule
    #--------------------------------------------------------------------------
    
    def get_prefix_def(self):
        prefix_def = ''
        for (ident, iri) in self.prefix_list:
            prefix_def += PREFIX_PATTERN.format(ident, iri) + '\n'
        return prefix_def
        
    def get_query(self):
        prefix_def = self.get_prefix_def()
        query = QUERY_PATTERN.format(prefix_def,
                                     self.construction,
                                     self.clause,
                                     self.binding)
        return query
  
    
    #--------------------------------------------------------------------------
    # Method(s) to update a graph by running the rule SPARQL query
    #--------------------------------------------------------------------------
    
    @timer_return
    def apply(self, graph):
        """ Update the <graph> by running the rule SPARQL query, and
            return resulting triples with the updated graph
        """
        
        try:
            sparql_query = self.get_query()
            result_triple_set = graph.query(sparql_query)
            
            for triple in result_triple_set:
                graph.add(triple) 
            
            return graph, result_triple_set
        
        except:
            logger.error(' *** Error while applying rule (rule.apply(graph)) *** ')  
            logger.debug(' ----- len(graph): {0}'.format(len(graph))) 
            logger.debug(' ----- sparql_query: {0}'.format(sparql_query))   
            logger.debug(' ----- len(result_triple_set): {0}'.format(len(result_triple_set))) 
            logger.debug(' ----- last triple: {0}'.format(triple)) 
  
    
#==============================================================================
# Development Test
#==============================================================================
    
if __name__ == '__main__':
    
    print('\n' + ' *** Development Test ***')
    
    print('\n' + ' -- Test Data:')
    print(' ----- rule_def_1: ')
    rule_def_1 = {
        'label': 'test-label',
        'comment': 'a simple comment for test',
        'clause': '?x a nsp:test.',
        'construction': 'nsp:new-x a nsp:test.'}
    print(rule_def_1)
    print(' ----- prefix_list_1: ')
    prefix_list_1 = []
    prefix_list_1.append(('owl', '<http://www.w3.org/2002/07/owl#>'))
    prefix_list_1.append(('rdf', '<http://www.w3.org/1999/02/22-rdf-syntax-ns#>'))
    print(prefix_list_1)
    
    print('\n' + ' -- construct an empty rule')
    rule = Rule()
    print(rule)
    
    print('\n' + ' -- load a rule definition from a dictionary')
    rule.load_dict(rule_def_1)
    print(rule)
    
    print('\n' + ' -- load a prefix list')
    rule.load_prefix_list(prefix_list_1)
    print(rule)
    
    print('\n' + ' -- get the SPARQL query')
    query = rule.get_query()
    print(query)
        
    print('\n' + ' *** - ***')