diff --git a/trustgraph-flow/trustgraph/cores/knowledge.py b/trustgraph-flow/trustgraph/cores/knowledge.py index 82224d71..b1798af0 100644 --- a/trustgraph-flow/trustgraph/cores/knowledge.py +++ b/trustgraph-flow/trustgraph/cores/knowledge.py @@ -20,7 +20,7 @@ class KnowledgeManager: async def delete_kg_core(self, request): - print("Updating doc...") + print("Updating doc...", flush=True) # You can't update the document ID, user or kind. @@ -42,50 +42,69 @@ class KnowledgeManager: processing_metadatas = None, ) - async def fetch_kg_core(self, request, flow): + async def fetch_kg_core(self, request, respond): - print("Fetch core...") + print("Fetch core...", flush=True) - async def publish(obj): - print(obj) -# await publisher.send(triples) + async def publish_triples(t): + await respond( + KnowledgeResponse( + error = None, + ids = None, + eos = False, + triples = t, + graph_embeddings = None, + ) + ) # Remove doc table row await self.table_store.get_triples( request.user, request.id, - publish + publish_triples, ) - async def publish_ge(obj): - print(obj) + async def publish_ge(g): + await respond( + KnowledgeResponse( + error = None, + ids = None, + eos = False, + triples = None, + graph_embeddings = g, + ) + ) # Remove doc table row await self.table_store.get_graph_embeddings( request.user, request.id, - publish + publish_ge, ) print("Fetch complete", flush=True) - return KnowledgeResponse( - error = None, - ids = None, - eos = True, - triples = None, - graph_embeddings = None, + await respond( + KnowledgeResponse( + error = None, + ids = None, + eos = True, + triples = None, + graph_embeddings = None, + ) ) - async def list_kg_cores(self, request, flow): + async def list_kg_cores(self, request, respond): ids = await self.table_store.list_kg_cores(request.user) - return KnowledgeResponse( - error = None, - ids = ids, - eos = False, - triples = None, - graph_embeddings = None + await respond( + KnowledgeResponse( + error = None, + ids = ids, + eos = False, + triples = None, + graph_embeddings = None + ) ) diff --git a/trustgraph-flow/trustgraph/cores/service.py b/trustgraph-flow/trustgraph/cores/service.py index eca6ceb3..93e84d1e 100755 --- a/trustgraph-flow/trustgraph/cores/service.py +++ b/trustgraph-flow/trustgraph/cores/service.py @@ -116,72 +116,7 @@ class Processor(AsyncProcessor): print(self.flows) - def __del__(self): - - pass - - async def load_document(self, document, processing, content): - - print("Ready for processing...") - - print(document, processing, len(content)) - - if processing.flow not in self.flows: - raise RuntimeError("Invalid flow ID") - - flow = self.flows[processing.flow] - - if document.kind == "text/plain": - kind = "text-load" - elif document.kind == "application/pdf": - kind = "document-load" - else: - raise RuntimeError("Document with a MIME type I don't know") - - q = flow["interfaces"][kind] - - if kind == "text-load": - doc = TextDocument( - metadata = Metadata( - id = document.id, - metadata = document.metadata, - user = processing.user, - collection = processing.collection - ), - text = content, - ) - schema = TextDocument - else: - doc = Document( - metadata = Metadata( - id = document.id, - metadata = document.metadata, - user = processing.user, - collection = processing.collection - ), - data = base64.b64encode(content).decode("utf-8") - - ) - schema = Document - - print(f"Submit on queue {q}...") - - pub = Publisher( - self.pulsar_client, q, schema=schema - ) - - await pub.start() - - # FIXME: Time wait kludge? - await asyncio.sleep(1) - - await pub.send(None, doc) - - await pub.stop() - - print("Document submitted") - - async def process_request(self, v, flow): + async def process_request(self, v, id): if v.operation is None: raise RequestError("Null operation") @@ -197,7 +132,11 @@ class Processor(AsyncProcessor): if v.operation not in impls: raise RequestError(f"Invalid operation: {v.operation}") - return await impls[v.operation](v, flow) + async def respond(x): + await self.knowledge_response_producer.send( + x, { "id": id } + ) + return await impls[v.operation](v, respond) async def on_knowledge_request(self, msg, consumer, flow): @@ -211,11 +150,11 @@ class Processor(AsyncProcessor): try: - resp = await self.process_request(v, flow) + await self.process_request(v, id) - await self.knowledge_response_producer.send( - resp, properties={"id": id} - ) +# await self.knowledge_response_producer.send( +# resp, properties={"id": id} +# ) return diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/knowledge.py b/trustgraph-flow/trustgraph/gateway/dispatch/knowledge.py index 3369b4e3..5e099a68 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/knowledge.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/knowledge.py @@ -34,22 +34,34 @@ class KnowledgeRequestor(ServiceRequestor): def from_response(self, message): - print(message) - - response = { - "eos": message.eos - } + print("Processing message") + # Response to list, if message.ids is not None: - response["ids"] = message.ids + print("-> IDS") + return { + "ids": message.ids + }, True if message.triples: - response["triples"] = serialize_triples(message.triples) + print("-> triples") + return { + "triples": serialize_triples(message.triples) + }, False if message.graph_embeddings: - response["graph-embeddings"] = serialize_graph_embeddings( - message.graph_embeddings - ) - - return response, True + print("-> ge") + return { + "graph-embeddings": serialize_graph_embeddings( + message.graph_embeddings + ) + }, False + + if message.eos is True: + print("-> eos") + return { + "eos": True + }, True + + raise RuntimeError("Unexpected case") diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py b/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py index 1ce5ac68..b8a84644 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/requestor.py @@ -82,7 +82,7 @@ class ServiceRequestor: resp, fin = self.from_response(resp) - print(resp, fin) + print(resp, fin, flush=True) if responder: await responder(resp, fin) diff --git a/trustgraph-flow/trustgraph/tables/knowledge.py b/trustgraph-flow/trustgraph/tables/knowledge.py index 1dbb4fd5..50fd9dce 100644 --- a/trustgraph-flow/trustgraph/tables/knowledge.py +++ b/trustgraph-flow/trustgraph/tables/knowledge.py @@ -351,17 +351,10 @@ class KnowledgeTableStore: try: - print("Executing!") - print(self.get_triples_stmt) - print(user, document_id) - resp = self.cassandra.execute( self.get_triples_stmt, (user, document_id) ) - print("HERE") - - print("Executed") break @@ -370,12 +363,8 @@ class KnowledgeTableStore: print(f"{e}, retry...", flush=True) await asyncio.sleep(1) - print("Unpack...") for row in resp: - print("ROW") - print(row) - if row[2]: metadata = [ Triple( @@ -402,6 +391,7 @@ class KnowledgeTableStore: metadata = Metadata( id = document_id, user = user, + collection = "default", # FIXME: What to put here? metadata = metadata, ), triples = triples @@ -432,7 +422,6 @@ class KnowledgeTableStore: for row in resp: - print("MD...") if row[2]: metadata = [ Triple( @@ -445,7 +434,6 @@ class KnowledgeTableStore: else: metadata = [] - print("EE...") entities = [ EntityEmbeddings( entity = Value(value = ent[0][0], is_uri = ent[0][1]), @@ -459,6 +447,7 @@ class KnowledgeTableStore: metadata = Metadata( id = document_id, user = user, + collection = "default", # FIXME: What to put here? metadata = metadata, ), entities = entities