Standalone Processor Server (#884 + #974)

ocrd/ocrd/cli/__init__.py

@@ -33,6 +33,7 @@ def get_help(self, ctx):
 from .log import log_cli
 from .processing_server import processing_server_cli
 from .processing_worker import processing_worker_cli
+from .processor_server import processor_server_cli
 
 
 @click.group()
@@ -53,3 +54,4 @@ def cli(**kwargs): # pylint: disable=unused-argument
 cli.add_command(resmgr_cli)
 cli.add_command(processing_server_cli)
 cli.add_command(processing_worker_cli)
+cli.add_command(processor_server_cli)

ocrd/ocrd/cli/processing_worker.py

@@ -23,12 +23,14 @@
 @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())
+              type=QueueServerParamType(),
+              required=True)
 @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):
+              type=DatabaseParamType(),
+              required=True)
+def processing_worker_cli(processor_name: str, agent_type: str, queue: str, database: str):
     """
     Start a processing worker (a specific ocr-d processor)
     """

ocrd/ocrd/cli/processor_server.py

@@ -0,0 +1,47 @@
+"""
+OCR-D CLI: start the processor server
+
+.. click:: ocrd.cli.processor_server:processor_server_cli
+    :prog: ocrd processor-server
+    :nested: full
+"""
+import click
+import logging
+from ocrd_utils import initLogging
+from ocrd_network import (
+    DatabaseParamType,
+    ProcessingServerParamType,
+    ProcessorServer,
+)
+
+
+@click.command('processor-server')
+@click.argument('processor_name', required=True, type=click.STRING)
+@click.option('--agent_address',
+              help='The URL of the processor server, format: host:port',
+              type=ProcessingServerParamType(),
+              required=True)
+@click.option('-d', '--database',
+              default="mongodb://localhost:27018",
+              help='The URL of the MongoDB, format: mongodb://host:port',
+              type=DatabaseParamType(),
+              required=True)
+def processor_server_cli(processor_name: str, agent_type: str, agent_address: str, database: str):
+    """
+    Start ocr-d processor as a server
+    """
+    initLogging()
+    # TODO: Remove before the release
+    logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
+
+    try:
+        # TODO: Better validate that inside the ProcessorServer itself
+        host, port = agent_address.split(':')
+        processor_server = ProcessorServer(
+            mongodb_addr=database,
+            processor_name=processor_name,
+            processor_class=None,  # For readability purposes assigned here
+        )
+        processor_server.run_server(host=host, port=int(port))
+    except Exception as e:
+        raise Exception("Processor server has failed with error") from e

ocrd/ocrd/decorators/__init__.py

@@ -1,10 +1,5 @@
-from os.path import isfile
 from os import environ
 import sys
