test_rule_atomic_extractor.py

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

#==============================================================================
# TENET: Extraction Rule Test
#------------------------------------------------------------------------------
# Script to test rules under development
#==============================================================================

import subprocess, os
from rdflib import Graph, Namespace
from rdflib.namespace import NamespaceManager, FOAF, RDF
from rdflib import URIRef, Literal, BNode

FILE_PATH = f'{os.path.dirname(os.path.abspath(__file__))}'
INPUT_DIR_PATH = f'{FILE_PATH}/test_data/'
OUTPUT_DIR_PATH = f'{FILE_PATH}/test_data/'

TEST_FILE_NAME = 'atom-extraction-devGraph-1'
INPUT_GRAPH_PATH = f'{INPUT_DIR_PATH}{TEST_FILE_NAME}.ttl'
OUTPUT_GRAPH_PATH = f'{OUTPUT_DIR_PATH}{TEST_FILE_NAME}.result.ttl'
OUTPUT_GRAPH_URI = f'https://amr.tetras-libre.fr/rdf/{TEST_FILE_NAME}/result'

from context import tenet
from tenet.transduction.rdfterm_computer import __update_uri_with_prefix
from tenet.transduction import rdfterm_computer, prefix_handle
from tenet.transduction import net
from tenet.scheme.amr_rule.transduction import atom_class_extractor as rule_1
from tenet.scheme.amr_rule.transduction import atom_individual_extractor as rule_2
from tenet.scheme.amr_rule.transduction import atom_property_extractor as rule_3
from tenet.scheme.amr_rule.transduction import atom_value_extractor as rule_4
from tenet.scheme.amr_rule.transduction import atom_phenomena_extractor as rule_5
from tenet.scheme import amr_rule



#==============================================================================
# Useful Methods
#==============================================================================

def load_test_graph():
    print(f'\n -- Test Graph Loading')
    graph = Graph()
    prefix_handle.update_graph_namespacemanager(graph)
    graph.parse(INPUT_GRAPH_PATH)
    print(f" ----- Graph Loaded ({len(graph)})")
    return graph


def print_triple(triple, num=-1):
    num_str = f'[{num}]' if num > -1 else '[-]'
    (s, p, o) = triple
    s = __update_uri_with_prefix(graph, s)
    p = __update_uri_with_prefix(graph, p)
    o = __update_uri_with_prefix(graph, o)
    print(f' {num_str} {s} {p} {o}')
    

def add_triples_in_graph(graph, triple_list):
    print(f'\n -- Adding triple(s) in graph')       
    print(f" ----- Graph length before update: {len(graph)}")
    print(f" ----- Number of triples to add: {len(triple_list)}")
    
    print(f" ----- Added triples:")
    n = 0
    for triple in triple_list:
        n += 1
        print_triple(triple, num=n)
        graph.add(triple)      
        
    print(f" ----- Graph length after update: {len(graph)}")
    
    print(f'\n -- Serialize test graph to {OUTPUT_GRAPH_PATH}')
    graph.serialize(destination=OUTPUT_GRAPH_PATH, 
                    format='turtle',
                    base=OUTPUT_GRAPH_URI)



#==============================================================================
# Development Test
#==============================================================================
        
def test_search_pattern_1(graph):
    _, pattern_set = rule_1.__search_pattern(graph)
    print(f'\n ----- number of selection found: {len(pattern_set)}')
    for row in pattern_set: 
        result_str = f'>>> '
        result_str += f'{row.leaf.n3(graph.namespace_manager)}'
        result_str += f' {row.conceptName.n3(graph.namespace_manager)}'
        print(result_str) 
    return pattern_set

        
def test_search_pattern_2(graph):
    _, pattern_set = rule_2.__search_pattern(graph)
    print(f'\n ----- number of selection found: {len(pattern_set)}')
    for row in pattern_set: 
        result_str = f'>>> '
        result_str += f'{row.baseLeaf.n3(graph.namespace_manager)}'
        result_str += f' {row.valueLabel.n3(graph.namespace_manager)}'
        result_str += f' {row.classNet.n3(graph.namespace_manager)}'
        print(result_str) 
    return pattern_set

        
