Extend the network client

src/ocrd/cli/__init__.py

@@ -35,6 +35,10 @@
 \b
 {config.describe('OCRD_MAX_PROCESSOR_CACHE')}
 \b
+{config.describe('OCRD_NETWORK_CLIENT_POLLING_SLEEP')}
+\b
+{config.describe('OCRD_NETWORK_CLIENT_POLLING_TIMEOUT')}
+\b
 {config.describe('OCRD_NETWORK_SERVER_ADDR_PROCESSING')}
 \b
 {config.describe('OCRD_NETWORK_SERVER_ADDR_WORKFLOW')}
@@ -54,7 +58,7 @@
 {config.describe('OCRD_LOGGING_DEBUG')}
 """
 
 def command_with_replaced_help(*replacements):
 
     class CommandWithReplacedHelp(click.Command):
         def get_help(self, ctx):
@@ -66,14 +70,14 @@
 
     return CommandWithReplacedHelp
 
 from ocrd.cli.ocrd_tool import ocrd_tool_cli
 from ocrd.cli.workspace import workspace_cli
 from ocrd.cli.process import process_cli
 from ocrd.cli.bashlib import bashlib_cli
 from ocrd.cli.validate import validate_cli
 from ocrd.cli.resmgr import resmgr_cli
 from ocrd.decorators import ocrd_loglevel
 from .zip import zip_cli
 from .log import log_cli
 from .network import network_cli
 

src/ocrd_network/cli/client.py

@@ -1,41 +1,9 @@
 import click
-from http.server import BaseHTTPRequestHandler, HTTPServer
+from json import dumps
 from typing import Optional
-
-from ocrd.decorators import parameter_option
-from ocrd_network import Client
+from ocrd.decorators.parameter_option import parameter_option, parameter_override_option
 from ocrd_utils import DEFAULT_METS_BASENAME
-
-
-STOP_WAITING_CALLBACK = False
-
-
-class ClientCallbackHandler(BaseHTTPRequestHandler):
-    """
-    A simple callback handler for the network client to be invoked when the Processing Worker
-    sends requests to the `callback_url` set in the processing request submitted to the Processing Server.
-    """
-
-    def do_POST(self):
-        self.send_response(200)
-        self.send_header("Content-Type", "text/plain")
-        self.end_headers()
-        self.wfile.write("finished".encode("utf-8"))
-        len = int(self.headers.get("Content-Length", 0))
-        data = self.rfile.read(len).decode("utf-8")
-        # TODO: how should the callback-content be handled/printed
-        print(f"Processor finished: {data}")
-        global STOP_WAITING_CALLBACK
-        STOP_WAITING_CALLBACK = True
-
-
-class ClientCallbackServer(HTTPServer):
-    """
-    A simple http-server that listens for callbacks from the Processing Server/Worker.
-    """
-    def __init__(self):
-        super().__init__(server_address=("0.0.0.0", 0), RequestHandlerClass=ClientCallbackHandler)
-        self.callback_url = f"http://172.17.0.1:{self.server_address[1]}"
+from ..client import Client
 
 
 @click.group('client')
@@ -55,6 +23,34 @@ def discovery_cli():
     pass
 
 
+@discovery_cli.command('processors')
+@click.option('--address',
+              help='The address of the Processing Server. If not provided, '
+                   'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+def check_deployed_processors(address: Optional[str]):
+    """
+    Get a list of deployed processing workers/processor servers.
+    Each processor is shown only once regardless of the amount of deployed instances.
+    """
+    client = Client(server_addr_processing=address)
+    processors_list = client.check_deployed_processors()
+    print(dumps(processors_list, indent=4))
+
+
+@discovery_cli.command('processor')
+@click.option('--address',
+              help='The address of the Processing Server. If not provided, '
+                   'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.argument('processor_name', required=True, type=click.STRING)
+def check_processor_ocrd_tool(address: Optional[str], processor_name: str):
+    """
+    Get the json tool of a deployed processor specified with `processor_name`
+    """
+    client = Client(server_addr_processing=address)
+    ocrd_tool = client.check_deployed_processor_ocrd_tool(processor_name=processor_name)
+    print(dumps(ocrd_tool, indent=4))
+
+
 @client_cli.group('processing')
 def processing_cli():
     """
@@ -63,55 +59,98 @@ def processing_cli():
     pass
 
 