-from contextlib import redirect_stdout
-from io import StringIO
-
-import click
 
 from ocrd_utils import (
     is_local_filename,
@@ -15,7 +10,7 @@
 from ocrd_utils import getLogger, initLogging, parse_json_string_with_comments
 from ocrd_validators import WorkspaceValidator
 
-from ocrd_network import ProcessingWorker
+from ocrd_network import ProcessingWorker, ProcessorServer
 
 from ..resolver import Resolver
 from ..processor.base import run_processor
@@ -38,8 +33,12 @@ def ocrd_cli_wrap_processor(
     overwrite=False,
     show_resource=None,
     list_resources=False,
+    # ocrd_network params start #
+    agent_type=None,
+    agent_address=None,
     queue=None,
     database=None,
+    # ocrd_network params end #
     **kwargs
 ):
     if not sys.argv[1:]:
@@ -56,96 +55,127 @@ 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
+
+    initLogging()
+
+    # Used for checking/starting network agents for the WebAPI architecture
+    # Has no side effects if neither of the 4 ocrd_network parameters are passed
+    check_and_run_network_agent(processorClass, agent_type, agent_address, database, queue)
+
+    LOG = getLogger('ocrd_cli_wrap_processor')
+    # LOG.info('kwargs=%s' % kwargs)
+    # Merge parameter overrides and parameters
+    if 'parameter_override' in kwargs:
+        set_json_key_value_overrides(kwargs['parameter'], *kwargs['parameter_override'])
+    # TODO OCR-D/core#274
+    # Assert -I / -O
+    # if not kwargs['input_file_grp']:
+    #     raise ValueError('-I/--input-file-grp is required')
+    # if not kwargs['output_file_grp']:
+    #     raise ValueError('-O/--output-file-grp is required')
+    resolver = Resolver()
+    working_dir, mets, _ = resolver.resolve_mets_arguments(working_dir, mets, None)
+    workspace = resolver.workspace_from_url(mets, working_dir)
+    page_id = kwargs.get('page_id')
+    # XXX not possible while processors do not adhere to # https://github.com/OCR-D/core/issues/505
+    # if overwrite
+    #     if 'output_file_grp' not in kwargs or not kwargs['output_file_grp']:
+    #         raise Exception("--overwrite requires --output-file-grp")
+    #     LOG.info("Removing files because of --overwrite")
+    #     for grp in kwargs['output_file_grp'].split(','):
+    #         if page_id:
+    #             for one_page_id in kwargs['page_id'].split(','):
+    #                 LOG.debug("Removing files in output file group %s with page ID %s", grp, one_page_id)
+    #                 for file in workspace.mets.find_files(pageId=one_page_id, fileGrp=grp):
+    #                     workspace.remove_file(file, force=True, keep_file=False, page_recursive=True)
+    #         else:
+    #             LOG.debug("Removing all files in output file group %s ", grp)
+    #             # TODO: can be reduced to `page_same_group=True` as soon as core#505 has landed (in all processors)
+    #             workspace.remove_file_group(grp, recursive=True, force=True, keep_files=False, page_recursive=True, page_same_group=False)
+    #     workspace.save_mets()
+    # XXX While https://github.com/OCR-D/core/issues/505 is open, set 'overwrite_mode' globally on the workspace
+    if overwrite:
+        workspace.overwrite_mode = True
+    report = WorkspaceValidator.check_file_grp(workspace, kwargs['input_file_grp'], '' if overwrite else kwargs['output_file_grp'], page_id)
+    if not report.is_valid:
+        raise Exception("Invalid input/output file grps:\n\t%s" % '\n\t'.join(report.errors))
+    # Set up profiling behavior from environment variables/flags
+    if not profile and 'OCRD_PROFILE' in environ:
+        if 'CPU' in environ['OCRD_PROFILE']:
+            profile = True
+    if not profile_file and 'OCRD_PROFILE_FILE' in environ:
+        profile_file = environ['OCRD_PROFILE_FILE']
+    if profile or profile_file:
+        import cProfile
+        import pstats
+        import io
+        import atexit
+        print("Profiling...")
+        pr = cProfile.Profile()
+        pr.enable()
+        def exit():
+            pr.disable()
+            print("Profiling completed")
+            if profile_file:
+                with open(profile_file, 'wb') as f:
+                    pr.dump_stats(profile_file)
+            s = io.StringIO()
+            pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats()
+            print(s.getvalue())
+        atexit.register(exit)
+    run_processor(processorClass, mets_url=mets, workspace=workspace, **kwargs)
+
+
+def check_and_run_network_agent(ProcessorClass, agent_type: str, agent_address: str, database: str, queue: str):
+    if not agent_type and (agent_address or database or queue):
+        raise ValueError("Options '--database', '--queue', and 'agent_address' are valid only with '--agent_type'")
+    if agent_type:
+        if not database:
+            raise ValueError("Options '--agent_type' and '--database' are mutually inclusive")
+        allowed_agent_types = ['server', 'worker']
+        if agent_type not in allowed_agent_types:
+            agents_str = ', '.join(allowed_agent_types)
+            raise ValueError(f"Wrong agent type parameter. Allowed agent types: {agents_str}")
+        if agent_type == 'server':
+            if not agent_address:
+                raise ValueError("Options '--agent_type=server' and '--agent_address' are mutually inclusive")
+            if queue:
+                raise ValueError("Options '--agent_type=server' and '--queue' are mutually exclusive")
+        if agent_type == 'worker':
+            if not queue:
+                raise ValueError("Options '--agent_type=worker' and '--queue' are mutually inclusive")
+            if agent_address:
+                raise ValueError("Options '--agent_type=worker' and '--agent_address' are mutually exclusive")
+
         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')
-        # LOG.info('kwargs=%s' % kwargs)
-        # Merge parameter overrides and parameters
-        if 'parameter_override' in kwargs:
-            set_json_key_value_overrides(kwargs['parameter'], *kwargs['parameter_override'])
-        # TODO OCR-D/core#274
-        # Assert -I / -O
-        # if not kwargs['input_file_grp']:
-        #     raise ValueError('-I/--input-file-grp is required')
-        # if not kwargs['output_file_grp']:
-        #     raise ValueError('-O/--output-file-grp is required')
-        resolver = Resolver()
-        working_dir, mets, _ = resolver.resolve_mets_arguments(working_dir, mets, None)
-        workspace = resolver.workspace_from_url(mets, working_dir)
-        page_id = kwargs.get('page_id')
-        # XXX not possible while processors do not adhere to # https://github.com/OCR-D/core/issues/505
-        # if overwrite
-        #     if 'output_file_grp' not in kwargs or not kwargs['output_file_grp']:
-        #         raise Exception("--overwrite requires --output-file-grp")
-        #     LOG.info("Removing files because of --overwrite")
-        #     for grp in kwargs['output_file_grp'].split(','):
-        #         if page_id:
-        #             for one_page_id in kwargs['page_id'].split(','):
-        #                 LOG.debug("Removing files in output file group %s with page ID %s", grp, one_page_id)
-        #                 for file in workspace.mets.find_files(pageId=one_page_id, fileGrp=grp):
-        #                     workspace.remove_file(file, force=True, keep_file=False, page_recursive=True)
-        #         else:
-        #             LOG.debug("Removing all files in output file group %s ", grp)
-        #             # TODO: can be reduced to `page_same_group=True` as soon as core#505 has landed (in all processors)
-        #             workspace.remove_file_group(grp, recursive=True, force=True, keep_files=False, page_recursive=True, page_same_group=False)
-        #     workspace.save_mets()
-        # XXX While https://github.com/OCR-D/core/issues/505 is open, set 'overwrite_mode' globally on the workspace
-        if overwrite:
-            workspace.overwrite_mode = True
-        report = WorkspaceValidator.check_file_grp(workspace, kwargs['input_file_grp'], '' if overwrite else kwargs['output_file_grp'], page_id)
-        if not report.is_valid:
-            raise Exception("Invalid input/output file grps:\n\t%s" % '\n\t'.join(report.errors))
-        # Set up profiling behavior from environment variables/flags
-        if not profile and 'OCRD_PROFILE' in environ:
-            if 'CPU' in environ['OCRD_PROFILE']:
-                profile = True
-        if not profile_file and 'OCRD_PROFILE_FILE' in environ:
-            profile_file = environ['OCRD_PROFILE_FILE']
-        if profile or profile_file:
-            import cProfile
-            import pstats
-            import io
-            import atexit
-            print("Profiling...")
-            pr = cProfile.Profile()
-            pr.enable()
-            def exit():
-                pr.disable()
-                print("Profiling completed")
-                if profile_file:
-                    with open(profile_file, 'wb') as f:
-                        pr.dump_stats(profile_file)
-                s = io.StringIO()
-                pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats()
-                print(s.getvalue())
-            atexit.register(exit)
-        run_processor(processorClass, mets_url=mets, workspace=workspace, **kwargs)
+        processor = ProcessorClass(workspace=None, dump_json=True)
+        if agent_type == 'worker':
+            try:
+                # TODO: Passing processor_name and ocrd_tool is reduntant
+                processing_worker = ProcessingWorker(
+                    rabbitmq_addr=queue,
+                    mongodb_addr=database,
+                    processor_name=processor.ocrd_tool['executable'],
+                    ocrd_tool=processor.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:
+                sys.exit(f"Processing worker has failed with error: {e}")
+        if agent_type == 'server':
+            try:
+                # TODO: Better validate that inside the ProcessorServer itself
+                host, port = agent_address.split(':')
+                processor_server = ProcessorServer(
+                    mongodb_addr=database,
+                    processor_name=processor.ocrd_tool['executable'],
+                    processor_class=ProcessorClass,
+                )
+                processor_server.run_server(host=host, port=int(port))
+            except Exception as e:
+                sys.exit(f"Processor server has failed with error: {e}")

ocrd/ocrd/decorators/ocrd_cli_options.py

@@ -1,3 +1,4 @@
+import click
 from click import option, Path
 from .parameter_option import parameter_option, parameter_override_option
 from .loglevel_option import loglevel_option
@@ -33,6 +34,8 @@ def cli(mets_url):
         parameter_option,
         parameter_override_option,
         loglevel_option,
+        option('--agent_type', type=click.STRING),
+        option('--agent_address', type=click.STRING),
         option('--queue', type=QueueServerParamType()),
         option('--database', type=DatabaseParamType()),
         option('-C', '--show-resource'),
@@ -45,4 +48,3 @@ def cli(mets_url):
     for param in params:
         param(f)
     return f
-

ocrd_network/ocrd_network/__init__.py

@@ -24,6 +24,7 @@
 # the network package. The reason, Mets Server is tightly coupled with the `OcrdWorkspace`.
 from .processing_server import ProcessingServer
 from .processing_worker import ProcessingWorker
+from .processor_server import ProcessorServer
 from .param_validators import (
     DatabaseParamType,
     ProcessingServerParamType,