Processing-Server

Dockerfile

@@ -14,6 +14,7 @@ COPY ocrd_models ./ocrd_models
 COPY ocrd_utils ./ocrd_utils
 RUN mv ./ocrd_utils/ocrd_logging.conf /etc
 COPY ocrd_validators/ ./ocrd_validators
+COPY ocrd_network/ ./ocrd_network
 COPY Makefile .
 COPY README.md .
 COPY LICENSE .

Makefile

@@ -9,7 +9,7 @@ TESTDIR = tests
 
 SPHINX_APIDOC = 
 
-BUILD_ORDER = ocrd_utils ocrd_models ocrd_modelfactory ocrd_validators ocrd
+BUILD_ORDER = ocrd_utils ocrd_models ocrd_modelfactory ocrd_validators ocrd_network ocrd
 
 FIND_VERSION = grep version= ocrd_utils/setup.py|grep -Po "([0-9ab]+\.?)+"
 

README.md

@@ -25,6 +25,7 @@
 	* [ocrd_models](#ocrd_models)
 	* [ocrd_modelfactory](#ocrd_modelfactory)
 	* [ocrd_validators](#ocrd_validators)
+	* [ocrd_network](#ocrd_network)
 	* [ocrd](#ocrd)
 * [bash library](#bash-library)
 	* [bashlib API](#bashlib-api)
@@ -122,6 +123,12 @@ Schemas and routines for validating BagIt, `ocrd-tool.json`, workspaces, METS, p
 
 See [README for `ocrd_validators`](./ocrd_validators/README.md) for further information.
 
+### ocrd_network
+
+Tools for offering (web-)services with OCR-D
+
+See [README for `ocrd_network`](./ocrd_network/README.md) for further information.
+
 ### ocrd
 
 Depends on all of the above, also contains decorators and classes for creating OCR-D processors and CLIs.

ocrd/ocrd/cli/__init__.py

@@ -31,6 +31,9 @@ def get_help(self, ctx):
 from ocrd.decorators import ocrd_loglevel
 from .zip import zip_cli
 from .log import log_cli
+from .processing_server import processing_server_cli
+from .processing_worker import processing_worker_cli
+
 
 @click.group()
 @click.version_option()
@@ -48,3 +51,5 @@ def cli(**kwargs): # pylint: disable=unused-argument
 cli.add_command(validate_cli)
 cli.add_command(log_cli)
 cli.add_command(resmgr_cli)
+cli.add_command(processing_server_cli)
+cli.add_command(processing_worker_cli)

ocrd/ocrd/cli/processing_server.py

@@ -0,0 +1,41 @@
+"""
+OCR-D CLI: start the processing server
+
+.. click:: ocrd.cli.processing_server:processing_server_cli
+    :prog: ocrd processing-server
+    :nested: full
+"""
+import click
+import logging
+from ocrd_utils import initLogging
+from ocrd_network import (
+    ProcessingServer,
+    ProcessingServerParamType
+)
+
+
+@click.command('processing-server')
+@click.argument('path_to_config', required=True, type=click.STRING)
+@click.option('-a', '--address',
+              default="localhost:8080",
+              help='The URL of the Processing server, format: host:port',
+              type=ProcessingServerParamType(),
+              required=True)
+def processing_server_cli(path_to_config, address: str):
+    """
+    Start and manage processing workers with the processing server
+
+    PATH_TO_CONFIG is a yaml file to configure the server and the workers. See
+    https://github.com/OCR-D/spec/pull/222/files#diff-a71bf71cbc7d9ce94fded977f7544aba4df9e7bdb8fc0cf1014e14eb67a9b273
+    for further information (TODO: update path when spec is available/merged)
+
+    """
+    initLogging()
+    # TODO: Remove before the release
+    logging.getLogger('paramiko.transport').setLevel(logging.INFO)
+    logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
+
+    # Note, the address is already validated with the type field
+    host, port = address.split(':')
+    processing_server = ProcessingServer(path_to_config, host, port)
+    processing_server.start()

ocrd/ocrd/cli/processing_worker.py

@@ -0,0 +1,61 @@
+"""
+OCR-D CLI: start the processing worker
+
+.. click:: ocrd.cli.processing_worker:processing_worker_cli
+    :prog: ocrd processing-worker
+    :nested: full
+"""
+import click
+import logging
+from ocrd_utils import (
+    initLogging,
+    get_ocrd_tool_json
+)
+from ocrd_network import (
+    DatabaseParamType,
+    ProcessingWorker,
+    QueueServerParamType,
+)
+
+
+@click.command('processing-worker')
+@click.argument('processor_name', required=True, type=click.STRING)
+@click.option('-q', '--queue',
+              default="amqp://admin:admin@localhost:5672/",
+              help='The URL of the Queue Server, format: amqp://username:password@host:port/vhost',
+              type=QueueServerParamType())
+@click.option('-d', '--database',
+              default="mongodb://localhost:27018",
+              help='The URL of the MongoDB, format: mongodb://host:port',
+              type=DatabaseParamType())
+def processing_worker_cli(processor_name: str, queue: str, database: str):
+    """
+    Start a processing worker (a specific ocr-d processor)
+    """
+    initLogging()
+    # TODO: Remove before the release
+    logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
+
+    # Get the ocrd_tool dictionary
+    # ocrd_tool = parse_json_string_with_comments(
+    #     run([processor_name, '--dump-json'], stdout=PIPE, check=True, universal_newlines=True).stdout
+    # )
+
+    ocrd_tool = get_ocrd_tool_json(processor_name)
+    if not ocrd_tool:
+        raise Exception(f"The ocrd_tool is empty or missing")
+
+    try:
+        processing_worker = ProcessingWorker(
+            rabbitmq_addr=queue,
+            mongodb_addr=database,
+            processor_name=ocrd_tool['executable'],
+            ocrd_tool=ocrd_tool,
+            processor_class=None,  # For readability purposes assigned here
+        )
+        # The RMQConsumer is initialized and a connection to the RabbitMQ is performed
+        processing_worker.connect_consumer()
+        # Start consuming from the queue with name `processor_name`
+        processing_worker.start_consuming()
+    except Exception as e:
+        raise Exception("Processing worker has failed with error") from e

ocrd/ocrd/decorators/__init__.py

@@ -1,6 +1,8 @@
 from os.path import isfile
 from os import environ
 import sys
+from contextlib import redirect_stdout
+from io import StringIO
 
 import click
 
@@ -10,9 +12,11 @@
     set_json_key_value_overrides,
 )
 
-from ocrd_utils import getLogger, initLogging
+from ocrd_utils import getLogger, initLogging, parse_json_string_with_comments
 from ocrd_validators import WorkspaceValidator
 
+from ocrd_network import ProcessingWorker
+
 from ..resolver import Resolver
 from ..processor.base import run_processor
 
@@ -35,6 +39,8 @@ def ocrd_cli_wrap_processor(
     overwrite=False,
     show_resource=None,
     list_resources=False,
+    queue=None,
+    database=None,
     **kwargs
 ):
     if not sys.argv[1:]:
@@ -51,6 +57,35 @@ def ocrd_cli_wrap_processor(
             list_resources=list_resources
         )
         sys.exit()
+    # If either of these two is provided but not both
+    if bool(queue) != bool(database):
+        raise Exception("Options --queue and --database require each other.")
+    # If both of these are provided - start the processing worker instead of the processor - processorClass
+    if queue and database:
+        initLogging()
+        # TODO: Remove before the release
+        # We are importing the logging here because it's not the ocrd logging but python one
+        import logging
+        logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
+
+        # Get the ocrd_tool dictionary
+        processor = processorClass(workspace=None, dump_json=True)
+        ocrd_tool = processor.ocrd_tool
+
+        try:
+            processing_worker = ProcessingWorker(
+                rabbitmq_addr=queue,
+                mongodb_addr=database,
+                processor_name=ocrd_tool['executable'],
+                ocrd_tool=ocrd_tool,
+                processor_class=processorClass,
+            )
+            # The RMQConsumer is initialized and a connection to the RabbitMQ is performed
+            processing_worker.connect_consumer()
+            # Start consuming from the queue with name `processor_name`
+            processing_worker.start_consuming()
+        except Exception as e:
+            raise Exception("Processing worker has failed with error") from e
     else:
         initLogging()
         LOG = getLogger('ocrd_cli_wrap_processor')

ocrd/ocrd/decorators/ocrd_cli_options.py

@@ -1,6 +1,8 @@
 from click import option
 from .parameter_option import parameter_option, parameter_override_option
 from .loglevel_option import loglevel_option
+from ocrd_network import QueueServerParamType, DatabaseParamType
+
 
 def ocrd_cli_options(f):
     """
@@ -26,6 +28,8 @@ def cli(mets_url):
         option('-O', '--output-file-grp', help='File group(s) used as output.', default='OUTPUT'),
         option('-g', '--page-id', help="ID(s) of the pages to process"),
         option('--overwrite', help="Overwrite the output file group or a page range (--page-id)", is_flag=True, default=False),
+        option('--queue', help="The URL of the Queue Server, format: username:password@host:port/vhost", type=QueueServerParamType()),
+        option('--database', help="The URL of the MongoDB, format: mongodb://host:port", type=DatabaseParamType()),
         option('-C', '--show-resource', help='Dump the content of processor resource RESNAME', metavar='RESNAME'),
         option('-L', '--list-resources', is_flag=True, default=False, help='List names of processor resources'),
         parameter_option,
@@ -41,3 +45,4 @@ def cli(mets_url):
     for param in params:
         param(f)
     return f
+

ocrd/ocrd/processor/helpers.py

@@ -1,7 +1,7 @@
 """
 Helper methods for running and documenting processors
 """
-from os import environ
+from os import chdir, environ, getcwd
 from time import perf_counter, process_time
 from functools import lru_cache
 import json
@@ -14,7 +14,7 @@
 
 from click import wrap_text
 from ocrd.workspace import Workspace
-from ocrd_utils import freeze_args, getLogger
+from ocrd_utils import freeze_args, getLogger, pushd_popd
 
 
 __all__ = [
@@ -84,16 +84,19 @@ def run_processor(
     log = getLogger('ocrd.processor.helpers.run_processor')
     log.debug("Running processor %s", processorClass)
 
+    old_cwd = getcwd()
     processor = get_processor(
         processor_class=processorClass,
         parameter=parameter,
-        workspace=workspace,
+        workspace=None,
         ocrd_tool=ocrd_tool,
         page_id=page_id,
         input_file_grp=input_file_grp,
         output_file_grp=output_file_grp,
         instance_caching=instance_caching
     )
+    processor.workspace = workspace
+    chdir(processor.workspace.directory)
 
     ocrd_tool = processor.ocrd_tool
     name = '%s v%s' % (ocrd_tool['executable'], processor.version)
@@ -104,21 +107,34 @@ def run_processor(
     t0_cpu = process_time()
     if any(x in environ.get('OCRD_PROFILE', '') for x in ['RSS', 'PSS']):
         backend = 'psutil_pss' if 'PSS' in environ['OCRD_PROFILE'] else 'psutil'
-        mem_usage = memory_usage(proc=processor.process,
-                                # only run process once
-                                max_iterations=1,
-                                interval=.1, timeout=None, timestamps=True, 
-                                # include sub-processes
-                                multiprocess=True, include_children=True, 
-                                # get proportional set size instead of RSS
-                                backend=backend)
+        try:
+            mem_usage = memory_usage(proc=processor.process,
+                                    # only run process once
+                                    max_iterations=1,
+                                    interval=.1, timeout=None, timestamps=True,
+                                    # include sub-processes
+                                    multiprocess=True, include_children=True,
+                                    # get proportional set size instead of RSS
+                                    backend=backend)
+        except Exception as err:
+            log.exception("Failure in processor '%s'" % ocrd_tool['executable'])
+            raise err
+        finally:
+            chdir(old_cwd)
         mem_usage_values = [mem for mem, _ in mem_usage]
         mem_output = 'memory consumption: '
         mem_output += ''.join(sparklines(mem_usage_values))
         mem_output += ' max: %.2f MiB min: %.2f MiB' % (max(mem_usage_values), min(mem_usage_values))
         logProfile.info(mem_output)
     else:
-        processor.process()
+        try:
+            processor.process()
+        except Exception as err:
+            log.exception("Failure in processor '%s'" % ocrd_tool['executable'])
+            raise err
+        finally:
+            chdir(old_cwd)
+
     t1_wall = perf_counter() - t0_wall
     t1_cpu = process_time() - t0_cpu
     logProfile.info("Executing processor '%s' took %fs (wall) %fs (CPU)( [--input-file-grp='%s' --output-file-grp='%s' --parameter='%s' --page-id='%s']" % (
@@ -198,7 +214,7 @@ def run_cli(
 
 def generate_processor_help(ocrd_tool, processor_instance=None):
     """Generate a string describing the full CLI of this processor including params.
-    
+
     Args:
          ocrd_tool (dict): this processor's ``tools`` section of the module's ``ocrd-tool.json``
          processor_instance (object, optional): the processor implementation
@@ -247,6 +263,8 @@ def wrap(s):
   -g, --page-id ID                Physical page ID(s) to process
   --overwrite                     Remove existing output pages/images
                                   (with --page-id, remove only those)
+  --queue                         The RabbitMQ server address in format: {host}:{port}/{vhost}"
+  --database                      The MongoDB address in format: mongodb://{host}:{port}"
   --profile                       Enable profiling
   --profile-file                  Write cProfile stats to this file. Implies --profile
   -p, --parameter JSON-PATH       Parameters, either verbatim JSON string

ocrd/setup.py

@@ -8,6 +8,7 @@
 install_requires.append('ocrd_models == %s' % VERSION)
 install_requires.append('ocrd_modelfactory == %s' % VERSION)
 install_requires.append('ocrd_validators == %s' % VERSION)
+install_requires.append('ocrd_network == %s' % VERSION)
 
 setup(
     name='ocrd',

ocrd_network/README.md

@@ -0,0 +1,5 @@
+# ocrd_network
+
+> OCR-D framework - web API
+
+See also: https://github.com/OCR-D/core

ocrd_network/ocrd_network/__init__.py

@@ -0,0 +1,31 @@
+# This network package is supposed to contain all the packages and modules to realize the network architecture:
+# https://github.com/OCR-D/spec/pull/222/files#diff-8d0dae8c9277ff1003df93c5359c82a12d3f5c8452281f87781921921204d283
+
+# For reference, currently:
+# 1. The WebAPI is available here: https://github.com/OCR-D/ocrd-webapi-implementation
+# The ocrd-webapi-implementation repo implements the Discovery / Workflow / Workspace endpoints of the WebAPI currently.
+# This Processing Server PR implements just the Processor endpoint of the WebAPI. 
+# Once we have this merged to core under ocrd-network, the other endpoints will be adapted to ocrd-network 
+# and then the ocrd-webapi-implementation repo can be archived for reference.
+
+# 2. The RabbitMQ Library (i.e., utils) is used as an API to abstract and 
+# simplify (from the view point of processing server and workers) interactions with the RabbitMQ Server.
+# The library was adopted from: https://github.com/OCR-D/ocrd-webapi-implementation/tree/main/ocrd_webapi/rabbitmq
+
+# 3. Some potentially more useful code to be adopted for the Processing Server/Worker is available here:
+# https://github.com/OCR-D/core/pull/884
+# Update: Should be revisited again for adopting any relevant parts (if necessary). 
+# Nothing relevant is under the radar for now.
+
+# 4. The Mets Server discussion/implementation is available here:
+# https://github.com/OCR-D/core/pull/966
+
+# Note: The Mets Server is still not placed on the architecture diagram and probably won't be a part of
+# the network package. The reason, Mets Server is tightly coupled with the `OcrdWorkspace`.
+from .processing_server import ProcessingServer
+from .processing_worker import ProcessingWorker
+from .param_validators import (
+    DatabaseParamType,
+    ProcessingServerParamType,
+    QueueServerParamType
+)