+@processing_cli.command('check-log')
+@click.option('--address',
+              help='The address of the Processing Server. If not provided, '
+                   'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-j', '--processing-job-id', required=True)
+def check_processing_job_status(address: Optional[str], processing_job_id: str):
+    """
+    Check the log of a previously submitted processing job.
+    """
+    client = Client(server_addr_processing=address)
+    response = client.check_job_log(job_id=processing_job_id)
+    print(response._content.decode(encoding='utf-8'))
+
+
+@processing_cli.command('check-status')
+@click.option('--address',
+              help='The address of the Processing Server. If not provided, '
+                   'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-j', '--processing-job-id', required=True)
+def check_processing_job_status(address: Optional[str], processing_job_id: str):
+    """
+    Check the status of a previously submitted processing job.
+    """
+    client = Client(server_addr_processing=address)
+    job_status = client.check_job_status(processing_job_id)
+    assert job_status
+    print(f"Processing job status: {job_status}")
+
+
 @processing_cli.command('processor')
 @click.argument('processor_name', required=True, type=click.STRING)
-@click.option('--address')
+@click.option('--address',
+              help='The address of the Processing Server. If not provided, '
+                   'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
 @click.option('-m', '--mets', required=True, default=DEFAULT_METS_BASENAME)
 @click.option('-I', '--input-file-grp', default='OCR-D-INPUT')
 @click.option('-O', '--output-file-grp', default='OCR-D-OUTPUT')
 @click.option('-g', '--page-id')
 @parameter_option
+@parameter_override_option
 @click.option('--result-queue-name')
 @click.option('--callback-url')
 @click.option('--agent-type', default='worker')
-def send_processing_request(
+@click.option('-b', '--block', default=False,
+              help='If set, the client will block till job timeout, fail or success.')
+def send_processing_job_request(
     address: Optional[str],
     processor_name: str,
     mets: str,
     input_file_grp: str,
     output_file_grp: Optional[str],
     page_id: Optional[str],
     parameter: Optional[dict],
+    parameter_override: Optional[dict],
     result_queue_name: Optional[str],
     callback_url: Optional[str],
     # TODO: This is temporally available to toggle
     #  between the ProcessingWorker/ProcessorServer
-    agent_type: Optional[str]
+    agent_type: Optional[str],
+    block: Optional[bool]
 ):
-    callback_server = ClientCallbackServer()
+    """
+    Submit a processing job to the processing server.
+    """
     req_params = {
         "path_to_mets": mets,
         "description": "OCR-D Network client request",
         "input_file_grps": input_file_grp.split(','),
-        "parameters": parameter if parameter else {},
-        "agent_type": agent_type,
-        "callback_url": callback_server.callback_url
+        "agent_type": agent_type
     }
     if output_file_grp:
         req_params["output_file_grps"] = output_file_grp.split(',')
     if page_id:
         req_params["page_id"] = page_id
+    if parameter:
+        if parameter == ['{}']:
+            req_params["parameters"] = {}
+        else:
+            req_params["parameters"] = parameter
+    if parameter_override:
+        req_params["parameters"] = parameter_override
     if result_queue_name:
         req_params["result_queue_name"] = result_queue_name
     if callback_url:
         req_params["callback_url"] = callback_url
-
     client = Client(server_addr_processing=address)
-    response = client.send_processing_request(processor_name=processor_name, req_params=req_params)
-    processing_job_id = response.get('job_id', None)
+    processing_job_id = client.send_processing_job_request(
+        processor_name=processor_name, req_params=req_params)
+    assert processing_job_id
     print(f"Processing job id: {processing_job_id}")
-    while not STOP_WAITING_CALLBACK:
-        callback_server.handle_request()
+    if block:
+        client.poll_job_status(job_id=processing_job_id)
 
 
 @client_cli.group('workflow')
@@ -122,6 +161,44 @@ def workflow_cli():
     pass
 
 
+@workflow_cli.command('check-status')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-j', '--workflow-job-id', required=True)
+def check_workflow_job_status(address: Optional[str], workflow_job_id: str):
+    """
+    Check the status of a previously submitted workflow job.
+    """
+    client = Client(server_addr_processing=address)
+    job_status = client.check_workflow_status(workflow_job_id)
+    assert job_status
+    print(f"Workflow job status: {job_status}")
+
+
+@workflow_cli.command('run')
+@click.option('--address', help='The address of the Processing Server. If not provided, '
+                                'the "OCRD_NETWORK_SERVER_ADDR_PROCESSING" env variable is used by default')
+@click.option('-m', '--path-to-mets', required=True)
+@click.option('-w', '--path-to-workflow', required=True)
+@click.option('-b', '--block', default=False,
+              help='If set, the client will block till job timeout, fail or success.')
+def send_workflow_job_request(
+    address: Optional[str],
+    path_to_mets: str,
+    path_to_workflow: str,
+    block: Optional[bool]
+):
+    """
+    Submit a workflow job to the processing server.
+    """
+    client = Client(server_addr_processing=address)
+    workflow_job_id = client.send_workflow_job_request(path_to_wf=path_to_workflow, path_to_mets=path_to_mets)
+    assert workflow_job_id
+    print(f"Workflow job id: {workflow_job_id}")
+    if block:
+        client.poll_workflow_status(job_id=workflow_job_id)
+
+
 @client_cli.group('workspace')
 def workspace_cli():
     """

src/ocrd_network/client.py

@@ -1,28 +1,62 @@
-from json import dumps, loads
-from requests import post as requests_post
 from ocrd_utils import config, getLogger, LOG_FORMAT
-
-from .constants import NETWORK_PROTOCOLS
+from .client_utils import (
+    get_ps_deployed_processors,
+    get_ps_deployed_processor_ocrd_tool,
+    get_ps_processing_job_log,
+    get_ps_processing_job_status,
+    get_ps_workflow_job_status,
+    poll_job_status_till_timeout_fail_or_success,
+    poll_wf_status_till_timeout_fail_or_success,
+    post_ps_processing_request,
+    post_ps_workflow_request,
+    verify_server_protocol
+)
 
 
 class Client:
-    def __init__(self, server_addr_processing: str = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING):
+    def __init__(
+        self,
+        server_addr_processing: str,
+        timeout: int = config.OCRD_NETWORK_CLIENT_POLLING_TIMEOUT,
+        wait: int = config.OCRD_NETWORK_CLIENT_POLLING_SLEEP
+    ):
         self.log = getLogger(f"ocrd_network.client")
+        if not server_addr_processing:
+            server_addr_processing = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING
         self.server_addr_processing = server_addr_processing
         verify_server_protocol(self.server_addr_processing)
+        self.polling_timeout = timeout
+        self.polling_wait = wait
+        self.polling_tries = int(timeout/wait)
+
+    def check_deployed_processors(self):
+        return get_ps_deployed_processors(ps_server_host=self.server_addr_processing)
+
+    def check_deployed_processor_ocrd_tool(self, processor_name: str):
+        return get_ps_deployed_processor_ocrd_tool(
+            ps_server_host=self.server_addr_processing, processor_name=processor_name)
+
+    def check_job_log(self, job_id: str):
+        return get_ps_processing_job_log(self.server_addr_processing, processing_job_id=job_id)
+
+    def check_job_status(self, job_id: str):
+        return get_ps_processing_job_status(self.server_addr_processing, processing_job_id=job_id)
+
+    def check_workflow_status(self, workflow_job_id: str):
+        return get_ps_workflow_job_status(self.server_addr_processing, workflow_job_id=workflow_job_id)
+
+    def poll_job_status(self, job_id: str) -> str:
+        return poll_job_status_till_timeout_fail_or_success(
+            ps_server_host=self.server_addr_processing, job_id=job_id, tries=self.polling_tries, wait=self.polling_wait)
+
+    def poll_workflow_status(self, job_id: str) -> str:
+        return poll_wf_status_till_timeout_fail_or_success(
+            ps_server_host=self.server_addr_processing, job_id=job_id, tries=self.polling_tries, wait=self.polling_wait)
+
+    def send_processing_job_request(self, processor_name: str, req_params: dict) -> str:
+        return post_ps_processing_request(
+            ps_server_host=self.server_addr_processing, processor=processor_name, job_input=req_params)
 
-    def send_processing_request(self, processor_name: str, req_params: dict):
-        req_url = f"{self.server_addr_processing}/processor/{processor_name}"
-        req_headers = {"Content-Type": "application/json; charset=utf-8"}
-        req_json = loads(dumps(req_params))
-        self.log.info(f"Sending processing request to: {req_url}")
-        self.log.debug(req_json)
-        response = requests_post(url=req_url, headers=req_headers, json=req_json)
-        return response.json()
-
-
-def verify_server_protocol(address: str):
-    for protocol in NETWORK_PROTOCOLS:
-        if address.startswith(protocol):
-            return
-    raise ValueError(f"Wrong/Missing protocol in the server address: {address}, must be one of: {NETWORK_PROTOCOLS}")
+    def send_workflow_job_request(self, path_to_wf: str, path_to_mets: str):
+        return post_ps_workflow_request(
+            ps_server_host=self.server_addr_processing, path_to_wf=path_to_wf, path_to_mets=path_to_mets)