Merge 857138b into b5dc567

.coveragerc

@@ -14,3 +14,4 @@ show_missing = True
 
 omit =
     redis_consumer/pbs/*
+    redis_consumer/grpc_clients.py

.pylintrc

@@ -9,7 +9,7 @@
 
 # Add files or directories to the blacklist. They should be base names, not
 # paths.
-ignore=CVS
+ignore=CVS,redis_consumer/pbs
 
 # Pickle collected data for later comparisons.
 persistent=yes
@@ -28,7 +28,7 @@ unsafe-load-any-extension=no
 # A comma-separated list of package or module names from where C extensions may
 # be loaded. Extensions are loading into the active Python interpreter and may
 # run arbitrary code
-extension-pkg-whitelist=
+extension-pkg-whitelist=grpc._cython.cygrpc
 
 
 [MESSAGES CONTROL]

.travis.yml

@@ -1,37 +1,36 @@
-sudo: required
+sudo: false
 dist: trusty
+
+git:
+  depth: false
+
 language: python
-matrix:
-  include:
-    - python: 2.7
+
+python:
+  - 2.7
+  - 3.5
+  - 3.6
+
+cache: pip
+
 install:
-  # source: http://conda.pydata.org/docs/travis.html
-  # We do this conditionally because it saves us some downloading if the
-  # version is the same.
-  - if [[ "$TRAVIS_PYTHON_VERSION" == "2.7" ]]; then
-      wget https://repo.continuum.io/miniconda/Miniconda-latest-Linux-x86_64.sh -O miniconda.sh;
-    else
-      wget https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh -O miniconda.sh;
-    fi
-
-  - bash miniconda.sh -b -p $HOME/miniconda
-  - export PATH="$HOME/miniconda/bin:$PATH"
-  - hash -r
-  - conda config --set always_yes yes --set changeps1 no
-  - conda update -q conda
-  # Useful for debugging any issues with conda
-  - conda info -a
-
-  - conda create -q -n test-environment python=$TRAVIS_PYTHON_VERSION numpy pytest
-  - source activate test-environment
-  - pip install pytest pytest-pep8 pytest-cov coveralls
   - pip install -r requirements.txt
+  - pip install pytest pytest-cov==2.5.1 pytest-pep8 coveralls
 
-  # set library path
-  - export LD_LIBRARY_PATH=$HOME/miniconda/envs/test-environment/lib/:$LD_LIBRARY_PATH
-
-# command to run tests
 script:
-  - PYTHONPATH=$PWD:$PYTHONPATH python -m pytest --cov=redis_consumer --pep8 redis_consumer tests
+  - python -m pytest --cov=redis_consumer --pep8 redis_consumer tests
+
+jobs:
+  include:
+    - stage: deploy
+      if: (branch = master OR branch =~ ^r[0-9]+.[0-9]+) AND type != pull_request
+      script:
+      - echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
+      - docker build -t "$TRAVIS_REPO_SLUG" .
+      - docker images
+      - docker tag "$TRAVIS_REPO_SLUG" "$TRAVIS_REPO_SLUG":latest
+      - if [[ "$TRAVIS_BRANCH" != "master" ]] ; then docker tag "$TRAVIS_REPO_SLUG" "$TRAVIS_REPO_SLUG":"${TRAVIS_BRANCH#r}" ; fi
+      - docker push "$TRAVIS_REPO_SLUG"
+
 after_success:
   - coveralls

consume-redis-events.py

@@ -58,6 +58,15 @@ def initialize_logger(debug_mode=False):
     logger.addHandler(console)
 
 
+def get_consumer(consumer_type, **kwargs):
+    ct = str(consumer_type).lower()
+    if ct == 'image':
+        return consumers.ImageFileConsumer(**kwargs)
+    if ct == 'zip':
+        return consumers.ZipFileConsumer(**kwargs)
+    raise ValueError('Invalid `consumer_type`: "{}"'.format(consumer_type))
+
+
 if __name__ == '__main__':
     initialize_logger(settings.DEBUG)
 
@@ -71,10 +80,13 @@ def initialize_logger(debug_mode=False):
 
     storage_client = storage.get_client(settings.CLOUD_PROVIDER)
 
-    consumer = consumers.PredictionConsumer(
-        redis_client=redis,
-        storage_client=storage_client,
-        final_status='done')
+    kwargs = {
+        'redis_client': redis,
+        'storage_client': storage_client,
+        'final_status': 'done'
+    }
+
+    consumer = get_consumer(settings.CONSUMER_TYPE, **kwargs)
 
     while True:
         try:

protos/function.proto

@@ -0,0 +1,12 @@
+syntax = "proto3";
+
+package tensorflow.serving;
+option cc_enable_arenas = true;
+
+// Metadata for an inference request such as the processing name and type
+message FunctionSpec {
+  // Required function name.
+  string name = 1;
+  // Required function type.
+  string type = 2;
+}

protos/prediction_service.proto

@@ -10,4 +10,4 @@ import "predict.proto";
 service PredictionService {
   // Predict -- provides access to loaded TensorFlow model.
   rpc Predict(PredictRequest) returns (PredictResponse);
-}
+}

protos/process.proto

@@ -0,0 +1,78 @@
+syntax = "proto3";
+
+package tensorflow.serving;
+option cc_enable_arenas = true;
+
+import "tensor.proto";
+import "function.proto";
+
+// ProcessRequest specifies which TensorFlow model to run, as well as
+// how inputs are mapped to tensors and how outputs are filtered before
+// returning to user.
+message ProcessRequest {
+  // Model Specification.
+  FunctionSpec function_spec = 1;
+
+  // Input tensors.
+  // Names of input tensor are alias names. The mapping from aliases to real
+  // input tensor names is expected to be stored as named generic signature
+  // under the key "inputs" in the model export.
+  // Each alias listed in a generic signature named "inputs" should be provided
+  // exactly once in order to run the processing.
+  map<string, TensorProto> inputs = 2;
+
+  // Output filter.
+  // Names specified are alias names. The mapping from aliases to real output
+  // tensor names is expected to be stored as named generic signature under
+  // the key "outputs" in the model export.
+  // Only tensors specified here will be run/fetched and returned, with the
+  // exception that when none is specified, all tensors specified in the
+  // named signature will be run/fetched and returned.
+  repeated string output_filter = 3;
+}
+
+// Response for ProcessRequest on successful run.
+message ProcessResponse {
+  // Output tensors.
+  map<string, TensorProto> outputs = 1;
+}
+
+message ChunkedProcessRequest {
+  // Model Specification.
+  FunctionSpec function_spec = 1;
+
+  // Input tensors.
+  // Names of input tensor are alias names. The mapping from aliases to real
+  // input tensor names is expected to be stored as named generic signature
+  // under the key "inputs" in the model export.
+  // Each alias listed in a generic signature named "inputs" should be provided
+  // exactly once in order to run the processing.
+  map<string, bytes> inputs = 2;
+
+  // Output filter.
+  // Names specified are alias names. The mapping from aliases to real output
+  // tensor names is expected to be stored as named generic signature under
+  // the key "outputs" in the model export.
+  // Only tensors specified here will be run/fetched and returned, with the
+  // exception that when none is specified, all tensors specified in the
+  // named signature will be run/fetched and returned.
+  repeated string output_filter = 3;
+
+  // Shape of chunked array.
+  repeated int64 shape = 4;
+
+  // Dtype of chunked array.
+  string dtype = 5;
+}
+
+// Response for ChunkedProcessRequest on successful run.
+message ChunkedProcessResponse {
+  // Output tensors.
+  map<string, bytes> outputs = 1;
+
+  // Shape of chunked array.
+  repeated int64 shape = 4;
+
+  // Dtype of chunked array.
+  string dtype = 5;
+}

protos/processing_service.proto

@@ -0,0 +1,13 @@
+syntax = "proto3";
+
+package tensorflow.serving;
+option cc_enable_arenas = true;
+
+import "process.proto";
+
+// ProcessingService provides access to data processing functions
+service ProcessingService {
+  // Process -- provides access to a data processing function
+  rpc Process(ProcessRequest) returns (ProcessResponse);
+  rpc StreamProcess(stream ChunkedProcessRequest) returns (stream ChunkedProcessResponse);
+}

redis_consumer/__init__.py

@@ -27,8 +27,7 @@
 from __future__ import division
 from __future__ import print_function
 
-from redis_consumer import grpc_client
-from redis_consumer import processing
+from redis_consumer import grpc_clients
 from redis_consumer import consumers
 from redis_consumer import settings
 from redis_consumer import storage