Merge pull request #11 from gabrielbazan/development

Publication on FTP Repositories

README.md

@@ -329,7 +329,7 @@ You can publish processes using the _{host}/configuration/processes_ URI with th
 
 Processes parameters (inputs and outputs) have _dataType_ OR _format_, they cannot have both or none.
 
-Here you can see a metadata example for the [L0](/tests/cosmo/processes/L0.py) process:
+Here you can see a metadata example for the [L0](/tests/processes/cosmo/L0.py) process:
 ```json
 {
   "identifier": "L0",

lycheepy/configuration/Dockerfile

@@ -1,9 +1,12 @@
-FROM python:2
+FROM ubuntu:16.04
+
+RUN apt update && apt upgrade -y
 
 WORKDIR /root/
 
 ADD requirements.txt /root/
 
+RUN apt install python2.7 python-virtualenv -y
 RUN virtualenv venv
 RUN /root/venv/bin/pip install -r requirements.txt
 

lycheepy/executions/Dockerfile

@@ -1,9 +1,12 @@
-FROM python:2
+FROM ubuntu:16.04
+
+RUN apt update && apt upgrade -y
 
 WORKDIR /root/
 
 ADD requirements.txt /root/
 
+RUN apt install python2.7 python-virtualenv -y
 RUN virtualenv venv
 RUN /root/venv/bin/pip install -r requirements.txt
 

lycheepy/worker/Dockerfile

@@ -1,16 +1,21 @@
-FROM python:2
+FROM ubuntu:16.04
 
-ADD requirements.txt /root/
-ADD worker/ /root/worker/
-ADD wait-service.sh /root/
+RUN apt update && apt upgrade -y
 
 WORKDIR /root/
 
+ADD requirements.txt /root/
+
+RUN apt install python2.7 python-virtualenv -y
 RUN virtualenv venv
 RUN /root/venv/bin/pip install -r requirements.txt
 
-RUN /root/venv/bin/pip install -e git+https://github.com/geopython/pywps.git@master#egg=pywps-dev
-RUN /root/venv/bin/pip install flufl.enum
+RUN apt install git python-dev gcc -y
+RUN /root/venv/bin/pip install -e git+https://github.com/geopython/pywps.git@5efa0513fe096b8eb3b3a0399b22403209623ff3#egg=pywps-dev
+RUN /root/venv/bin/pip install flufl.enum pathlib
+
+ADD worker/ /root/worker/
+ADD wait-service.sh /root/
 
 RUN mkdir /root/worker/processes/
 RUN mkdir /root/workdir/

lycheepy/worker/worker/distribution/worker.py

@@ -1,4 +1,4 @@
-import uuid
+from uuid import uuid1
 from celery import Celery
 from pywps import Service
 from pywps.app.WPSRequest import WPSRequest
@@ -8,6 +8,9 @@
 from gateways.repository import RepositoryFactory
 from gateways.processes import ProcessesGateway
 
+import logging
+from json import dumps
+
 
 app = Celery(BROKER_APPLICATION_NAME)
 
@@ -25,18 +28,18 @@ def run_process(identifier, wps_request_json):
         PROCESSES_GATEWAY_DIRECTORY
     )
 
-    service = Service([processes.get_instance(identifier, LOCAL_PROCESSES_REPOSITORY)], 'the_cfg_file')
-
     request = WPSRequest()
     request.json = wps_request_json
     request.status = 'false'
 
-    response = service.processes.get(identifier).execute(
-        request,
-        uuid.uuid1()
-    )
+    logging.info('WPS request: {}'.format(dumps(wps_request_json)))
+
+    with processes.get_process_context(identifier, LOCAL_PROCESSES_REPOSITORY) as process_context:
+        service = Service([process_context.get_process_instance()], CONFIGURATION_FILE)
+        response = service.processes.get(identifier).execute(request, uuid1())
+        outputs = OutputsSerializer.to_json(response.outputs)
 
-    outputs = OutputsSerializer.to_json(response.outputs)
+    logging.info('Process outputs: {}'.format(dumps(outputs)))
 
     return dict(process=identifier, outputs=outputs)
 

lycheepy/worker/worker/gateways/repository/ftp.py

@@ -0,0 +1,38 @@
+from ftplib import FTP
+
+
+class Ftp(object):
+
+    def __init__(self, ip, username, password, timeout):
+        self.connection = FTP(ip, timeout=timeout)
+        self.connection.login(user=username, passwd=password)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, kind, value, tb):
+        self.close()
+
+    def close(self):
+        self.connection.quit()
+
+    def upload_file(self, local_path, remote_path):
+        self.connection.storbinary('STOR ' + remote_path, open(local_path, 'rb'))
+
+    def create_directory_if_not_exists(self, path):
+        if not self.directory_exists(path):
+            self.create_directory(path)
+
+    def directory_exists(self, path):
+        current_path = self.connection.pwd()
+        exists = True
+        try:
+            self.connection.cwd(path)
+        except Exception:
+            exists = False
+        finally:
+            self.connection.cwd(current_path)
+            return exists
+
+    def create_directory(self, path):
+        self.connection.mkd(path)

lycheepy/worker/worker/gateways/repository/ftp_repository.py

@@ -1,6 +1,14 @@
+from os.path import join, basename
 from repository import Repository
+from ftp import Ftp
 
 
 class FtpRepository(Repository):
 
-    def publish(self, name, raster_file): pass
+    def __init__(self, ip, username, password, timeout):
+        self.connection = Ftp(ip, username, password, timeout)
+
+    def publish(self, name, raster_file):
+        with self.connection:
+            self.connection.create_directory_if_not_exists(name)
+            self.connection.upload_file(raster_file, join(name, basename(raster_file)))

lycheepy/worker/worker/pywps.cfg

@@ -0,0 +1,4 @@
+
+
+[logging]
+level=INFO

lycheepy/worker/worker/settings.py

@@ -1,6 +1,8 @@
 from gateways.repository import Repositories
 
 
+CONFIGURATION_FILE = '/root/worker/pywps.cfg'
+
 BROKER_APPLICATION_NAME = 'lycheepy'
 BROKER_PROCESS_EXECUTION_TASK_NAME = 'run_process'
 BROKER_CHAIN_PROCESS_EXECUTION_TASK_NAME = 'run_chain_process'

lycheepy/wps/Dockerfile

@@ -1,14 +1,18 @@
-FROM python:2
+FROM ubuntu:16.04
+
+RUN apt update && apt upgrade -y
 
 WORKDIR /root/
 
 ADD requirements.txt /root/
 
+RUN apt install python2.7 python-virtualenv -y
 RUN virtualenv venv
 RUN /root/venv/bin/pip install -r requirements.txt
 
-RUN /root/venv/bin/pip install -e git+https://github.com/geopython/pywps.git@master#egg=pywps-dev
-RUN /root/venv/bin/pip install flufl.enum
+RUN apt install git python-dev gcc -y
+RUN /root/venv/bin/pip install -e git+https://github.com/geopython/pywps.git@5efa0513fe096b8eb3b3a0399b22403209623ff3#egg=pywps-dev
+RUN /root/venv/bin/pip install flufl.enum pathlib
 
 ADD wps/ /root/wps/
 ADD wait-service.sh /root/

lycheepy/wps/wps/pywps.cfg

@@ -33,6 +33,6 @@ maxprocesses=10
 parallelprocesses=20
 
 [logging]
-level=DEBUG
+level=INFO
 file=/root/logs/pywps.log
 database=sqlite:////root/logs/pywps-logs.sqlite3

tests/cosmo/settings.py → tests/cosmo_settings.py

@@ -1,17 +1,8 @@
-from os.path import join, dirname
+from os.path import join
+from settings import PROCESSES_DIRECTORY
 
 
-HOST = 'http://localhost'
-CONFIGURATION_URL = '{}/configuration'.format(HOST)
-PROCESSES_URL = '{}/processes'.format(CONFIGURATION_URL)
-CHAINS_URL = '{}/chains'.format(CONFIGURATION_URL)
-WPS_URL = '{}/wps/'.format(HOST)
-EXECUTIONS_URL = '{}/executions/executions'.format(HOST)
-CSW_URL = '{}/repository/geoserver/csw'.format(HOST)
-
-
-DIR = dirname(__file__)
-PROCESSES_DIRECTORY = join(DIR, 'processes')
+PROCESSES_DIRECTORY = join(PROCESSES_DIRECTORY, 'cosmo')
 
 
 PROCESSES = [
@@ -168,11 +159,3 @@
         L1C=['GEC']
     )
 )
-
-
-with open(join(DIR, 'execute_cosmo.xml'), 'r') as execute:
-    CHAIN_EXECUTE = execute.read()
-
-
-with open(join(DIR, 'csw_get_records.xml'), 'r') as csw:
-    CSW_GET_RECORDS = csw.read()

tests/gateways/__init__.py

@@ -0,0 +1,5 @@
+from processes import ProcessesGateway
+from chains import ChainsGateway
+from executions import ExecutionsGateway
+from wps import WpsGateway
+from csw import CswGateway

tests/gateways/chains.py

@@ -0,0 +1,27 @@
+from requests import get, post, put, delete
+
+
+class ChainsGateway(object):
+
+    def __init__(self, uri):
+        self.uri = uri
+
+    def list(self):
+        return get(self.uri)
+
+    def create(self, specification):
+        return post(self.uri, json=specification)
+
+    def get(self, identifier):
+        return get('{}/{}'.format(self.uri, identifier))
+
+    def update(self, identifier, specification, process_file):
+        pass
+
+    def delete(self, identifier):
+        return delete('{}/{}'.format(self.uri, identifier))
+
+    def exists(self, identifier):
+        response = get('{}?identifier__eq={}'.format(self.uri, identifier))
+        assert response.status_code == 200, 'Failed to query chain'
+        return response.json().get('results')