Add Capo profile for serviceUrls for capability and workflow. Advance on...

Add Capo profile for serviceUrls for capability and workflow. Advance on getting capability_launcher working.

services/capability/development.ini

@@ -10,7 +10,7 @@ session.auto = true
 
 [server:main]
 use = egg:waitress#main
-listen = 0.0.0.0:3456
+listen = 0.0.0.0:3457
 
 
 [loggers]

services/capability/src/capability/capability_launcher.py

@@ -5,9 +5,13 @@ Sends a request for a capability execution to the capability service
 """
 import json
 import argparse
+import sys
+
 import requests
+from channels.amqp_helpers import Channel, CapabilityEventChannel
+from pycapo import CapoConfig
 
-from workspaces.schema import Capability
+from workspaces.schema import Capability, CapabilityEvent
 from ._version import ___version___ as version
 
 
@@ -32,23 +36,38 @@ def _make_arg_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         description=description, formatter_class=argparse.RawTextHelpFormatter
     )
-    parser.add_argument(
-        "-c",
-        "--capability",
-        action="store",
-        required=True,
-        help="file that contains a capability definition",
-    )
+    parser.add_argument("capability", action='store', nargs=1,
+                        help='Name of the capability to run')
 
     return parser
 
 
+def receive_capability_event(event: CapabilityEvent):
+    print(event)
+    if event.complete:
+        sys.exit(0)
+
+
 def main():
     args = _make_arg_parser().parse_args()
-    capability = Capability.from_file(args.capability)
+
+    settings = CapoConfig().settings('edu.nrao.archive.workspaces.CapabilitySettings')
+
     # Submit HTTP capability request to server
-    r = requests.post("http://0.0.0.0:3456/capability/request", json=json.dumps(capability.to_dict()))
-    # Get answer back and do something with that
+    r = requests.post(f"{settings.serviceUrl}/capability/request",
+                      json=json.dumps({'capability': args.capability}))
+
+    # we should get back a request ID; if we do, submit it and obtain
+    # an execution ID
+    request_id = r.json()['id']
+    r = requests.post(f'{settings.serviceUrl}/capability/request/{request_id}/submit')
+    execution_id = r.json()['id']
+
+    # at this point we can start listening for events from rabbitmq concerning
+    # this execution
+    with Channel(CapabilityEventChannel()) as channel:
+        channel.listen(receive_capability_event,
+                       pattern=f'request-{request_id}.execution-{execution_id}.#')
 
 
 if __name__ == "__main__":

services/capability/src/capability/server.py

 import json
 
-import requests
+import pyramid
 import zope.sqlalchemy
 from pyramid.config import Configurator
 from pyramid.renderers import JSONP
+from pyramid.request import Request
 from pyramid.view import view_config, view_defaults
 from pyramid_beaker import session_factory_from_settings
 from workspaces.services import get_session_factory, get_engine, CapabilityInfo, \
     CapabilityService
-from workspaces.schema import Capability
+from workspaces.schema import Capability, CapabilityRequest
 
 
 # ---------------------------------------------------------
@@ -24,14 +25,14 @@ def lookup_request(request):
 
 @view_defaults(route_name='capability_request', renderer='json')
 class CapabilityRestService:
-    def __init__(self, request: requests.Request):
-        self.http_request = request
+    def __init__(self, request: Request):
+        self.request = request
 
     @view_config(request_method="POST", renderer='json')
     def create(self):
         # create a capability request for this ... request
-        capability = Capability.from_json(self.http_request.json)
-        request = capability.create_request()
+        req = self.request.json_body
+        request = self.request.capabilities.create_request(req.capability_name)
         return request
 
     @view_config(request_method='POST', route_name='submit_capability_request', renderer='json')

shared/channels/src/channels/amqp_helpers.py

@@ -51,6 +51,7 @@ class WorkflowEventChannel(ChannelDefinition[WorkflowEvent]):
     def exchange(self) -> str:
         return self.EXCHANGE
 
+
 class CapabilityEventChannel(ChannelDefinition[CapabilityEvent]):
     # TODO: find out real value
     C_EXCHANGE = 'workspaces.capability-service.capability-status'

shared/workspaces/src/workspaces/schema.py

@@ -118,11 +118,20 @@ class CapabilityEvent:
     """
     # TODO: determine shape of Capability Event information
 
-    def __init__(self):
-        raise NotImplementedError
+    @property
+    def request_id(self):
+        return self.json['request_id']
 
-    def json(self) -> str:
-        raise NotImplementedError
+    @property
+    def execution_id(self):
+        return self.json['execution_id']
+
+    @property
+    def complete(self):
+        return self.json['is_complete']
+
+    def __init__(self, json):
+        self.json = json
 
     def __str__(self):
         return f'<CapabilityEvent with data{self.json()}>'
@@ -131,7 +140,7 @@ class CapabilityEvent:
         return f'<CapabilityEvent {self.__dict__}>'
 
     def __eq__(self, other):
-        raise NotImplementedError
+        return self.json == other.json
 
 
 Base = declarative_base(metaclass=abc.ABCMeta)

shared/workspaces/src/workspaces/services.py

@@ -34,8 +34,8 @@ class CapabilityService(CapabilityServiceIF):
 
     def create_request(self,
                        capability_name: str,
-                       parameters: List[ParameterIF],
-                       products: List[FutureProductIF]) -> "CapabilityRequestIF":
+                       parameters: List[ParameterIF]=None,
+                       products: List[FutureProductIF]=None) -> "CapabilityRequestIF":
         self.capability_info.create_capability_request(capability_name, parameters, products)
 
     def run_capability(self, request: CapabilityRequest) -> CapabilityExecution: