#include <ObserverJob.h>

Classes
struct	Node

Public Member Functions
	ObserverJob (const std::shared_ptr< ObserverManager > &manager, const GraphQueryPtr &query, const BindingsHandler &callback)

	~ObserverJob ()

void	stop ()

void	processInsertion (const TripleContainerPtr &triples)

void	processRemoval (const TripleContainerPtr &triples)

auto &	manager () const

	ObserverJob (const std::shared_ptr< ObserverManager > &manager, const GraphQueryPtr &query, const BindingsHandler &callback)

	~ObserverJob ()

void	stop ()

void	processInsertion (const TripleContainerPtr &triples)

void	processRemoval (const TripleContainerPtr &triples)

auto &	manager () const

Protected Types
using	NodeParents = std::vector< std::shared_ptr< Node > >

using	NodeParents = std::vector< std::shared_ptr< Node > >

Protected Member Functions
NodeParents	createGraph (const std::shared_ptr< GraphTerm > &term, const NodeParents &parents)

std::shared_ptr< ObserverJob::Node >	createNode (const std::shared_ptr< GraphPattern > &pattern)

void	initializeNode (const std::shared_ptr< Node > &node)

void	initializeNode (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

GraphQueryPtr	makeQuery (const std::vector< Node * > &reverseSequence)

GraphQueryPtr	makeAtomicQuery (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

bool	matches (const Node &node, const Triple &triple)

void	remove (const std::shared_ptr< Node > &node, const Triple &triple)

void	insert (const std::shared_ptr< Node > &node, const Triple &triple)

void	doInsert (const std::shared_ptr< Node > &node, const BindingsPtr &newBindings)

void	doInsert (const std::shared_ptr< Node > &node, const BindingsPtr &parentBindings, const BindingsPtr &nodeBindings)

BindingsPtr	applyBuiltins (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

NodeParents	createGraph (const std::shared_ptr< GraphTerm > &term, const NodeParents &parents)

std::shared_ptr< ObserverJob::Node >	createNode (const std::shared_ptr< GraphPattern > &pattern)

void	initializeNode (const std::shared_ptr< Node > &node)

void	initializeNode (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

GraphQueryPtr	makeQuery (const std::vector< Node * > &reverseSequence)

GraphQueryPtr	makeAtomicQuery (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

bool	matches (const Node &node, const Triple &triple)

void	remove (const std::shared_ptr< Node > &node, const Triple &triple)

void	insert (const std::shared_ptr< Node > &node, const Triple &triple)

void	doInsert (const std::shared_ptr< Node > &node, const BindingsPtr &newBindings)

void	doInsert (const std::shared_ptr< Node > &node, const BindingsPtr &parentBindings, const BindingsPtr &nodeBindings)

BindingsPtr	applyBuiltins (const std::shared_ptr< Node > &node, const BindingsPtr &bindings)

Protected Attributes
std::shared_ptr< ObserverManager >	manager_

GraphQueryPtr	query_

BindingsHandler	callback_

std::vector< std::shared_ptr< Node > >	nodes_

NodeParents	terminalNodes_

Detailed Description

An observer job is a query that is observed by an observer manager.

Definition at line 19 of file ObserverJob.h.

Member Typedef Documentation

◆ NodeParents [1/2]

using knowrob::ObserverJob::NodeParents = std::vector<std::shared_ptr<Node> >

protected

Definition at line 69 of file ObserverJob.h.

◆ NodeParents [2/2]

using knowrob::ObserverJob::NodeParents = std::vector<std::shared_ptr<Node> >

protected

Definition at line 69 of file ObserverJob.h.

Constructor & Destructor Documentation

◆ ObserverJob() [1/2]

ObserverJob::ObserverJob	(	const std::shared_ptr< ObserverManager > &	manager,
		const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Create an observer job.

Parameters

manager	the observer manager.
query	the query to observe.
callback	the callback to invoke when the query matches.

Definition at line 10 of file ObserverJob.cpp.

         : manager_(manager), query_(query), callback_(callback) {
     // construct the observation graph given the query
     terminalNodes_ = createGraph(query->term(), {});
     // initialize the tables for the query by running a DB query
     // for each node in the observation graph.
     for (const auto &node: nodes_) {
         initializeNode(node);
     }
 }

◆ ~ObserverJob() [1/2]

ObserverJob::~ObserverJob ( )

Definition at line 23 of file ObserverJob.cpp.

                           {
     stop();
 }

◆ ObserverJob() [2/2]

knowrob::ObserverJob::ObserverJob	(	const std::shared_ptr< ObserverManager > &	manager,
		const GraphQueryPtr &	query,
		const BindingsHandler &	callback
	)

Create an observer job.

Parameters

manager	the observer manager.
query	the query to observe.
callback	the callback to invoke when the query matches.

◆ ~ObserverJob() [2/2]

knowrob::ObserverJob::~ObserverJob ( )

Member Function Documentation

◆ applyBuiltins() [1/2]

BindingsPtr ObserverJob::applyBuiltins	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

Definition at line 214 of file ObserverJob.cpp.

                                                                                                    {
     if (node->builtins.empty()) {
         return bindings;
     } else {
         auto bindings_rw = std::make_shared<Bindings>(*bindings);
         for (const auto &builtin: node->builtins) {
             builtin->apply(bindings_rw);
         }
         return bindings_rw;
     }
 }

◆ applyBuiltins() [2/2]

BindingsPtr knowrob::ObserverJob::applyBuiltins	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

◆ createGraph() [1/2]

ObserverJob::NodeParents ObserverJob::createGraph	(	const std::shared_ptr< GraphTerm > &	term,
		const NodeParents &	parents
	)

protected

Definition at line 41 of file ObserverJob.cpp.

                                                                                          {
     switch (term->termType()) {
         case GraphTermType::Pattern: {
             auto node = createNode(std::static_pointer_cast<GraphPattern>(term));
             for (const auto &parent: parents) {
                 node->parents.push_back(parent.get());
                 parent->children.push_back(node);
             }
             return {node};
         }
         case GraphTermType::Builtin: {
             if (!parents.empty()) {
                 for (const auto &parent: parents) {
                     auto builtin = std::static_pointer_cast<GraphBuiltin>(term);
                     parent->builtins.push_back(builtin);
                 }
             } else {
                 KB_WARN("Builtins are not allowed at the top level of an observed query.");
             }
             return parents;
         }
         case GraphTermType::Sequence: {
             auto sequence = std::static_pointer_cast<GraphSequence>(term);
             ObserverJob::NodeParents nextParents = parents;
             for (const auto &subTerm: sequence->terms()) {
                 nextParents = createGraph(subTerm, nextParents);
             }
             return nextParents;
         }
         case GraphTermType::Union: {
             auto unionTerm = std::static_pointer_cast<GraphUnion>(term);
             ObserverJob::NodeParents nextParents;
             for (const auto &subTerm: unionTerm->terms()) {
                 auto subParents = createGraph(subTerm, parents);
                 nextParents.insert(nextParents.end(), subParents.begin(), subParents.end());
             }
             return nextParents;
         }
     }
     return {};
 }

◆ createGraph() [2/2]

NodeParents knowrob::ObserverJob::createGraph	(	const std::shared_ptr< GraphTerm > &	term,
		const NodeParents &	parents
	)

protected

◆ createNode() [1/2]

std::shared_ptr< ObserverJob::Node > ObserverJob::createNode ( const std::shared_ptr< GraphPattern > & pattern )

protected

Definition at line 33 of file ObserverJob.cpp.

                                                                                                  {
     auto node = std::make_shared<Node>();
     node->pattern = pattern;
     nodes_.push_back(node);
     return node;
 }

◆ createNode() [2/2]

std::shared_ptr<ObserverJob::Node> knowrob::ObserverJob::createNode ( const std::shared_ptr< GraphPattern > & pattern )

protected

◆ doInsert() [1/4]

void ObserverJob::doInsert	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	newBindings
	)

protected

Definition at line 309 of file ObserverJob.cpp.

                                                                                           {
     auto solHash = newBindings->hash();
  
     if(node->solutions.find(solHash) != node->solutions.end()) {
         // the solution is already in the node
         return;
     }
     node->solutions[solHash] = newBindings;
  
     if (node->children.empty()) {
         // if the node is terminal, then call the callback with the new bindings.
         callback_(newBindings);
     } else {
         // if the node is not terminal, then a graph query must be constructed for each child node
         //  with the new bindings applied. If there are new solutions, then call insert on the child node
         //  with the new solutions to propagate the new data through the graph.
  
         for (auto &childNode: node->children) {
             auto newQuery = makeAtomicQuery(childNode, newBindings);
             manager_->query(newQuery, [this, childNode, newBindings](const BindingsPtr &bindings) {
                 doInsert(childNode, newBindings, bindings);
             });
         }
     }
 }

◆ doInsert() [2/4]

void knowrob::ObserverJob::doInsert	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	newBindings
	)

protected

◆ doInsert() [3/4]

void ObserverJob::doInsert	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	parentBindings,
		const BindingsPtr &	nodeBindings
	)

protected

Definition at line 300 of file ObserverJob.cpp.

                                              {
     auto newBindings_rw = std::make_shared<Bindings>(*parentBindings);
     newBindings_rw->operator+=(*nodeBindings);
     auto newBindings = applyBuiltins(node, newBindings_rw);
     doInsert(node, newBindings);
 }

◆ doInsert() [4/4]

void knowrob::ObserverJob::doInsert	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	parentBindings,
		const BindingsPtr &	nodeBindings
	)

protected

◆ initializeNode() [1/4]

void ObserverJob::initializeNode ( const std::shared_ptr< Node > & node )

protected

Definition at line 112 of file ObserverJob.cpp.

                                                                       {
     std::vector<GraphQueryPtr> queries;
     std::queue<std::pair<Node *, std::vector<Node *>>> queryQueue;
     queryQueue.push({nodeToInit.get(), { nodeToInit.get() }});
  
     // build queries from the node and its parents
     while (!queryQueue.empty()) {
         auto [node, seq] = queryQueue.front();
         queryQueue.pop();
  
         if (node->parents.empty()) {
             queries.push_back(makeQuery(seq));
         } else {
             for (const auto &parent: node->parents) {
                 std::vector<Node *> newSeq = seq;
                 newSeq.push_back(parent);
                 queryQueue.push({parent, newSeq});
             }
         }
     }
  
     // run queries against the database
     for (const auto &nodeQuery: queries) {
         manager_->query(nodeQuery, [this, nodeToInit](const BindingsPtr &bindings) {
             initializeNode(nodeToInit, bindings);
         });
     }
 }

◆ initializeNode() [2/4]

void knowrob::ObserverJob::initializeNode ( const std::shared_ptr< Node > & node )

protected

◆ initializeNode() [3/4]

void ObserverJob::initializeNode	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

Definition at line 141 of file ObserverJob.cpp.

                                                                                            {
     auto answerHash = answer->hash();
     auto it = node->solutions.find(answerHash);
     if (it == node->solutions.end()) {
         // store the answer in the node
         node->solutions[answerHash] = answer;
         // check if the node is a terminal node
         if (node->children.empty()) {
             // if so, call the callback
             callback_(answer);
         }
     }
 }

◆ initializeNode() [4/4]

void knowrob::ObserverJob::initializeNode	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

◆ insert() [1/2]

void ObserverJob::insert	(	const std::shared_ptr< Node > &	node,
		const Triple &	triple
	)

protected

Definition at line 264 of file ObserverJob.cpp.

                                                                               {
     auto nodePattern = node->pattern->value();
  
     // compute bindings for the triple
     auto tripleBindings = std::make_shared<Bindings>();
     if (nodePattern->subjectTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->subjectTerm());
         tripleBindings->set(var, std::make_shared<IRIAtom>(triple.subject()));
     }
     if (nodePattern->propertyTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->propertyTerm());
         tripleBindings->set(var, std::make_shared<IRIAtom>(triple.predicate()));
     }
     if (nodePattern->objectTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->objectTerm());
         auto o_triple = Atomic::makeTripleValue(triple);
         tripleBindings->set(var, o_triple);
     }
  
     //insert(node, tripleBindings);
     if (node->parents.empty()) {
         // the node is a root node
         doInsert(node, tripleBindings);
     }
     else {
         for (auto &parentNode: node->parents) {
             for (auto &parentSolPair: parentNode->solutions) {
                 auto &parentSolution = parentSolPair.second;
                 if (parentSolution->isConsistentWith(*tripleBindings)) {
                     doInsert(node, parentSolution, tripleBindings);
                 }
             }
         }
     }
 }

