elibixby

class ProtocolEnum(type):

  registry: Dict[Type[Protocol], Set[Type[Protocol]] = defaultdict(set)

  def __new__(cls, name, bases, dct):
      for base in bases:
          if issubclass(base, Protocol) and not base is Protocol:
              registry[base].add(cls)

elibixby

[tool.poetry]
name = "registry"
version = "0.1.0"
description = ""
authors = ["Eli Bixby <eli@cradle.bio>"]

[tool.poetry.dependencies]
python = ">=3.9,<3.11"

[tool.poetry.group.git.dependencies]

elibixby

  def dense_to_sparse(t): 
    num_elements = tf.reduce_prod(tf.shape(t)) 
    def _index_dim(dim): 
      return tf.reshape( 
        tf.tile( 
          tf.expand_dims(tf.range(0, dim), -1), 
          multiples=[1, num_elements/dim] 
        ), 
        [-1] 
      ) 

elibixby

class MyEstimator(TowerEstimator, DNNClassifier):
  pass


def my_optimizer_fn(grads_and_vars, params):
  optimizer = AdagradOptimizer(learning_rate=params.learning_rate, momentum=params.momentum)
  return optimizer.apply_gradients(grads_and_vars)

elibixby

import tensorflow as tf


def make_test_data():
    return [ 
        tf.train.SequenceExample(
            feature_lists=tf.train.FeatureLists(
                feature_list={
                    'chars': tf.train.FeatureList(
                        feature=[

elibixby

elibixby@elibixby-4:~/repos/GoogleCloudPlatform/python-docs-samples$ nox -e "py35(sample='./ml_engine/online_prediction')"
nox > Running session py35(sample='./ml_engine/online_prediction')
nox > The virtualenv name was hashed to avoid being too long.
nox > virtualenv /usr/local/google/home/elibixby/repos/GoogleCloudPlatform/python-docs-samples/.nox/012715f6 -p python3.5
nox > pip install --upgrade -r testing/requirements.txt
nox > pip install --upgrade git+https://github.com/GoogleCloudPlatform/python-repo-tools.git
nox > chdir ./ml_engine/online_prediction
nox > pip install --upgrade -r requirements.txt
nox > pytest --cov --cov-config /usr/local/google/home/elibixby/repos/GoogleCloudPlatform/python-docs-samples/.coveragerc --cov-report term
=============================================================================================== test session starts ===============================================================================================

elibixby

(tf1.0) elibixby@elibixby-4:~/repos/GoogleCloudPlatform/python-docs-samples$ nox ls
nox > Running session check_requirements
nox > The virtualenv name was hashed to avoid being too long.
nox > virtualenv /usr/local/google/home/elibixby/repos/GoogleCloudPlatform/python-docs-samples/.nox/43fc5920
nox > pip install --upgrade git+https://github.com/GoogleCloudPlatform/python-repo-tools.git
nox > gcprepotools check-requirements ./spanner/cloud-client/requirements.txt
/usr/local/google/home/elibixby/repos/GoogleCloudPlatform/python-docs-samples/.nox/43fc5920/local/lib/python2.7/site-packages/requests/packages/urllib3/util/ssl_.py:334: SNIMissingWarning: An HTTPS request has been made, but the SNI (Subject Name Indication) extension to TLS is not available on this platform. This may cause the server to present an incorrect TLS certificate, which can cause validation failures. You can upgrade to a newer version of Python to solve this. For more information, see https://urllib3.readthedocs.io/en/latest/advanced-usage.

elibixby

def rolling_window(tensor, dtype, shape, capacity=None):
  with tf.name_scope('rolling_window'):
    window_size = shape[0]
    if capacity is None:
      capacity = shape[0] * 2 + 1 

    q = tf.FIFOQueue(capacity, [dtype], shapes=[shape[1:]])
    enqueue = q.enqueue_many(tensor)
    tf.train.add_queue_runner(
        tf.train.QueueRunner(queue=q, enqueue_ops=[enqueue])

elibixby

# Copyright 2016 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,

elibixby

def delete_tmp_files(credentials, tmp_dir):
    logging.info('Deleting temporary files')
    parsed = urlparse.urlparse(tmp_dir)
    client = storage.Client(credentials=credentials)
    bucket = client.bucket(parsed.netloc)
    print(parsed.path)
    blobs = bucket.list_blobs(prefix=parsed.path[1:])
    print(list(blobs))
    bucket.delete_blobs(list(blobs), client=client)