pyirk.ruleengine

Created: 2022-09-06 19:14:39 author: Carsten Knoll

This module contains code to enable semantic inferences based on special items (e.g. instances of I41__semantic_rule)

Module Contents

Classes

PremiseType

LiteralWrapper

RuleApplicator

Class to handle the application of a single semantic rule. Deploys several RuleApplicatorWorkers (depending on the OR-subscopes)

RuleApplicatorWorker

Performs the application of one premise branch of a rule

ReportingRuleResult

ReportingMultiRuleResult

AlgorithmicRuleApplicationWorker

This class executes algorithmic rules.

HypothesisReasoner

Functions

apply_all_semantic_rules

Extract all semantic rules and apply them.

apply_semantic_rules

apply_semantic_rule

Create a RuleApplicator instance for the rules, execute its apply-method, return the result (list of new statements)

get_all_rules

filter_relevant_stms

From a list of Statement instances select only those which are qualifiers and whose subject is an RE with .role == SUBJECT. In other words: omit those instances which are created as dual relation edges

edge_matcher

:param e1d: attribute data of edges from “main graph” (see RuleApplicator) :param e2d: attribute data of edges from “prototype graph” (see RuleApplicator)

crpr

custom representation function which accepts Entities and literals

compare_relation_statements

decide whether a given relation fulfills all given statements

is_node_for_simple_graph

exclude nodes which are defined inside certain scopes

get_simple_properties

Data

LITERAL_BASE_URI

VERBOSITY

wildcard_relation_uri

AtlasView

API

pyirk.ruleengine.LITERAL_BASE_URI

‘irk:/tmp/literals’

pyirk.ruleengine.VERBOSITY

False

pyirk.ruleengine.apply_all_semantic_rules(mod_context_uri=None) List[pyirk.core.Statement]

Extract all semantic rules and apply them.

:returns: list of newly created statements

pyirk.ruleengine.apply_semantic_rules(*rules: List, mod_context_uri: str = None) List[pyirk.core.Statement]
pyirk.ruleengine.apply_semantic_rule(rule: pyirk.core.Item, mod_context_uri: str = None) List[pyirk.core.Statement]

Create a RuleApplicator instance for the rules, execute its apply-method, return the result (list of new statements)

pyirk.ruleengine.get_all_rules()
pyirk.ruleengine.filter_relevant_stms(re_list: List[pyirk.core.Statement], return_items=True) List[pyirk.core.Statement]

From a list of Statement instances select only those which are qualifiers and whose subject is an RE with .role == SUBJECT. In other words: omit those instances which are created as dual relation edges

:param re_list: :return:

class pyirk.ruleengine.PremiseType(*args, **kwds)

Bases: enum.Enum

GRAPH

0

SPARQL

1

class pyirk.ruleengine.LiteralWrapper(value)

Initialization

class pyirk.ruleengine.RuleApplicator(rule: pyirk.core.Entity, mod_context_uri: Optional[str] = None)

Class to handle the application of a single semantic rule. Deploys several RuleApplicatorWorkers (depending on the OR-subscopes)

Initialization

get_premise_type() pyirk.ruleengine.PremiseType
extract_premise_stm_lists()
apply() pyirk.core.RuleResult

Application of a semantic rule either in a specified module context or in the currently active module.

(A rule has to be applied in a module context because newly created entities must belong to some module)

_apply() pyirk.core.RuleResult

Perform the actual application of the rule (either via subgraph monomorphism or via SPARQL query)

create_prototypes_for_variable_literals()
create_prototypes_for_fiat_entities()
create_simple_graph() networkx.DiGraph

Create graph without regarding qualifiers. Nodes: uris (of items and relations)

:return:

get_all_node_relations() dict

returns a dict of all graph-relevant relations {(uri1, uri2): [Container(rel_uri=uri3), …], ….}. Keys: 2-tuples of uris. Values: Lists of Containers (due to multi-edges)

_make_literal(value) str

create (if necessary) and return an uri for an literal value

class pyirk.ruleengine.RuleApplicatorWorker(parent: pyirk.ruleengine.RuleApplicator, premise_stms: List[pyirk.core.Statement], premise_items: List[pyirk.core.Item])

Performs the application of one premise branch of a rule

Initialization

max_subgraph_monomorphisms

3000

property rule
apply_sparql_premise() pyirk.core.RuleResult
_resolve_local_node(node=None, uri=None)

return item or literal

_get_understandable_local_nodes()

Generate a humand understandable version of self.local_nodes.a It is intendend for debugging only.

_get_understandable_result_maps(result_maps: List[dict]) List[list]

Generate a human understandable version of the result_map-dicts which are returned by self.match_subgraph_P() It is intendend for debugging only.

apply_graph_premise() pyirk.core.RuleResult
_process_result_map(result_maps) pyirk.core.RuleResult
  • process the found subgraphs with the assertion

get_extended_result_map(result_map: dict) dict

This function create a dict which contains every entry of result_map + mappings from subjectivized predicates.

:param result_map: mapping from P-node to G-node-item (as returned by self.match_subgraph_P)

_get_all_edge_predicate_relations(uri1, uri2, ensure_length1=False)
get_condition_funcs_and_args()
prepare_consequent_functions()

Creates 3 lists: - a list of the consequent functions (might create a new item, new statement or have other side effects) - a list of argument nodes (which will serve as keys for the actual arguments) - a list of the node names

get_asserted_relation_templates() List[Tuple[int, pyirk.core.Relation, int]]

Create a list like [(0, R25, 1), …]

_fill_extended_local_nodes(force=False)

join local_nodes + asserted_nodes + parent.literal_variable_nodes