◆ insert() [2/2]

void knowrob::ObserverJob::insert	(	const std::shared_ptr< Node > &	node,
		const Triple &	triple
	)

protected

◆ makeAtomicQuery() [1/2]

GraphQueryPtr ObserverJob::makeAtomicQuery	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

Definition at line 98 of file ObserverJob.cpp.

                                                                                                        {
     std::vector<std::shared_ptr<GraphTerm>> terms;
     terms.push_back(applyBindings(node->pattern, *bindings));
     for (const auto &builtin: node->builtins) {
         std::shared_ptr<GraphTerm> builtinTerm = builtin;
         terms.push_back(applyBindings(builtinTerm, *bindings));
     }
     if (terms.size() == 1) {
         return std::make_shared<GraphQuery>(terms[0]);
     } else {
         return std::make_shared<GraphQuery>(std::make_shared<GraphSequence>(terms));
     }
 }

◆ makeAtomicQuery() [2/2]

GraphQueryPtr knowrob::ObserverJob::makeAtomicQuery	(	const std::shared_ptr< Node > &	node,
		const BindingsPtr &	bindings
	)

protected

◆ makeQuery() [1/2]

GraphQueryPtr ObserverJob::makeQuery ( const std::vector< Node * > & reverseSequence )

protected

Definition at line 83 of file ObserverJob.cpp.

                                                                              {
     std::vector<std::shared_ptr<GraphTerm>> terms;
     for (auto it = reverseSequence.rbegin(); it != reverseSequence.rend(); ++it) {
         terms.push_back((*it)->pattern);
         for (const auto &builtin: (*it)->builtins) {
             terms.push_back(builtin);
         }
     }
     if (terms.size() == 1) {
         return std::make_shared<GraphQuery>(terms[0]);
     } else {
         return std::make_shared<GraphQuery>(std::make_shared<GraphSequence>(terms));
     }
 }

