From 98faec7798a4886b7b4a00cde46dd21125a62b0d Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 15 Apr 2025 00:12:50 +0100 Subject: [PATCH] PDF done but not working --- trustgraph-base/trustgraph/base/__init__.py | 1 + .../trustgraph/base/async_processor.py | 3 + .../trustgraph/base/input_output.py | 117 +++++++++++++++++ .../trustgraph/base/request_response.py | 118 ++++++++++++++++++ .../trustgraph/config/service/service.py | 5 +- .../trustgraph/decoding/pdf/pdf_decoder.py | 21 ++-- 6 files changed, 247 insertions(+), 18 deletions(-) create mode 100644 trustgraph-base/trustgraph/base/input_output.py create mode 100644 trustgraph-base/trustgraph/base/request_response.py diff --git a/trustgraph-base/trustgraph/base/__init__.py b/trustgraph-base/trustgraph/base/__init__.py index 4bf074e4..79cb5af2 100644 --- a/trustgraph-base/trustgraph/base/__init__.py +++ b/trustgraph-base/trustgraph/base/__init__.py @@ -6,4 +6,5 @@ from . producer import Producer from . publisher import Publisher from . subscriber import Subscriber from . metrics import ProcessorMetrics, ConsumerMetrics, ProducerMetrics +from . input_output import InputOutputProcessor diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index 39875342..1203aa09 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/trustgraph-base/trustgraph/base/async_processor.py @@ -50,6 +50,9 @@ class AsyncProcessor: @property def pulsar_host(self): return self.client.pulsar_host + def on_config(self, handler): + self.config_handlers.append(handler) + async def start(self): self.config_sub_task = self.subscribe( diff --git a/trustgraph-base/trustgraph/base/input_output.py b/trustgraph-base/trustgraph/base/input_output.py new file mode 100644 index 00000000..1e72c444 --- /dev/null +++ b/trustgraph-base/trustgraph/base/input_output.py @@ -0,0 +1,117 @@ + +from pulsar.schema import JsonSchema + +from .. schema import Error +from .. schema import config_request_queue, config_response_queue +from .. schema import config_push_queue +from .. log_level import LogLevel +from .. base import AsyncProcessor, Consumer, Producer + +from .. base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics + +class InputOutputProcessor(AsyncProcessor): + + def __init__(self, **params): + + self.id = params.get("id") + self.subscriber = params.get("subscriber") + self.input_schema = params.get("input_schema") + self.output_schema = params.get("output_schema") + + ProcessorMetrics(id=self.id).info( + { + "subscriber": self.subscriber, + "input_schema": self.input_schema.__name__, + "output_schema": self.output_schema.__name__, + } + ) + + super(InputOutputProcessor, self).__init__( + **params | { + "id": self.id, + "input_schema": self.input_schema.__name__, + "output_schema": self.output_schema.__name__, + } + ) + + self.on_config(self.on_configuration) + + self.subs = {} + self.pubs = {} + + print("Service initialised.") + + async def start_handler(self, flow, defn): + + input_metrics = ConsumerMetrics(self.id, flow) + output_metrics = ProducerMetrics(self.id, flow) + + self.subs[flow] = self.subscribe( + queue = defn["input"], + subscriber = subscriber, + schema = self.input_schema, + handler = self.on_message, + metrics = input_metrics, + ) + + self.pubs[flow] = self.publish( + queue = defn["output"], + schema = self.output_schema, + metrics = output_metrics, + ) + + self.subs[flow].start() + + print("Started flow for", flow) + + async def on_configuration(self, config, version): + + if "flows" not in config: return + + if self.id in config["flows"]: + + wanted_keys = config["flows"][self.id].keys() + current_keys = self.subs.keys() + + for key in wanted_keys: + if key not in current_keys: + self.start_handler(key, config["flows"][key]) + + for key in current_keys: + if key not in wanted_keys: + self.stop_handler(key, config["flows"][key]) + + print("Handled config update") + + async def start(self): + pass + + @staticmethod + def add_args(parser, default_subscriber): + + AsyncProcessor.add_args(parser) + + parser.add_argument( + '-s', '--subscriber', + default=default_subscriber, + help=f'Queue subscriber name (default: {default_subscriber})' + ) + + # parser.add_argument( + # '--rate-limit-retry', + # type=int, + # default=default_rate_limit_retry, + # help=f'Rate limit retry (default: {default_rate_limit_retry})' + # ) + + # parser.add_argument( + # '--rate-limit-timeout', + # type=int, + # default=default_rate_limit_timeout, + # help=f'Rate limit timeout (default: {default_rate_limit_timeout})' + # ) + +def run(): + + Processor.launch(module, __doc__) + diff --git a/trustgraph-base/trustgraph/base/request_response.py b/trustgraph-base/trustgraph/base/request_response.py new file mode 100644 index 00000000..6ccaed80 --- /dev/null +++ b/trustgraph-base/trustgraph/base/request_response.py @@ -0,0 +1,118 @@ + +from pulsar.schema import JsonSchema + +from .. schema import Error +from .. schema import config_request_queue, config_response_queue +from .. schema import config_push_queue +from .. log_level import LogLevel +from .. base import AsyncProcessor, Consumer, Producer + +from .. base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics + +class RequestResponseProcessor(AsyncProcessor): + + def __init__(self, **params): + + id = params.get("id") + self.subscriber = params.get("subscriber") + self.request_schema = params.get("request_schema") + self.response_schema = params.get("response_schema") + + ProcessorMetrics(id=id).info( + { + "subscriber": self.subscriber, + "request_schema": self.request_schema.__name__, + "response_schema": self.response_schema.__name__, + } + ) + + super(Processor, self).__init__( + **params | { + "request_schema": self.request_schema.__name__, + "response_schema": self.response_schema.__name__, + } + ) + + self.on_config(self.on_configuration) + + self.subs = {} + self.pubs = {} + + print("Service initialised.") + + async def start_handler(self, flow, defn): + + request_metrics = ConsumerMetrics(self.id, flow) + response_metrics = ProducerMetrics(self.id, flow) + + self.subs[flow] = self.subscribe( + queue = defn["input"] + subscriber = subscriber, + schema = self.request_schema, + handler = self.on_message, + metrics = request_metrics, + ) + + self.pubs[flow] = self.publish( + queue = defn["output"], + schema = self.response_schema, + metrics = response_metrics, + ) + + self.subs[flow].start() + + print("Started flow for", flow) + + async def on_configuration(self, config, version): + + if "flows" not in config: return + + if self.id in config["flows"]: + + wanted_keys = config["flows"][self.id].keys() + current_keys = self.subs.keys() + + for key in wanted_keys: + if key not in current_keys: + self.start_handler(key, config["flows"][key]) + + for key in current_keys: + if key not in wanted_keys: + self.stop_handler(key, config["flows"][key]) + + print("Handled config update") + + async def start(self): + + await self.request_sub.start() + + async def on_message(self, msg, consumer): + + try: + + v = msg.value() + + # Sender-produced ID + id = msg.properties()["id"] + + print(f"Handling {id}...", flush=True) + + consumer.acknowledge(msg) + + except Exception as e: + + resp = "error" + + await self.response_pub.send(resp, properties={"id": id}) + + consumer.acknowledge(msg) + + @staticmethod + def add_args(parser): + + AsyncProcessor.add_args(parser) + +def run(): + + Processor.launch(module, __doc__) + diff --git a/trustgraph-flow/trustgraph/config/service/service.py b/trustgraph-flow/trustgraph/config/service/service.py index e64c31b2..93f4101b 100644 --- a/trustgraph-flow/trustgraph/config/service/service.py +++ b/trustgraph-flow/trustgraph/config/service/service.py @@ -1,11 +1,9 @@ """ -Config service. Fetchs an extract from the Wikipedia page -using the API. +Config service. Manages system global configuration state """ from pulsar.schema import JsonSchema -from prometheus_client import Histogram, Counter from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush from trustgraph.schema import Error @@ -109,7 +107,6 @@ class Processor(AsyncProcessor): async def on_message(self, msg, consumer): - try: v = msg.value() diff --git a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py index a2cc2adb..0fabe0aa 100755 --- a/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py +++ b/trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py @@ -11,26 +11,22 @@ from langchain_community.document_loaders import PyPDFLoader from ... schema import Document, TextDocument, Metadata from ... schema import document_ingest_queue, text_ingest_queue from ... log_level import LogLevel -from ... base import ConsumerProducer +from ... base import InputOutputProcessor module = "pdf-decoder" -default_input_queue = document_ingest_queue -default_output_queue = text_ingest_queue default_subscriber = module -class Processor(ConsumerProducer): +class Processor(InputOutputProcessor): def __init__(self, **params): - input_queue = params.get("input_queue", default_input_queue) - output_queue = params.get("output_queue", default_output_queue) - subscriber = params.get("subscriber", default_subscriber) + id = params.get("id") + subscriber = params.get("subscriber") super(Processor, self).__init__( **params | { - "input_queue": input_queue, - "output_queue": output_queue, + "id": id, "subscriber": subscriber, "input_schema": Document, "output_schema": TextDocument, @@ -39,7 +35,7 @@ class Processor(ConsumerProducer): print("PDF inited") - async def handle(self, msg): + async def on_message(self, msg, consumer): print("PDF message received") @@ -71,10 +67,7 @@ class Processor(ConsumerProducer): @staticmethod def add_args(parser): - ConsumerProducer.add_args( - parser, default_input_queue, default_subscriber, - default_output_queue, - ) + InputOutputProcessor.add_args(parser, default_subscriber) def run():