Skip to content

Issue 135: Rename and document function and internals to clarify InstantaneousEvent vs. Influence #136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vulnmaster merged 1 commit into from
Feb 20, 2026
Merged

Issue 135: Rename and document function and internals to clarify InstantaneousEvent vs. Influence #136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 23 additions & 23 deletions case_prov/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ def interval_end_should_exist(
return None


def infer_prov_instantaneous_influence_event(
def infer_prov_influence(
in_graph: rdflib.Graph,
n_prov_thing: rdflib.term.IdentifiedNode,
n_predicate: rdflib.URIRef,
Expand All @@ -213,7 +213,9 @@ def infer_prov_instantaneous_influence_event(
**kwargs: typing.Any,
) -> typing.Tuple[rdflib.term.IdentifiedNode, TmpTriplesType]:
"""
PROV InstantaneousEvents that are also Influences need to be defined inhering in the two things related by the unqualified property.
The PROV InstantaneousEvents that are also Influences need to be defined inhering in the two things related by the unqualified property.

This handles qualifying influences of a form "Given 'x u y', 'u' an unqualified influence property, qualify the influence as 'x q I , I p y'." E.g., 'x prov:wasGeneratedBy y' -> 'x prov:qualifiedGeneration G; G a prov:Generation; G prov:activity y .'" Other influences reverse direction or involve multiple links, and are handled with logic not in this function.

:returns: Returns a node N matching the pattern 'n_prov_thing n_predicate N', as well as a supplemental set of triples. If a node N is not found in the graph, a node is created and linked in the supplemental triples; hence the length of the supplemental triples being >0 can be used as an indicator that the node was created.
"""
Expand All @@ -226,13 +228,13 @@ def infer_prov_instantaneous_influence_event(
}[n_predicate]

ret_triples: TmpTriplesType = set()
n_instantaneous_event: typing.Optional[rdflib.IdentifiedNode] = None
n_influence: typing.Optional[rdflib.IdentifiedNode] = None
for n_value in in_graph.objects(n_prov_thing, n_predicate):
assert isinstance(n_value, rdflib.term.IdentifiedNode)
n_instantaneous_event = n_value
n_influence = n_value
break
if n_instantaneous_event is None:
# Define event node.
if n_influence is None:
# Define influence node.
if isinstance(n_prov_thing, rdflib.URIRef) and isinstance(
n_prov_related_thing, rdflib.URIRef
):
Expand All @@ -248,32 +250,30 @@ def infer_prov_instantaneous_influence_event(
)
else:
node_uuid = local_uuid()
n_instantaneous_event = rdf_namespace[slug + node_uuid]
n_influence = rdf_namespace[slug + node_uuid]
else:
n_instantaneous_event = rdflib.BNode()
# Link event node.
ret_triples.add((n_prov_thing, n_predicate, n_instantaneous_event))
# Type event node.
n_instantaneous_event_type = {
n_influence = rdflib.BNode()
# Link influence node.
ret_triples.add((n_prov_thing, n_predicate, n_influence))
# Type influence node.
n_influence_type = {
NS_PROV.qualifiedCommunication: NS_PROV.Communication,
NS_PROV.qualifiedDerivation: NS_PROV.Derivation,
NS_PROV.qualifiedGeneration: NS_PROV.Generation,
NS_PROV.qualifiedInvalidation: NS_PROV.Invalidation,
NS_PROV.qualifiedUsage: NS_PROV.Usage,
}[n_predicate]
ret_triples.add(
(n_instantaneous_event, NS_RDF.type, n_instantaneous_event_type)
)
# Port timestamp to event node.
if n_instantaneous_event_type == NS_PROV.Generation:
ret_triples.add((n_influence, NS_RDF.type, n_influence_type))
# Port timestamp to influence node if also an instantaneous event node.
if n_influence_type == NS_PROV.Generation:
for l_object in in_graph.objects(n_prov_thing, NS_PROV.generatedAtTime):
assert isinstance(l_object, rdflib.Literal)
ret_triples.add((n_instantaneous_event, NS_PROV.atTime, l_object))
elif n_instantaneous_event_type == NS_PROV.Invalidation:
ret_triples.add((n_influence, NS_PROV.atTime, l_object))
elif n_influence_type == NS_PROV.Invalidation:
for l_object in in_graph.objects(n_prov_thing, NS_PROV.invalidatedAtTime):
assert isinstance(l_object, rdflib.Literal)
ret_triples.add((n_instantaneous_event, NS_PROV.atTime, l_object))
# Link provenentially-tied node to event node.
ret_triples.add((n_influence, NS_PROV.atTime, l_object))
# Link provenentially-tied node to influence node.
n_inherent_influence_predicate = {
NS_PROV.qualifiedCommunication: NS_PROV.activity,
NS_PROV.qualifiedDerivation: NS_PROV.entity,
Expand All @@ -283,12 +283,12 @@ def infer_prov_instantaneous_influence_event(
}[n_predicate]
ret_triples.add(
(
n_instantaneous_event,
n_influence,
n_inherent_influence_predicate,
n_prov_related_thing,
)
)
return (n_instantaneous_event, ret_triples)
return (n_influence, ret_triples)


def infer_interval_terminus(
Expand Down
12 changes: 6 additions & 6 deletions case_prov/case_prov_rdf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,7 @@ def main() -> None:

def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None:
"""
This subroutine is provided to supplement case_prov.infer_prov_instantaneous_influence_event usage.
This subroutine is provided to supplement case_prov.infer_prov_influence usage.
"""
nonlocal tmp_triples
for inference_triple in inference_triples:
Expand All @@ -360,7 +360,7 @@ def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None
(
n_communication,
inference_triples,
) = case_prov.infer_prov_instantaneous_influence_event(
) = case_prov.infer_prov_influence(
tmp_graph,
n_informed_activity,
NS_PROV.qualifiedCommunication,
Expand All @@ -385,7 +385,7 @@ def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None
(
n_derivation,
inference_triples,
) = case_prov.infer_prov_instantaneous_influence_event(
) = case_prov.infer_prov_influence(
tmp_graph,
n_action_result,
NS_PROV.qualifiedDerivation,
Expand Down Expand Up @@ -418,7 +418,7 @@ def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None
(
n_generation,
inference_triples,
) = case_prov.infer_prov_instantaneous_influence_event(
) = case_prov.infer_prov_influence(
tmp_graph,
n_entity,
NS_PROV.qualifiedGeneration,
Expand All @@ -443,7 +443,7 @@ def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None
(
n_invalidation,
inference_triples,
) = case_prov.infer_prov_instantaneous_influence_event(
) = case_prov.infer_prov_influence(
tmp_graph,
n_entity,
NS_PROV.qualifiedInvalidation,
Expand All @@ -468,7 +468,7 @@ def _pull_inference_triples(inference_triples: case_prov.TmpTriplesType) -> None
(
n_usage,
inference_triples,
) = case_prov.infer_prov_instantaneous_influence_event(
) = case_prov.infer_prov_influence(
tmp_graph,
n_activity,
NS_PROV.qualifiedUsage,
Expand Down