def test_search_pattern_3(graph):
    _, pattern_set = rule_3.__search_pattern(graph)
    print(f'\n ----- number of selection found: {len(pattern_set)}')
    for row in pattern_set: 
        result_str = f'>>> '
        result_str += f'{row.baseLeaf.n3(graph.namespace_manager)}'
        result_str += f' {row.predicateLabel.n3(graph.namespace_manager)}'
        print(result_str) 
    return pattern_set

        
def test_search_pattern_4(graph):
    _, pattern_set = rule_4.__search_pattern(graph)
    print(f'\n ----- number of selection found: {len(pattern_set)}')
    for row in pattern_set: 
        result_str = f'>>> '
        result_str += f'{row.valueLabel.n3(graph.namespace_manager)}'
        print(result_str) 
    return pattern_set

        
def test_search_pattern_5(graph):
    _, pattern_set = rule_5.__search_pattern(graph)
    print(f'\n ----- number of selection found: {len(pattern_set)}')
    for row in pattern_set: 
        result_str = f'>>> '
        result_str += f'{row.baseLeaf.n3(graph.namespace_manager)}'
        result_str += f' {row.phenomenaLabel.n3(graph.namespace_manager)}'
        result_str += f' {row.relationConceptPhenomena.n3(graph.namespace_manager)}'
        result_str += f' {row.relationConceptLabel.n3(graph.namespace_manager)}'
        print(result_str) 
    return pattern_set

        
def test_search_structure(graph):
    result_set = rule_3.__search_structure(graph)
    print(f'\n ----- number of selection found: {len(result_set)}')
    for row in result_set: 
        result_str = f'>>> '
        result_str += f'{row.sentenceId.n3(graph.namespace_manager)}'
        print(result_str) 
    return result_set
    

#==============================================================================
# Unit Test
#==============================================================================

def test_rule_application(graph, rule):    
    print('\n -- Rule Test')
    
    rule_label, new_triple_list = rule(graph)
    print(f' ----- label: {rule_label}')
    print(f' ----- new_triple_list ({len(new_triple_list)}):')
    
    add_triples_in_graph(graph, new_triple_list)
    


#==============================================================================
# Test Script
#==============================================================================

if __name__ == '__main__':
      
    print('\n *** Test Preparation ***')
    graph = load_test_graph()
    print('\n \n')
    
    print('\n ///////////////////// Extraction Rule 1')
    print('\n *** Step Test ***')
    
    print('\n -- Step 1: Search Pattern')
    pattern_set = test_search_pattern_1(graph)
    test_search_structure(graph)
    
    print('\n \n')
    
    print('\n *** Unit Test ***')
    test_rule_application(graph, rule_1.extract_atom_class)
    print('\n \n')
    
    print('\n ///////////////////// Extraction Rule 2')
    print('\n *** Step Test ***')
    
    print('\n -- Step 1: Search Pattern')
    pattern_set = test_search_pattern_2(graph)
    
    print('\n \n')
    
    print('\n *** Unit Test ***')
    test_rule_application(graph, rule_2.extract_atom_individual)
    print('\n \n')
    print('\n \n')
    
    print('\n ///////////////////// Extraction Rule 3')
    print('\n *** Step Test ***')
    
    print('\n -- Step 1: Search Pattern')
    pattern_set = test_search_pattern_3(graph)
    
    print('\n \n')
    
    print('\n *** Unit Test ***')
    test_rule_application(graph, rule_3.extract_atom_property)
    print('\n \n')
    
    print('\n ///////////////////// Extraction Rule 4')
    print('\n *** Step Test ***')
    
    print('\n -- Step 1: Search Pattern')
    pattern_set = test_search_pattern_4(graph)
    
    print('\n \n')
    
    print('\n *** Unit Test ***')
    test_rule_application(graph, rule_4.extract_atom_value)
    print('\n \n')
    
    print('\n ///////////////////// Extraction Rule 5')
    print('\n *** Step Test ***')
    
    print('\n -- Step 1: Search Pattern')
    pattern_set = test_search_pattern_5(graph)
    
    print('\n \n')
    
    print('\n *** Unit Test ***')
    test_rule_application(graph, rule_5.extract_atom_phenomena)
    print('\n \n')

    print('\n *** - ***')