◆ makeQuery() [2/2]

GraphQueryPtr knowrob::ObserverJob::makeQuery ( const std::vector< Node * > & reverseSequence )

protected

◆ manager() [1/2]

auto& knowrob::ObserverJob::manager ( ) const

inline

Returns: the observer manager.

Definition at line 53 of file ObserverJob.h.

53 { return manager_; }

◆ manager() [2/2]

auto& knowrob::ObserverJob::manager ( ) const

inline

Returns: the observer manager.

Definition at line 53 of file ObserverJob.h.

53 { return manager_; }

◆ matches() [1/2]

bool ObserverJob::matches	(	const Node &	node,
		const Triple &	triple
	)

protected

Definition at line 155 of file ObserverJob.cpp.

                                                                 {
     auto &pat = node.pattern->value();
  
     auto &s_pat = pat->subjectTerm();
     if (s_pat->isAtomic()) {
         auto s_pat_atom = std::static_pointer_cast<Atomic>(s_pat);
         if (s_pat_atom->stringForm() != triple.subject()) return false;
     } else if (!s_pat->isVariable()) {
         return false;
     }
  
     auto &p_pat = pat->propertyTerm();
     if (p_pat->isAtomic()) {
         auto p_pat_atom = std::static_pointer_cast<Atomic>(p_pat);
         if (p_pat_atom->stringForm() != triple.predicate()) return false;
     } else if (!p_pat->isVariable()) {
         return false;
     }
  
     auto &o_pat = pat->objectTerm();
     if (o_pat->isAtomic()) {
         auto o_pat_atom = std::static_pointer_cast<Atomic>(o_pat);
  
         if (triple.isObjectIRI() || triple.isObjectBlank()) {
             auto o_triple = triple.valueAsString();
             if (o_pat_atom->stringForm() != o_triple) return false;
         } else {
             auto o_triple = Atomic::makeTripleValue(triple);
             if (*o_pat_atom != *o_triple) return false;
         }
     } else if (!o_pat->isVariable()) {
         return false;
     }
  
     return true;
 }

