Extend the network client

CHANGELOG.md

@@ -8,6 +8,17 @@ Versioned according to [Semantic Versioning](http://semver.org/).
 Changed:
 
   * ocrd_network: Use `ocrd-all-tool.json` bundled by core instead of download from website, #1257, #1260
+  * `ocrd network client processing processor` supports blocking behavior with `--block` by polling job status, #1265, #1269
+
+Added:
+
+  * `ocrd network client workflow check-status` to get the status of a workflow job, #1269
+  * `ocrd network client processing check-status` to get the status of a processing (processor) job, #1269
+  * `ocrd network client workflow run` Run, optionally blocking, a workflow on the processing server, #1265, #1269
+  * Environment variables `OCRD_NETWORK_CLIENT_POLLING_SLEEP` and `OCRD_NETWORK_CLIENT_POLLING_TIMEOUT` to control polling interval and timeout for `ocrd network client {processing processor,workflow run`, #1269
+  * `ocrd network client discovery processors` to list the processors deployed in the processing server, #1269
+  * `ocrd network client discovery processor` to get the `ocrd-tool.json` of a deployed processor, #1269
+  * `ocrd network client processing check-log` to retrieve the log data for a processing job, #1269
   * `ocrd workspace clone`/`Resolver.workspace_from_url`: with `clobber_mets=False`, raise a FileExistsError for existing mets.xml on disk, #563, #1268
   * `ocrd workspace find --download`: print the the correct, up-to-date field, not `None`, #1202, #1266
 

Makefile

@@ -286,7 +286,7 @@ network-module-test: assets
 INTEGRATION_TEST_IN_DOCKER = docker exec core_test
 network-integration-test:
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml up -d
-	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v --ignore-glob="$(TESTDIR)/network/*ocrd_all*.py"
+	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v --ignore-glob="tests/network/*ocrd_all*.py"
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml down --remove-orphans
 
 network-integration-test-cicd:

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,9 +1,11 @@
 import click
-from typing import Optional
-
-from ocrd.decorators import parameter_option
-from ocrd_network import Client
+from json import dumps
+from typing import List, Optional, Tuple
+from ocrd.decorators.parameter_option import parameter_option, parameter_override_option
 from ocrd_utils import DEFAULT_METS_BASENAME
+from ocrd_utils.introspect import set_json_key_value_overrides
+from ocrd_utils.str import parse_json_string_or_file
+from ..client import Client
 
 
 @click.group('client')
@@ -23,6 +25,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():
     """
@@ -31,56 +61,92 @@ def processing_cli():
     pass
 
 
-@processing_cli.command('processor')
+@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('run')
 @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(
-        address: Optional[str],
-        processor_name: str,
-        mets: str,
-        input_file_grp: str,
-        output_file_grp: Optional[str],
-        page_id: Optional[str],
-        parameter: 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]
+@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: List[str],
+    parameter_override: List[Tuple[str, str]],
+    result_queue_name: Optional[str],
+    callback_url: Optional[str],
+    # TODO: This is temporally available to toggle
+    #  between the ProcessingWorker/ProcessorServer
+    agent_type: Optional[str],
+    block: Optional[bool]
 ):
+    """
+    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,
+        "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
+    req_params["parameters"] = set_json_key_value_overrides(parse_json_string_or_file(*parameter), *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)
+    client = Client(server_addr_processing=address)
+    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}")
+    if block:
+        client.poll_job_status(job_id=processing_job_id)
 
 
 @client_cli.group('workflow')
@@ -91,6 +157,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,37 +1,63 @@
-from json import dumps, loads
-from requests import post as requests_post
+from typing import Optional
 from ocrd_utils import config, getLogger, LOG_FORMAT
+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
+)
 
-from .constants import NETWORK_PROTOCOLS
 
-
-# TODO: This is just a conceptual implementation and first try to
-#  trigger further discussions on how this should look like.
 class Client:
     def __init__(
         self,
-        server_addr_processing: str = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING,
-        server_addr_workflow: str = config.OCRD_NETWORK_SERVER_ADDR_WORKFLOW,
-        server_addr_workspace: str = config.OCRD_NETWORK_SERVER_ADDR_WORKSPACE
+        server_addr_processing: Optional[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
-        self.server_addr_workflow = server_addr_workflow
-        self.server_addr_workspace = server_addr_workspace
-
-    def send_processing_request(self, processor_name: str, req_params: dict):
         verify_server_protocol(self.server_addr_processing)
-        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}")
+        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_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)