match_subgraph_P() List[dict]
_get_by_uri(uri)

return literal or entity based on uri

_node_matcher(n1d: dict, n2d: dict) bool

:param n1d: attribute data of node from “main graph” e.g. {‘itm’: <Item I22[“mathematical knowledge artifact”]>} :param n2d: attribute data of node from “prototype graph” e.g. { ‘itm’: {‘R3’: None, ‘R4’: <Item I1[“general item”]>}, ‘entity’: <Item Ia8139[“P1”]> }

:return: boolean matching result

a pair of nodes should match if - n2 is an external entity for self and the uris match - n2 is not an external entity (no further restrictions)

see also: function edge_matcher

static _is_subjectivized_predicate(itm: pyirk.builtin_entities.Item) bool
static _ignore_item(itm: pyirk.builtin_entities.Item) bool
get_premise_type() pyirk.ruleengine.PremiseType
create_prototype_subgraph_from_rule() None

Create a prototype graph from the scopes ‘setting’ and ‘premise’.

_create_psg_nodes() None

Create P and add all nodes (except those for literal values, which will be added adhoc during processing of the statements, see _create_psg_edges)

_create_psg_edges() None
ensure_node_of_P(i)
_get_weakly_connected_components(P) addict.Addict

Get weakly connected components and sort them (separate those which contain only external variables).

Background: external variables are allowed to be disconnected from the rest

During the application of consequent functions unwanted statements might be created. This method serves to delete them. Unwanted statements are e.g. statements about items which are defined in scopes of rules.

pyirk.ruleengine.wildcard_relation_uri

None

pyirk.ruleengine.AtlasView

None

pyirk.ruleengine.edge_matcher(e1d: pyirk.ruleengine.AtlasView, e2d: pyirk.ruleengine.AtlasView) bool

:param e1d: attribute data of edges from “main graph” (see RuleApplicator) :param e2d: attribute data of edges from “prototype graph” (see RuleApplicator)

because we compare MultiDigraphs we get AtlasView-instances, i.e. read-only dicts like AtlasView({0: inner_dict0, 1: inner_dict1, …}). Keys are edge-indices for that ‘multi-edge’, values are like inner_dict0 = { ‘itm1’: <Relation R64[“has scope type”]>, ‘itm2’: <Item I16[“scope”]>, ‘rel_uri’: ‘irk:/builtins#R8’, ‘rel_entity’: <Relation R8[“has domain of argument 1”]> }

:return: boolean matching result

An edge should match if - the relation uri is the same

class pyirk.ruleengine.ReportingRuleResult(raworker: pyirk.ruleengine.RuleApplicatorWorker, raw_result_count: int = None, rule: pyirk.core.Item = None)

Bases: pyirk.core.RuleResult

Initialization

classmethod get_new_instance(res: pyirk.core.RuleResult)

Create an instance of this class based on the instance of a super class.

add_bound_statement(stm: pyirk.core.Statement, raw_binding_info: dict)

:param stm: :param raw_binding_info: dict like {0: <Item Ia1555[“x1”]>, 1: <Item Ia4365[“x2”]>, ‘vlit0’: 42}

_add_statement_report(stm: pyirk.core.Statement, raw_binding_info: dict)
extend_with_binding_info(part: pyirk.core.RuleResult, raw_binding_info: dict)
report(max=None, sep='')
report_str(max=None, sep='')
get_explanation(bindinfo: List[Tuple[pyirk.Entity]])

:param bindinfo: list of 2-tuples like [(<Item Ia7458[“p1”]>, <Item Ia1158[“person1”]>),…]

The bindinfo contains pairs of (, ) tuples. This function uses this information together with the rule-specific explanation_text_template (like “{p1} {rel1} {p2}”) to create strings like: “person1 has_neighbor person2”.

_get_statement_report_count()
_get_stm_container_list()
class pyirk.ruleengine.ReportingMultiRuleResult(rule_list: List[pyirk.core.Item])

Bases: pyirk.ruleengine.ReportingRuleResult

Initialization

property rule
save_html_report(fpath: str, write_file=True, verbose=False)

Create a readable html version of the report :param fpath: str; path of output file

get_report_content()
_get_report_title()
_get_rule_res_list()
pyirk.ruleengine.crpr(obj)

custom representation function which accepts Entities and literals

pyirk.ruleengine.compare_relation_statements(rel1: pyirk.core.Relation, stm_list: List[pyirk.core.Statement], stm_data: addict.Addict = None)

decide whether a given relation fulfills all given statements

pyirk.ruleengine.is_node_for_simple_graph(entity: pyirk.core.Entity) bool

exclude nodes which are defined inside certain scopes

:param item: :return:

pyirk.ruleengine.get_simple_properties(item: pyirk.core.Item) dict
class pyirk.ruleengine.AlgorithmicRuleApplicationWorker

This class executes algorithmic rules.

Initialization

static hardcoded_I810(zb, consequent_function: callable)

Convert 4 negative facts into one positive fact

static hardcoded_I830(zb, consequent_function: callable, *args)

Raise exception if one person is different from more than 4 non-placeholder persons

static hardcoded_I840(zb, consequent_function: callable, *args)
get_single_predicate_report(pred)
get_predicates_report(predicate_list)

Gather data for each relevant predicate how many possibilities of subject-object-pairs exist, which do not contradict an oppo_pred-statement, where oppo_pred is ‘R43__is_opposite_of’ the considered predicate. (“oppo” means opposite)

class pyirk.ruleengine.HypothesisReasoner(zb, base_uri)

Initialization

uri_suffix

‘HYPOTHESIS’

register_module()
hypothesis_reasoning_step(rule_list)