◆ matches() [2/2]

bool knowrob::ObserverJob::matches	(	const Node &	node,
		const Triple &	triple
	)

protected

◆ processInsertion() [1/2]

void ObserverJob::processInsertion ( const TripleContainerPtr & triples )

Process the insertion of triples.

Parameters

triples the triples to insert.

Definition at line 192 of file ObserverJob.cpp.

                                                                     {
     for (auto &triplePtr: *triples) {
         auto &triple = *triplePtr;
         for (auto &node: nodes_) {
             if (matches(*node, triple)) {
                 insert(node, triple);
             }
         }
     }
 }

◆ processInsertion() [2/2]

void knowrob::ObserverJob::processInsertion ( const TripleContainerPtr & triples )

Process the insertion of triples.

Parameters

triples the triples to insert.

◆ processRemoval() [1/2]

void ObserverJob::processRemoval ( const TripleContainerPtr & triples )

Process the removal of triples.

Parameters

triples the triples to remove.

Definition at line 203 of file ObserverJob.cpp.

                                                                   {
     for (auto &triplePtr: *triples) {
         auto &triple = *triplePtr;
         for (auto &node: nodes_) {
             if (matches(*node, triple)) {
                 remove(node, triple);
             }
         }
     }
 }

◆ processRemoval() [2/2]

void knowrob::ObserverJob::processRemoval ( const TripleContainerPtr & triples )

Process the removal of triples.

Parameters

triples the triples to remove.

◆ remove() [1/2]

void ObserverJob::remove	(	const std::shared_ptr< Node > &	node,
		const Triple &	triple
	)

protected

Definition at line 226 of file ObserverJob.cpp.

                                                                               {
     auto nodePattern = node->pattern->value();
  
     // compute bindings for the triple
     auto tripleBindings = std::make_shared<Bindings>();
     if (nodePattern->subjectTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->subjectTerm());
         tripleBindings->set(var, std::make_shared<IRIAtom>(triple.subject()));
     }
     if (nodePattern->propertyTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->propertyTerm());
         tripleBindings->set(var, std::make_shared<IRIAtom>(triple.predicate()));
     }
     if (nodePattern->objectTerm()->isVariable()) {
         auto var = std::static_pointer_cast<Variable>(nodePattern->objectTerm());
         auto o_triple = Atomic::makeTripleValue(triple);
         tripleBindings->set(var, o_triple);
     }
  
     // remove all solutions of this node or children nodes
     // that are consistent with the removed triple
     std::queue<Node*> nodeQueue;
     nodeQueue.push(node.get());
     while (!nodeQueue.empty()) {
         auto nextNode = nodeQueue.front();
         nodeQueue.pop();
         for (auto &solPair: nextNode->solutions) {
             auto &solution = solPair.second;
             if (solution->isConsistentWith(*tripleBindings)) {
                 nextNode->solutions.erase(solPair.first);
             }
         }
         for (auto &childNode: nextNode->children) {
             nodeQueue.push(childNode.get());
         }
     }
 }

◆ remove() [2/2]

void knowrob::ObserverJob::remove	(	const std::shared_ptr< Node > &	node,
		const Triple &	triple
	)

protected

◆ stop() [1/2]

void ObserverJob::stop ( )

Stop the observer job.

Definition at line 27 of file ObserverJob.cpp.

                        {
     nodes_.clear();
     terminalNodes_.clear();
     manager_ = nullptr;
 }

◆ stop() [2/2]

void knowrob::ObserverJob::stop ( )

Stop the observer job.

Member Data Documentation

◆ callback_

BindingsHandler knowrob::ObserverJob::callback_

protected

Definition at line 58 of file ObserverJob.h.

◆ manager_

std::shared_ptr< ObserverManager > knowrob::ObserverJob::manager_

protected

Definition at line 56 of file ObserverJob.h.

◆ nodes_

std::vector< std::shared_ptr< Node > > knowrob::ObserverJob::nodes_

protected

Definition at line 67 of file ObserverJob.h.

◆ query_

GraphQueryPtr knowrob::ObserverJob::query_

protected

Definition at line 57 of file ObserverJob.h.

◆ terminalNodes_

NodeParents knowrob::ObserverJob::terminalNodes_

protected

Definition at line 70 of file ObserverJob.h.

The documentation for this class was generated from the following files:

build/_CPack_Packages/Linux/DEB/knowrob-2.1.0-Linux/usr/include/knowrob/storage/ObserverJob.h
src/storage/ObserverJob.cpp

Classes

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Detailed Description

Member Typedef Documentation

◆ NodeParents [1/2]

◆ NodeParents [2/2]

Constructor & Destructor Documentation

◆ ObserverJob() [1/2]

◆ ~ObserverJob() [1/2]

◆ ObserverJob() [2/2]

◆ ~ObserverJob() [2/2]

Member Function Documentation

◆ applyBuiltins() [1/2]

◆ applyBuiltins() [2/2]

◆ createGraph() [1/2]

◆ createGraph() [2/2]

◆ createNode() [1/2]

◆ createNode() [2/2]

◆ doInsert() [1/4]

◆ doInsert() [2/4]

◆ doInsert() [3/4]

◆ doInsert() [4/4]

◆ initializeNode() [1/4]

◆ initializeNode() [2/4]

◆ initializeNode() [3/4]

◆ initializeNode() [4/4]

◆ insert() [1/2]

◆ insert() [2/2]

◆ makeAtomicQuery() [1/2]

◆ makeAtomicQuery() [2/2]

◆ makeQuery() [1/2]

◆ makeQuery() [2/2]

◆ manager() [1/2]

◆ manager() [2/2]

◆ matches() [1/2]

◆ matches() [2/2]

◆ processInsertion() [1/2]

◆ processInsertion() [2/2]

◆ processRemoval() [1/2]

◆ processRemoval() [2/2]

◆ remove() [1/2]

◆ remove() [2/2]

◆ stop() [1/2]

◆ stop() [2/2]

Member Data Documentation

◆ callback_

◆ manager_

◆ nodes_

◆ query_

◆ terminalNodes_