text:
"""Storage Engine for interfacing with a cluster of Redis servers.

TODO: Consider limiting batch sizes on things like Hash puts.
"""

import bisect
from collections import defaultdict
from contextlib import contextmanager
import itertools
import logging
import random
import traceback

import gevent
import redis
from redis.exceptions import ConnectionError
from redis.sentinel import Sentinel

from taba import client
from taba.server.storage.redis_connection_pool import (
    BlockingSentinelMasterGeventConnectionPool)
from taba.server.storage.util import CompoundOperation
from taba.server.storage.util import Operation
from taba.util.thread_util import YieldByCount

LOG = logging.getLogger(__name__)

# The number of times to retry a transaction due to a WatchError before giving
# up entirely and raising an exception. This should be set very high, since
# separate shards are independent transactions (i.e. a shard can fail, while
# other shards committed their changes, leaving the DB in an inconsistent
# state)
MAX_TX_RETRIES = 20

# The Check-and-Set batch size. This controls the number of keys which will be
# locked, retrieved, and updated in a single transaction.
CAS_BATCH_SIZE = 20
CAS_PARALLEL_REQUESTS = 5

# The Get batch size. This controls the number of keys which will be retrieved
# per pipelined Get call on a single shard.
GET_BATCH_SIZE = 1000

# The timeout for a connection in seconds.  The current redis client will
# attempt to connect twice before giving up, so the effective timeout will be
# 2x this.
CONNECTION_TIMEOUT = 5.000

# The amount of time to sleep between attempts to acquire a lock.
LOCK_WAIT_SLEEP = 0.010

class RedisServerEndpoint(object):
  """Redis Server end-point specification struct"""

  def __init__(self, shard_name, vbucket_start, vbucket_end):
    """
    Args:
      shard_name - Sentinel name of the Redis shard.
      vbucket_start - First Virtual Bucket in the range the end-point handles.
      vbucket_end - Last Virtual Bucket in the range the end-point handles.
    """
    self.shard_name = shard_name
    self.vbucket_start = vbucket_start
    self.vbucket_end = vbucket_end

class RedisEngine(object):
  """Class for interfacing with a cluster of Redis servers.

  Values are transparently distributed across the cluster using Virtual Buckets
  on the hash of the key. Keys are hashed into one of a large number of Virtual
  Buckets. These buckets are assigned in ranges to specific shards. The cluster
  can be re-sharded by reassigning the Virtual Bucket ranges.
  """

  def __init__(self,
      sentinels,
      endpoints,
      num_vbuckets,
      pool_size=2,
      timeout=CONNECTION_TIMEOUT):
    """
    Args:
      sentinels - List of sentinel connection info.
      endpoints - List of RedisServerEndpoint objects.
      num_vbuckets - The total number of configured Virtual Buckets.
      pool_size - The number of connections in each connection pool for each
          end-point.
      timeout - Socket connection timeout.
    """
    self.num_vbuckets = num_vbuckets
    self.num_endpoints = len(endpoints)

    # Sort the end-points by vbucket start, and make sure all the vbuckets
    # are accounted for.
    endpoints = sorted(endpoints, key=lambda e: e.vbucket_start)
    if endpoints[0].vbucket_start != 0:
      raise ValueError('First Virtual Bucket must be 0')

    if endpoints[-1].vbucket_end != self.num_vbuckets - 1:
      raise ValueError('Last Virtual Bucket does not match total')

    for i in xrange(0, len(endpoints) - 2):
      if endpoints[i].vbucket_end + 1 != endpoints[i + 1].vbucket_start:
        raise ValueError(
            'Virtual Bucket range mismatch between end-points %d and %d' %
            (i, i + 1))

    self.sentinel = Sentinel(sentinels, socket_timeout=timeout)

    self.shards = []
    self.vbucket_starts = []
    for endpoint in endpoints:
      endpoint_client = self.sentinel.master_for(
          endpoint.shard_name,
          connection_pool_class=BlockingSentinelMasterGeventConnectionPool,
          pool_size=pool_size)

      self.shards.append(endpoint_client)
      self.vbucket_starts.append(endpoint.vbucket_start)

  def ShutDown(self):
    """Called when the process is shutting-down. Any necessary clean-up actions
    should be performed here. (Needed to conform to the StorageEngine API).
    """
    # Disconnect each of the redis connections in each of the connection pools.
    for shard in self.shards:
      shard.connection_pool.shutdown()

  ###################################################################
  # SIMPLE / STRING / INTEGER
  ###################################################################

  def Get(self, key):
    """Return the value for a given key from the correct shard.

    Args:
      key - The key to lookup.

    Returns:
      CompoundOperation object with the result of the lookup. The response_value
      field contains the value pertaining to the key.
    """
    # Resubmit the Get operation as as Batch Get.
    op = self.BatchGet([key])

    # Unpack the batch operation results.
    if op.success and op.response_value:
      op.response_value = op.response_value[0]

    return op

  def BatchGet(self, keys, batch_size=GET_BATCH_SIZE):
    """Return the values for a set of keys, from across all shards.

    Args:
      keys - List of keys to lookup.
      batch_size - Get keys per shard in batches of this size.

    Returns:
      Operation with the results of the lookups. The response_value
      field has the form [value1, value2, ...], ordered by the list of keys
      that were looked up.
    """
    def _ShardGet(shard, keys, vkeys, values):
      op = CompoundOperation()
      for i in xrange(0, len(vkeys), batch_size):
        response_value = shard.mget([vkey for vkey in vkeys[i:i + batch_size]])
        op.AddOp(Operation(success=True, response_value=response_value))

      op.response_value = list(itertools.chain.from_iterable(
          (sub_op.response_value for sub_op in op.sub_operations)))
      return op

    return self._ShardedOp([(key, None) for key in keys], _ShardGet)

  def Put(self, key, value, expiration_time=None):
    """Put a value in a key under the correct shard.

    Args:
      key - The key to put.
      value - The value to put.
      expiration_time - If specified, set the key to expire after this many
                        seconds.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field is set to whether or not the Put was successful.
    """
    op = self.BatchPut([(key, value)], expiration_time=expiration_time)

    if op.success and op.response_value:
      op.response_value = op.response_value[0]

    return op

  def BatchPut(self, key_value_tuples, expiration_time=None):
    """Put a batch of values into keys, each under the correct shard.

    Args:
      key_value_tuples - A list of tuples of the form (key, value)
      expiration_time - If specified, set the keys to expire after this many
                        seconds.

    Returns:
      A CompoundOperation with the results of the queries. The response_value
      field is set to a list of boolean values [success1, success2,...],
      ordered by the list of (key, val) tuples that were Put.
    """
    def _ShardPut(shard, keys, vkeys, values):
      if (expiration_time == None) or (expiration_time <= 0):
        successes = [shard.mset(dict(zip(vkeys, values)))]
      else:
        pipe = shard.pipeline()
        map(pipe.setex, vkeys, [expiration_time for _ in vkeys], values)
        successes = pipe.execute()

      return Operation(success=True, response_value=successes)

    return self._ShardedOp(key_value_tuples, _ShardPut)

  def PutNotExist(self, key, value):
    """Put a value into a key, only if the value does not already exist.

    Args:
      key - The key to put.
      value - The value to put.

    Returns:
      Operation object. On success, the response_value will contain the number
      of keys modified (i.e. 1 if the key was set, 0 if it already existed)
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.setnx(vkey, value)

    return Operation(success=True, response_value=result)

  def Increment(self, key):
    """Atomically increment an integer.

    Args:
      key - The key of the integer to increment.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.incr(vkey)

    return Operation(success=True, response_value=result)

  def Delete(self, key):
    """Delete a key.

    Args:
      key - The key to delete.

    Returns:
      An Operation object with the results of the query. The response value
      field is set to whether or not the key was deleted successfully.
    """
    op = self.BatchDelete([key])

    if op.success and op.response_value:
      op.response_value = op.response_value[0]

    return op

  def BatchDelete(self, keys):
    """Delete a set of keys.

    Args:
      keys - A list of keys to delete.

    Returns:
      A CompoundOperation with the results of the queries. The response_value
      field is set to a list of boolean values [success1, success2,...],
      ordered by the list of keys that were deleted.
    """
    def _ShardDelete(shard, keys, vkeys, values):
      successes = [shard.delete(*vkeys)]
      return Operation(success=True, response_value=successes)

    return self._ShardedOp([(key, None) for key in keys], _ShardDelete)

  def BatchCheckAndMultiSet(self, keys, get_updates_fn):
    """Set a batch of keys transactionally using optimistic locking. Each key is
    locked and retrieved from its shard. The retrieved values are passed, one at
    a time, to the callback function. The return value of the callback is then
    put in the key, under the appropriate shard. If the value changed between
    the time it was locked and when the Put happens, the Put will fail, and the
    whole operation will be retried, until it succeeds.

    Args:
      keys - List of keys to update transactionally.
      get_updates_fn - Callable which, given a key and a value, returns the
          updated value for that key, which will be put back in the DB. If a put
          fails due to a lock error, this function will be called again with the
          updated value.

    Returns:
      A CompountOperation with the result of the queries.
    """
    _ShardCheckAndSet = _ShardCheckAndSetCallable(self, get_updates_fn)
    op = self._ShardedOp([(key, None) for key in keys], _ShardCheckAndSet)
    return op

  ###################################################################
  # SETS
  ###################################################################

  def SetAdd(self, key, values):
    """Add values to the Set under key.

    Args:
      key - The key of the Set.
      values - List of values to add to the Set.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field is an integer denoting the number of new items added
      to the Set.
    """
    _, shard, vkey = self._GetShardInfo(key)
    num_new_values = shard.sadd(vkey, *values)

    return Operation(success=True, response_value=num_new_values)

  def SetMembers(self, key):
    """Retrieve all the members of the Set at key.

    Args:
      key - The key of the Set.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field contains a set() object with the members of the Set.
    """
    _, shard, vkey = self._GetShardInfo(key)
    members = shard.smembers(vkey)

    return Operation(success=True, response_value=set(members))

  def SetIsMember(self, key, value):
    """Test whether a value is a member of the Set at key.

    Args:
      key - The key of the Set.
      value - The String value to test membership in the Set.

    Returns:
      A CompoundOperation object with the result of the query.  The
      response_value field contains a boolean indicating whether the value is
      in the Set.
    """
    _, shard, vkey = self._GetShardInfo(key)
    is_member = shard.sismember(vkey, value)

    return Operation(success=True, response_value=is_member)

  def SetRemove(self, key, values):
    """Remove values from the Set at key.

    Args:
      key - The key of the Set.
      values - List of values to remove from the Set.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field contains an integer denoting the number of
      items removed from the Set.
    """
    _, shard, vkey = self._GetShardInfo(key)
    num_removed = shard.srem(vkey, *values)

    return Operation(success=True, response_value=num_removed)

  def SetRemoveConditional(self, key, values, conditional_fn):
    """Remove values from the Set at key, but only if an external condition is
    met. The conditional_fn will be called after a transaction is started, but
    before it is committed. If the function returns False, the transaction is
    aborted. This allows arbitrary coordination (e.g. cross-shard conditions).

    Args:
      key - The key of the Set.
      values - List of values to remove from the Set.
      conditional_fn - Callable to be invoked within the transaction. If it
          returns True, the transaction will be committed; otherwise it will be
          aborted.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field contains an integer denoting the number of
      items removed from the Set, or if the transaction is aborted, the value
      False.
    """
    _, shard, vkey = self._GetShardInfo(key)

    def _TransactionBody(pipe):
      pipe.watch(vkey)

      proceed = conditional_fn()

      if proceed:
        pipe.multi()
        pipe.srem(vkey, *values)
        return pipe.execute()

      else:
        return [False]

    op = TransactionSkeleton(shard, _TransactionBody)

    return Operation(
        success=all([op.success, op.response_value, op.response_value[0]]),
        response_value=op.response_value)

  def SetPop(self, key):
    """Pop (remove and return) a random value from the Set at key.

    Args:
      key - The key of the Set.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field contains the value that was popped from the Set.
    """
    _, shard, vkey = self._GetShardInfo(key)
    popped = shard.spop(vkey)

    return Operation(success=True, response_value=popped)

  def SetSize(self, key):
    """Returns the size (cardinality) of the set stored at key.

    Args:
      key - The key of the Set.

    Returns:
      A CompoundOperation object with the result of the query. The
      response_value field contains an integer denoting the size of the set
      stored at key.
    """
    _, shard, vkey = self._GetShardInfo(key)
    size = shard.scard(vkey)

    return Operation(success=True, response_value=size)

  ###################################################################
  # HASH
  ###################################################################

  def HashGet(self, key, field):
    """Retrieve the value stored at a field in a Hash.

    Args:
      key - The key of the Hash.
      field - Field in the Hash to retrieve.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.hget(vkey, field)

    return Operation(success=True, response_value=result)

  def HashMultiGet(self, key, fields):
    """Retrieve a list of fields from a Hash.

    Args:
      key - The key of the Hash.
      fields - List of fields to retrieve.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    values = shard.hmget(vkey, fields)

    hash_values = dict(itertools.izip(fields, values))
    return Operation(success=True, response_value=hash_values)

  def HashFields(self, key):
    """Retrieve the field names of a Hash.

    Args:
      key - The key of the Hash.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.hkeys(vkey)

    return Operation(success=True, response_value=result)

  def HashGetAll(self, key):
    """Retrieve an entire Hash.

    Args:
      key - The key of the Hash.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.hgetall(vkey)

    return Operation(success=True, response_value=result)

  def HashPut(self, key, field, value):
    """Put a value to a field of a Hash.

    Args:
      key - The key of the Hash.
      field - Field in the Hash to set.
      value - Value to set.

    Returns:
      An Operation object with the result of the query.
    """
    return self.HashMultiPut(key, {field: value})

  def HashMultiPut(self, key, mapping):
    """Put a set of fields and values to a Hash.

    Args:
      key - The key of the Hash.
      mapping - Dict of keys and values to set.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    response = shard.hmset(vkey, mapping)

    return Operation(success=True, response_value=response)

  def HashReplace(self, key, mapping):
    """Completely replace the values of a Hash.

    Args:
      key - The key of the Hash.
      mapping - Dictionary of keys and values to set.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)

    pipe = shard.pipeline()
    pipe.delete(vkey)
    pipe.hmset(vkey, mapping)
    response = pipe.execute()

    return Operation(success=True, response_value=response)

  def HashSize(self, key):
    """Get the number of fields in a Hash.

    Args:
      key - The key of the Hash,

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    response = shard.hlen(vkey)

    return Operation(success=True, response_value=response)

  def HashDelete(self, key, field):
    """Delete a field from a Hash.

    Args:
      key - The key of the Hash.
      field - Field to delete from the Hash.

    Returns:
      An Operation object with the result of the query.
    """
    return self.HashBatchDelete(key, [field])

  def HashBatchDelete(self, key, fields):
    """Delete a list field from a Hash.

    Args:
      key - The key of the Hash.
      fields - Fields to delete from the Hash.

    Returns:
      An Operation object with the result of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)
    removed = shard.hdel(vkey, *fields)

    return Operation(success=True, response_value=removed)

  def HashCheckAndMultiSet(self, key, fields, get_updates_fn):
    """In a transaction, retrieve a (full or partial) Hash, callback to get
    updated values, and set the updates. If the underlying Hash changes in the
    process, the transaction is retried with the new values.

    Args:
      key - the key of the Hash.
      fields - Fields to retrieve from the Hash. if None, all the fields are
          retrieved.
      get_updates_fn - Callable which accepts the current value of the Hash
          fields requested, and returns a Dict of {field: value} updates to
          apply back to the Hash. Note that the results do not have to be
          limited the the Hash fields that were retrieved. If the callback
          returns None, the transaction is aborted / no updates are applied.

    Returns:
      Operation objects with the results of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)

    def _TransactionBody(pipe):
      pipe.watch(vkey)

      if fields is None:
        hash_values = pipe.hgetall(vkey)
      else:
        values = pipe.hmget(vkey, fields)
        hash_values = dict(zip(fields, values))

      # Get the updates from the callback. A return value of None means no
      # updates need to be applied.
      new_hash_values = get_updates_fn(hash_values)
      if new_hash_values is None:
        return [True]

      # Put the new values into the DB.
      pipe.multi()
      pipe.hmset(vkey, new_hash_values)

      return pipe.execute()

    op = TransactionSkeleton(shard, _TransactionBody)

    return Operation(
        success=all([op.success, op.response_value, op.response_value[0]]),
        response_value=op.response_value)

  def HashCheckAndReplace(self, key, get_updates_fn, tx_retries=MAX_TX_RETRIES):
    """In a transaction, retrieve a full Hash, callback to get updated values,
    and completely replace the Hash with a new Dict. If the underlying Hash
    changes in the process, the transaction is retried with the new values.

    Args:
      key - the key of the Hash.
      get_updates_fn - Callable which accepts the current value of the Hash
          fields requested, and returns a Dict of {field: value} to replace the
          Hash with. If the callback returns None, the transaction is aborted /
          no replacement is made.
      tx_retries - Maximum number of transactional retries to allow before
        giving up entirely.

    Returns:
      Operation objects with the results of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)

    def _TransactionBody(pipe):
      pipe.watch(vkey)

      hash_values = pipe.hgetall(vkey)

      # Get the updates from the callback. A return value of None means no
      # updates need to be applied.
      new_hash_values = get_updates_fn(hash_values)
      if new_hash_values is None:
        return [True]

      # Put the new values into the DB.
      pipe.multi()
      pipe.delete(vkey)
      if len(new_hash_values) > 0:

        pipe.hmset(vkey, new_hash_values)

      return pipe.execute()

    op = TransactionSkeleton(shard, _TransactionBody, max_retries=tx_retries)

    return Operation(
        success=all([op.success, op.response_value, op.response_value[0]]),
        response_value=op.response_value)

  def HashCheckAndDelete(self, key, get_deletes_fn, tx_retries=MAX_TX_RETRIES):
    """In a transaction, retrieve a full Hash, callback to get fields to delete,
    and delete those fields from the Hash. If the underlying Hash changes in the
    process, the transaction is retried with the new values.

    Args:
      key - the key of the Hash.
      get_deletes_fn - Callable which accepts the current value of the Hash
          fields requested, and returns an iterable of field names to delete
          from the Hash. If the callback returns None, the transaction is
          aborted / no deletions are made.
      tx_retries - Maximum number of transactional retries to allow before
        giving up entirely.

    Returns:
      Operation objects with the results of the query.
    """
    _, shard, vkey = self._GetShardInfo(key)

    def _TransactionBody(pipe):
      pipe.watch(vkey)

      hash_values = pipe.hgetall(vkey)

      # Get the updates from the callback. A return value of None means no
      # updates need to be applied.
      field_deletions = get_deletes_fn(hash_values)
      if field_deletions is None:
        return [True]

      # Put the new values into the DB.
      pipe.multi()

      for field in field_deletions:
        pipe.hdel(vkey, field)

      return pipe.execute()

    op = TransactionSkeleton(shard, _TransactionBody, max_retries=tx_retries)

    return Operation(
        success=all([op.success, op.response_value, op.response_value[0]]),
        response_value=op.response_value)

  ###################################################################
  # LISTS
  ###################################################################

  def ListLeftPush(self, key, value):
    """Push a value onto the left end of a List.

    Args:
      key - The key of the List.
      value - Value to push onto the List.

    Returns:
      Operation object.
    """
    _, shard, vkey = self._GetShardInfo(key)
    result = shard.lpush(vkey, value)

    return Operation(success=result)

  def ListGetAndDeleteAtomic(self, key):
    """Retrieve the entire contents of a List, and then delete it, in an atomic
    operation.

    Args:
      key - The key of the List.

    Returns:
      Operation object.
    """
    _, shard, vkey = self._GetShardInfo(key)

    pipe = shard.pipeline()
    pipe.lrange(vkey, 0, -1)
    pipe.delete(vkey)
    result = pipe.execute()

    return Operation(success=True, response_value=result[0])

  def ListLenBatch(self, keys):
    """Retrieve the length of multiple Lists.

    Args:
      keys - The keys of the Lists.

    Returns:
      Operation object. On success, the response_value field contains the list
      of List lengths.
    """
    def _ShardListLenBatch(shard, keys, vkeys, vals):
      pipe = shard.pipeline()
      map(pipe.llen, vkeys)
      result = pipe.execute()
      return Operation(success=True, response_value=result)

    return self._ShardedOp([(k, None) for k in keys], _ShardListLenBatch)

  ###################################################################
  # MISC
  ###################################################################

  @contextmanager
  def Lock(self, lock_name, duration):
    """Context manager for acquiring/releasing a global lock.

    Args:
      lock_name - Name of the lock to obtain.
      duration - Timeout of the lock in seconds.
    """
    (_, shard, vlock_name) = self._GetShardInfo(lock_name)

    with shard.lock(vlock_name, duration, LOCK_WAIT_SLEEP):
      yield

  def Keys(self, pattern):
    """Retrieve the set of keys in the database matching a glob-style pattern.

    NOTE: This command runs in O(n), where n is the number of keys in the
    database. It should only be used for maintenance or debugging purposes.

    Args:
      pattern - Glob-style pattern to retrieve keys for.

    Returns:
      An Operation object with the result of the query. The response_value field
      contains a list of keys matching the pattern.
    """
    keys = set()

    try:
      # Get the keys from each shard for prefixed and non-prefixed keys.
      for shard in self.shards:
        shard_keys = shard.keys(pattern)
        shard_vkeys = shard.keys('*|%s' % pattern)

        keys.update(shard_keys)
        keys.update(['|'.join(vkey.split('|')[1:]) for vkey in shard_vkeys])

      op = Operation(success=True, response_value=keys)

    except ConnectionError:
      client.Counter('redis_connection_error')
      op = Operation(success=False, traceback=traceback.format_exc())
    except Exception:
      op = Operation(success=False, traceback=traceback.format_exc())

    return op

  ###################################################################
  # SHARDING HELPERS
  ###################################################################

  def _GetShardInfo(self, key):
    """For a given client key, calculate the associated shard number and
    virtual key.

    Args:
      key - Client key string.

    Returns:
      A tuple of (shard number, shard, virtual key)
    """
    key_vbucket = self._GetVbucket(key)

    shard_num = bisect.bisect(self.vbucket_starts, key_vbucket) - 1
    shard = self.shards[shard_num]
    vkey = self._MakeVkey(key, key_vbucket)

    return (shard_num, shard, vkey)

  def _GetVbucket(self, key):
    """For a given client key, calculate the associated Virtual Bucket. This is
    the hash of the key string modded by the number of Virtual Buckets.

    Args:
      key - Client key string.

    Returns:
      Virtual bucket number.
    """
    return hash(key) % self.num_vbuckets

  def _MakeVkey(self, key, vbucket):
    """For a given client key and Virtual Bucket, generate the virtual key.

    Args:
      key - Client key string.
      vbucket - Associated Virtual Bucket number.

    Returns:
      The virtual key string.
    """
    return "%d|%s" % (vbucket, key)

  def _ShardedOp(self, key_value_tuples, shard_execute_fn):
    """Perform an operation on the necessary shards in the cluster, and
    aggregate the results. The callback will be invoked for each shard, with the
    keys, virtual keys, and and values that belong to that shard.

    Args:
      key_value_tuples - List of tuples of (key, value) for the operation.
      shard_execute_fn - Callback for executing the desired operation in each
          shard. The signature is f(shard, keys, vkeys, values) => Operation

    Returns:
      A CompoundOperation with the results of the queries. The response_value
      field is the combined response_value of each sub-operation (assuming all
      the sub-operations succeeded and returned a response_value)
    """
    # Compute the shard info for all the keys, and group them by the shard in
    # which they reside.
    vkeys_by_shard = defaultdict(list)
    for i, (key, val) in enumerate(key_value_tuples):
      (shard_num, _, vkey) = self._GetShardInfo(key)
      vkeys_by_shard[shard_num].append((i, key, vkey, val))

    # Split the request for each shard into a separate greenlet to allow the
    # requests to happen in parallel.
    op = CompoundOperation()
    responses = []

    def _ShardGreenletWrapper(shard_num, vkey_tuples):
      indices, keys, vkeys, values = [list(i) for i in zip(*vkey_tuples)]

      try:
        sub_op = shard_execute_fn(self.shards[shard_num], keys, vkeys, values)
      except ConnectionError:
        client.Counter('redis_connection_error_shard_%s' % shard_num)
        sub_op = Operation(success=False, traceback=traceback.format_exc())
      except Exception:
        sub_op = Operation(success=False, traceback=traceback.format_exc())

      # Aggregate the results.
      op.AddOp(sub_op)
      if sub_op.response_value:
        responses.extend(zip(indices, sub_op.response_value))

    greenlets = [
        gevent.spawn(_ShardGreenletWrapper, shard_num, vkey_tuples)
        for shard_num, vkey_tuples in vkeys_by_shard.iteritems()]

    gevent.joinall(greenlets)

    # Sort the combined responses back into the request order.
    responses = sorted(responses, key=lambda r: r[0])
    op.response_value = [r[1] for r in responses]

    return op

class _ShardCheckAndSetCallable(object):
  """Callable class to implement the per-shard logic for BatchCheckAndMultiSet.
  This is implemented as a class instead of a closure to reduce memory use and
  avoid leaks.
  """
  def __init__(self, engine, get_updates_fn):
    """
    Args:
      engine - RedisEngine object to make calls through.
      get_updates_fn - See RedisEngine.BatchCheckAndMultiSet.
    """
    self.engine = engine
    self.get_updates_fn = get_updates_fn

  def __call__(self, *args, **kwargs):
    return self._ShardCheckAndSet(*args, **kwargs)

  def _ShardCheckAndSet(self, shard, keys, vkeys, values):
    op = CompoundOperation()

    # Split the request into batches to avoid locking too many keys at once.
    greenlet_args = [(
            shard,
            keys[i:i + CAS_BATCH_SIZE],
            vkeys[i:i + CAS_BATCH_SIZE],
            values[i:i + CAS_BATCH_SIZE])
        for i in xrange(0, len(keys), CAS_BATCH_SIZE)]

    greenlets = []
    for j in xrange(0, len(greenlet_args), CAS_PARALLEL_REQUESTS):
      greenlet_batch = [
          gevent.spawn(self._ShardCheckAndSetBatch, *args)
          for args in greenlet_args[j:j + CAS_PARALLEL_REQUESTS]]

      greenlets.extend(greenlet_batch)
      gevent.joinall(greenlet_batch)

    # Get the results and combine them into a single Operation.
    [op.AddOp(g.get()) for g in greenlets]
    op.response_value = []
    for sub_op in op.sub_operations:
      if sub_op.success:
        op.response_value.extend(sub_op.response_value)

    return op

  def _ShardCheckAndSetBatch(self, shard, keys, vkeys, values):
    retries = 0
    while True:
      # Open a transactional pipeline.
      pipe = shard.pipeline(True)

      try:
        # Lock the keys to start the operation.
        pipe.watch(*vkeys)

        # Batch Get the keys to be updated.
        values = pipe.mget(vkeys)

        # Get the new values from the client.
        put_kv_tuples = []
        for key, value in YieldByCount(itertools.izip(keys, values)):
          put_kv_tuples.extend(self.get_updates_fn(key, value))

        # Put the new values into the DB.
        pipe.multi()

        val_dict = {}
        for key, value in put_kv_tuples:
          vbucket = self.engine._GetVbucket(key)
          vkey = self.engine._MakeVkey(key, vbucket)
          val_dict[vkey] = value

        pipe.mset(val_dict)

        response = pipe.execute()

      except redis.WatchError:
        # Lock error occurred. Try the operation again.
        gevent.sleep(0.1 * retries + 0.1 * random.random())

        retries += 1
        if retries > MAX_TX_RETRIES:
          raise

        continue

      finally:
        # Make sure we always reset the pipe.
        pipe.reset()

      # If we make it here without a WatchError, the operation succeeded.
      op = Operation(success=True, response_value=response, retries=retries)
      break

    return op

def TransactionSkeleton(client, body_fn, max_retries=MAX_TX_RETRIES):
  """Basic skeleton implementation of a Redis transaction, handling retries and
  resets. The body of the transaction is left to a callback.

  Args:
    client - Redis object to open a pipeline on.
    body_fn - Callback to execute the body of the transaction (i.e. watch,
        multi, exec, ...). The signature of the function should be:
          body_fn(pipeline) -> Operation
    max_retries - Maximum number of transactional retries to allow before
        giving up entirely.

  Returns:
    Operation object.
  """
  retries = 0

  while True:
    # Open a transactional pipeline.
    pipe = client.pipeline(True)

    try:
      response = body_fn(pipe)

    except (redis.WatchError, redis.ConnectionError):
      # Lock error or connection error occurred. Try the operation again.
      gevent.sleep(0.1 * retries + 0.1 * random.random())

      retries += 1
      if retries > max_retries:
        raise

      continue

    finally:
      # Make sure we always reset the pipe.
      pipe.reset()

    # If we make it here without a WatchError, the operation succeeded.
    op = Operation(success=True, response_value=response, retries=retries)
    break

  return op


text:
FROM ubuntu:14.04.4
MAINTAINER Lloyd H. Meinholz <meinholz@gmail.com>

ENV DEBIAN_FRONTEND noninteractive
ENV MAVEN_VERSION 3.3.9
ENV GROOVY_VERSION 2.4.5

COPY download-maven.sh /tmp/download-maven.sh
COPY download-groovy.sh /tmp/download-groovy.sh

RUN apt-get update && apt-get install -y \
    apt-transport-https \
    build-essential \
    ca-certificates \
    lsb-release \
    libssl-dev \
    curl \
    git \
    unzip \
    rlwrap \
    libfontconfig \
    libfontconfig-dev \
    python-software-properties \
    software-properties-common && \
    rm -rf /var/lib/apt/lists/* && \
    locale-gen en_US.UTF-8 && \
    dpkg-reconfigure locales && \

    # Install Oracle Java from PPA & set Oracle Java as the default
    add-apt-repository ppa:webupd8team/java -y && \
    apt-get update && \
    echo oracle-java8-installer shared/accepted-oracle-license-v1-1 select true | /usr/bin/debconf-set-selections && \
    apt-get -y install oracle-java8-installer && apt-get clean && \
    update-java-alternatives -s java-8-oracle && \
    echo "export JAVA_HOME=/usr/lib/jvm/java-8-oracle" >> ~/.bashrc && \

    # Install maven
    /tmp/download-maven.sh && \
    tar xf /tmp/apache-maven-${MAVEN_VERSION}-bin.tar.gz -C /opt && \
    mv /opt/apache-maven-${MAVEN_VERSION} /opt/maven-${MAVEN_VERSION} && \
    echo "export M2_HOME=/opt/maven-${MAVEN_VERSION}" >> ~/.bashrc && \
    echo "export PATH=/opt/maven-${MAVEN_VERSION}/bin:\$PATH" >> ~/.bashrc && \

    # Install groovy
    /tmp/download-groovy.sh && \
    unzip /tmp/apache-groovy-sdk-${GROOVY_VERSION}.zip -d /opt && \
    echo "export GROOVY_HOME=/opt/groovy-${GROOVY_VERSION}" >> ~/.bashrc && \
    echo "export PATH=/opt/groovy-${GROOVY_VERSION}/bin:\$PATH" >> ~/.bashrc

CMD ["/bin/bash"]

redpajama_set_name: RedPajamaGithub


text:
"""
Tools to for cryoscope analysis
Brian Tarasinski
Dec 2017
Edited by Adriaan Rol
"""
import logging
from copy import copy
from typing import Union
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.ticker
from pycqed.analysis.tools.plotting import (set_xlabel, set_ylabel,
                                            flex_colormesh_plot_vs_xy)

import scipy.signal as ss
import scipy.optimize as so
import scipy.interpolate as si

def normalize_sincos(
        data,
        window_size_frac=500,
        window_size=None,
        do_envelope=True):

    if window_size is None:
        window_size = len(data) // window_size_frac

        # window size for savgol filter must be odd
        window_size -= (window_size + 1) % 2

    mean_data_r = ss.savgol_filter(data.real, window_size, 0, 0)
    mean_data_i = ss.savgol_filter(data.imag, window_size, 0, 0)

    mean_data = mean_data_r + 1j * mean_data_i

    if do_envelope:
        envelope = np.sqrt(
            ss.savgol_filter(
                (np.abs(
                    data -
                    mean_data))**2,
                window_size,
                0,
                0))
    else:
        envelope = 1
    norm_data = ((data - mean_data) / envelope)
    return norm_data

def fft_based_freq_guess_complex(y):
    """
    guess the shape of a sinusoidal complex signal y (in multiples of
        sampling rate), by selecting the peak in the fft.
    return guess (f, ph, off, amp) for the model
        y = amp*exp(2pi i f t + ph) + off.
    """
    fft = np.fft.fft(y)[1:len(y)]
    freq_guess_idx = np.argmax(np.abs(fft))
    if freq_guess_idx >= len(y) // 2:
        freq_guess_idx -= len(y)
    freq_guess = 1 / len(y) * (freq_guess_idx + 1)

    phase_guess = np.angle(fft[freq_guess_idx]) + np.pi / 2
    amp_guess = np.absolute(fft[freq_guess_idx]) / len(y)
    offset_guess = np.mean(y)

    return freq_guess, phase_guess, offset_guess, amp_guess

class CryoscopeAnalyzer:
    def __init__(
            self,
            time,
            complex_data,
            norm_window_size=61,
            demod_freq=None,
            derivative_window_length=None,
            derivative_order=2,
            nyquist_order=0,
            demod_smooth=None):
        """
        analyse a cryoscope measurement.

        time: array of times (lengths of Z pulse)
        complex_data: measured data, combine x- and y- results in a
            complex number

        norm_window_size: window size used for normalizing sine and cosine

        demod_freq: frequency for demodulation. Is guessed if None.

        derivative_window_length, derivative_order: parameters of the sovgol
            filter used for extracting frequency.
        Needs some playing around sometimes.

        demod_smooth: when the demodulated signal should be smoothed before
        taking derivative, set this to a tuple (window_length, order),
        again parametrizing a sovgol filter.

        """
        self.time = time
        self.data = complex_data
        self.norm_data = normalize_sincos(self.data, window_size=61)
        self.demod_freq = demod_freq
        self.derivative_window_length = derivative_window_length
        self.demod_smooth = demod_smooth
        self.nyquist_order = nyquist_order

        self.sampling_rate = 1 / (self.time[1] - self.time[0])

        if self.derivative_window_length is None:
            self.derivative_window_length = 7 / self.sampling_rate

        self.derivative_window_size = max(
            3, int(self.derivative_window_length * self.sampling_rate))
        self.derivative_window_size += (self.derivative_window_size + 1) % 2

        if self.demod_freq is None:
            self.demod_freq = - \
                fft_based_freq_guess_complex(self.norm_data)[
                    0] * self.sampling_rate

        self.demod_data = np.exp(
            2 * np.pi * 1j * self.time * self.demod_freq) * self.norm_data

        if self.demod_smooth:
            n, o = self.demod_smooth
            r, i = self.demod_data.real, self.demod_data.imag
            r = ss.savgol_filter(r, n, o, 0)
            i = ss.savgol_filter(i, n, o, 0)
            self.demod_data = r + 1j * i

        # extract the phase. unwrapping only works well if demodulation is
        # good!
        self.phase = np.unwrap(np.angle(self.demod_data))

        # extract frequency by a lowpass-derivative filter.

        # use a savitzky golay filter: it take sliding window of length
        # `window_length`, fits a polynomial, returns derivative at
        # middle point
        self.detuning = ss.savgol_filter(
            self.phase / (
                2 * np.pi),
            window_length=self.derivative_window_size,
            polyorder=derivative_order,
            deriv=1) * self.sampling_rate
        self.real_detuning = self.get_real_detuning(self.nyquist_order)

    def get_real_detuning(self, nyquist_order=None):
        if nyquist_order is None:
            nyquist_order = self.nyquist_order

        real_detuning = self.detuning-self.demod_freq+self.sampling_rate*nyquist_order
        return real_detuning

    def get_amplitudes(self):
        """
        Converts the real detuning to amplitude
        """
        real_detuning = self.get_real_detuning()
        if hasattr(self, 'freq_to_amp'):

            amplitudes = self.freq_to_amp(real_detuning)
            return amplitudes
        else:
            raise NotImplementedError('Add a "freq_to_amp" method.')

    def plot_short_time_fft(self, ax=None,
                            title='Short time Fourier Transform',
                            window_size=100, **kw):

        if ax is None:
            ax = plt.gca()
        ax.set_title(title)

        # If window size is longer than data this plot will throw an exception
        window_size = np.min([window_size, len(self.norm_data)])
        f, t, Zxx = ss.stft(self.norm_data, fs=self.sampling_rate,
                            nperseg=window_size,
                            noverlap=0.95 * window_size, return_onesided=False)
        m = np.argsort(f)
        ax.pcolormesh(self.time[0] + t, f[m], np.abs(Zxx)[m, :])
        ax.set_ylabel('Frequency')
        ax.set_xlabel('Time')

        formatter = matplotlib.ticker.EngFormatter(unit='s')
        ax.xaxis.set_major_formatter(formatter)
        formatter = matplotlib.ticker.EngFormatter(unit='Hz')
        ax.yaxis.set_major_formatter(formatter)

    def plot_raw_data(self, ax=None, title="Raw cryoscope data",
                      style=".-", **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        ax.plot(self.time, self.data.real, style, label="Re", color="C0")
        ax.plot(self.time, self.data.imag, style, label="Im", color="C1")
        ax.legend()
        set_xlabel(ax, 'Time', 's')
        set_ylabel(ax, "Amplitude", 'a.u.')

    def plot_normalized_data(self, ax=None, title='Normalized cryoscope data',
                             style=".-", **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)

        ax.plot(
            self.time,
            self.norm_data.real,
            style,
            label="Re",
            color="C0")
        ax.plot(self.time, self.norm_data.imag, style, label="Im", color="C1")
        ax.legend()
        ax.set_xlabel("Time")
        ax.set_ylabel("Amplitude")
        formatter = matplotlib.ticker.EngFormatter(unit='s')
        ax.xaxis.set_major_formatter(formatter)

    def plot_demodulated_data(self, ax=None,
                              title='Demodulated cryoscope data',
                              style=".-", **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        ax.plot(
            self.time,
            self.demod_data.real,
            style,
            label="Re",
            color="C0")
        ax.plot(
            self.time,
            self.demod_data.imag,
            style,
            label="Im",
            color="C1")
        ax.legend()
        ax.set_xlabel("Time")
        ax.set_ylabel("Amplitude")
        formatter = matplotlib.ticker.EngFormatter(unit='s')
        ax.xaxis.set_major_formatter(formatter)

    def plot_normalized_data_circle(self, ax=None,
                                    title='Normalized cryoscope data', **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        ax.set_xlabel("Re")
        ax.set_ylabel("Im")
        ax.plot(self.norm_data.real, self.norm_data.imag, ".")

    def plot_phase(self, ax=None, title="Cryoscope demodulated phase",
                   wrap=False, **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        if wrap:
            ax.plot(self.time, self.phase % (2 * np.pi), ".", color="C0")
        else:
            ax.plot(self.time, self.phase, ".", label="Im", color="C0")
        set_xlabel(ax, 'Time', 's')
        set_ylabel(ax, 'Phase', 'deg')

    def plot_detuning(self, ax=None,
                      title="Detuning from demodulation frequency", **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        ax.plot(self.time, self.detuning, ".-", color="C0")
        set_xlabel(ax, 'Time', 's')
        set_ylabel(ax, 'Frequency', 'Hz')

    def plot_frequency(self, ax=None, title='Detuning frequency',
                       nyquists=None, style=".-", show_demod_freq=True, **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)

        if nyquists is None:
            nyquists = [self.nyquist_order]
        for n in nyquists:
            if show_demod_freq:
                ax.axhline(-self.demod_freq + self.sampling_rate *
                           n, linestyle='--', c='grey')
            real_detuning = self.get_real_detuning(n)
            ax.plot(self.time, real_detuning, style)
        set_xlabel(ax, 'Time', 's')
        set_ylabel(ax, 'Frequency', 'Hz')

    def plot_amplitude(self, ax=None, title='Cryoscope amplitude',
                       nyquists=None, style=".-", **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)
        amp = self.get_amplitudes()
        ax.plot(self.time, amp, style)
        set_xlabel(ax, 'Time', 's')
        set_ylabel(ax, 'Amplitude', 'V')

def sincos_model_real_imag(times, freq, phase):
    r, i = np.cos(2 *
                  np.pi *
                  times *
                  freq +
                  phase), np.sin(2 *
                                 np.pi *
                                 times *
                                 freq +
                                 phase)
    return np.hstack((r, i))

class DacArchAnalysis:
    """
    Given cryscope time series from a square region in time and amplitude,
    fit complex oscillations to obtain a dac arch. Tries to be smart about
    supersampled signals, after constructing the arc, fits a polynomial
    in order to facilitate interpolation.
    """

    def __init__(
            self,
            times,
            amps,
            data,
            exclusion_indices=[],
            poly_fit_order=2,
            nyquist_calc='auto',
            invert_frequency_sign=False,
            plot_fits=False):
        """
        Extract a dac arch from a set of cryoscope-style measurements.

        times: array of pulse lengths
        amps: array of pulse amplitudes

        data: 2D array of measurement results (size of len(times x amps)).
            Complex numbers containing x- and y- values

        poly_fit_order: order of model used for fitting the dac-arch

        invert_frequency_sign: boolean. might be useful if x and y are
            interchanged in measurement

        plot_fits: plots how the fit is going, for display.
        """
        self.data = data
        self.times = times
        self.amps = np.array(amps)

        self.poly_fit_order = poly_fit_order

        self.sampling_rate = 1 / (self.times[1] - self.times[0])

        self.freqs = []
        self.excl_amps = []
        self.excl_freqs = []
        self.exclusion_indices = exclusion_indices

        self.norm_data = []

        for d in self.data:
            self.norm_data.append(normalize_sincos(d, window_size=11))

        self.norm_data = np.array(self.norm_data)

        for idx, nd in enumerate(self.norm_data):
            guess_f, guess_ph, *_ = fft_based_freq_guess_complex(nd)
            guess_f *= self.sampling_rate

            nd_real_imag = np.hstack([nd.real, nd.imag])
            try:
                fit, err = so.curve_fit(sincos_model_real_imag,
                                        self.times, nd_real_imag,
                                        p0=[guess_f, guess_ph])
            except Exception as e:
                logging.warning(e)
                logging.warning('Fitting failed for trace: {}, '
                                'replacing with nan'.format(idx))
                fit = [np.nan]
                err = [np.nan]

            if plot_fits:
                plt.figure()
                plt.plot(self.times, nd.real, "-.b")
                plt.plot(self.times, nd.imag, ".-r")

                tt = np.linspace(self.times[0], self.times[-1], 300)
                plt.plot(tt, sincos_model_real_imag(tt, *fit)[:len(tt)], "--b")
                plt.plot(tt, sincos_model_real_imag(tt, *fit)[len(tt):], "--r")

            self.freqs.append(fit[0])

        self.freqs = np.array(self.freqs)

        self.nyquist = self._calc_nyquist(nyquist_calc)

        self.freqs = self.freqs + self.nyquist * self.sampling_rate

        if invert_frequency_sign:
            self.freqs = -self.freqs

        # Exclude data from excludion indices
        self.filt_freqs = np.delete(self.freqs, self.exclusion_indices)
        self.filt_amps = np.delete(self.amps, self.exclusion_indices)
        self.excl_freqs = self.freqs[self.exclusion_indices]
        self.excl_amps = self.amps[self.exclusion_indices]

        self.poly_fit = np.polyfit(self.filt_amps, self.filt_freqs,
                                   self.poly_fit_order)

        self._inv_interpolation = None

    def amp_to_freq(self, amp):
        """
        Find the frequency that corresponds to a given amplitude by
        evaluating the fit to the extracted data.
        """
        return np.polyval(self.poly_fit, amp)

    def freq_to_amp(self, freq, kind='root_parabola', **kw):
        """
        Find the amplitude that corresponds to a given frequency, by
        numerically inverting the fit.

        freq: The frequency or set of frequencies.
        kind: Which technique to use:
            "interpolate": Uses numerical interpolation to find the inverse.
            Only works if freq is in the range of measured dac values.
            "root": Finds the inverse of the model numerical. Slow, but can
            extrapolate.

        **kw : get passed on to methods that implement the different "kind"
            of calculations.
        """

        if kind == 'interpolate':
            if self._inv_interpolation is None:
                no_samples = 50
                self.sampled_amps = np.linspace(
                    np.min(
                        self.amps), np.max(
                        self.amps), no_samples)

                self.sampled_freqs = self.amp_to_freq(self.sampled_amps)

                self._inv_interpolation = si.interp1d(
                    self.sampled_freqs, self.sampled_amps, kind='cubic',
                    bounds_error=False,
                    fill_value='extrapolate')
            return self._inv_interpolation(freq)
        if kind == 'root':
            return np.vectorize(self._freq_to_amp_root)(freq)

        if kind == 'root_parabola':
            # return self._freq_to_amp_root_parabola(freq, **kw)
            return freq_to_amp_root_parabola(freq=freq,
                                             poly_coeffs=self.poly_fit, **kw)

        raise ValueError("`kind` not understood")

    def _calc_nyquist(self, nyquist_calc):
        """
        calculates nyquist orders based on frequencies
        """
        if nyquist_calc == 'auto':
            # Ignoring invalid for the nan comparison
            with np.errstate(invalid='ignore'):
                freqs = copy(self.freqs)
                freqs[self.exclusion_indices] = np.nan
                nyquist = np.cumsum(freqs[1:] < freqs[:-1])
                nyquist = np.hstack(([0], nyquist))
        elif nyquist_calc == 'disabled':
            nyquist = np.zeros(len(self.freqs))
        else:
            raise NotImplementedError()
            # FIXME: proper support for auto nyquist with
            # a proper nyquist should be extracted
        return nyquist

    def _freq_to_amp_root(self, freq):
        """
        Find the amplitude corresponding to a given frequency by numerically
        inverting the fit.
        """

        poly = np.array(self.poly_fit)
        poly[-1] -= freq

        roots = np.roots(poly)

        # return the solution that is real and closest to the givenamp range

        real_mask = np.abs(roots.imag) < 1e-8

        if not any(real_mask):
            return None

        dist_from_range = np.abs(roots[real_mask] - np.mean(self.amps))

        return roots[real_mask][np.argmin(dist_from_range)].real

    def plot_freqs(self, ax=None, title='', **kw):
        if ax is None:
            ax = plt.gca()
        ax.set_title(title)

        amps_sorted = [x for x, _ in sorted(
            zip(self.filt_amps, self.filt_freqs))]
        freqs_sorted = [y for _, y in sorted(
            zip(self.filt_amps, self.filt_freqs))]

        ax.plot(amps_sorted, freqs_sorted, ".-")
        ax.scatter(self.excl_amps, self.excl_freqs, marker='x', color='C3')
        aa = np.linspace(min(self.amps), max(self.amps), 50)
        ax.plot(aa, np.polyval(self.poly_fit, aa), label='fit')
        set_xlabel(ax, "Amplitude", 'V')
        set_ylabel(ax, 'Detuning', 'Hz')

    def plot_ffts(self, ax=None, title='', nyquist_unwrap=False, **kw):
        if ax is None:
            ax = plt.gca()
        if nyquist_unwrap:
            raise NotImplementedError
        ax.set_title(title)
        ffts = np.fft.fft(self.norm_data)

        freqs = np.arange(len(ffts[0])) * self.sampling_rate / len(ffts[0])

        flex_colormesh_plot_vs_xy(xvals=np.array(self.amps), yvals=freqs,
                                  zvals=np.abs(ffts).T,
                                  ax=ax)

        ax.scatter(self.filt_amps, self.filt_freqs % self.sampling_rate, color="C1",
                   facecolors='none', label='Dominant freqs.')

        ax.scatter(self.excl_amps, self.excl_freqs % self.sampling_rate,
                   color="C3",
                   marker='x')
        aa = np.linspace(min(self.amps), max(self.amps), 300)

        ax.plot(aa, np.polyval(self.poly_fit, aa) % self.sampling_rate, "r",
                label='fit')
        set_xlabel(ax, "Amplitude", 'V')  # a.u.
        set_ylabel(ax, 'Detuning', 'Hz')
        ax.legend()

def freq_to_amp_root_parabola(freq, poly_coeffs, positive_branch=True):
    """
    Converts freq in Hz to amplitude in V.

    Requires "poly_coeffs" to be set to the polynomial values
    extracted from the cryoscope flux arc.

    Assumes a parabola to find the roots but should also work for a higher
    order polynomial, except that it will pick the wrong branch.

    N.B. this method assumes that the polycoeffs are with respect to the
        amplitude in units of V.
    """
    # recursive allows dealing with an array of freqs
    if isinstance(freq, (list, np.ndarray)):
        return np.array([freq_to_amp_root_parabola(
            freq=f, poly_coeffs=poly_coeffs,
            positive_branch=positive_branch) for f in freq])

    p = np.poly1d(poly_coeffs)
    sols = (p-freq).roots

    # sols returns 2 solutions (for a 2nd order polynomial)
    if positive_branch:
        sol = np.max(sols)
    else:
        sol = np.min(sols)

    # imaginary part is ignored, instead sticking to closest real value
    return np.real(sol)


text:
.. _why_is_my_route_not_found:

Why is my route not found?
==========================

Once you've drilled down into Envoy responses and discovered Envoy generating local responses with the message
"Sending local reply with details route_not_found" the next question is _why_?

Often you can look at your route configuration and the headers sent, and see what is missing.
One often overlooked problem is host:port matching. If your route configuration matches the domain
www.host.com but the client is sending requests to www.host.com:443, it will not match.

If this is the problem you are encountering you can solve it one of two ways. First by changing your
configuration to match host:port pairs, going from

.. code-block:: yaml

  domains:
    - "www.host.com"

to

.. code-block:: yaml

  domains:
    - "www.host.com"
    - "www.host.com:80"
    - "www.host.com:443"

The other is to strip ports entirely using :ref:`stripping port from host header <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.strip_matching_host_port>`. Not that this will only stip port 80 from insecure requests and 443 from secure request. It does
not just stop ports when routes are matched, but changes
the host sent downstream to also not include the port.


text:
var _ = require('lodash');

/**
 * @dgProcessor memberDocsProcessor
 * @description
 * Merge the member docs into their container doc, and remove them from the main docs collection
 */
module.exports = function memberDocsProcessor(log, getDocFromAlias, createDocMessage) {

  var mergeableTypes = {
    method: 'methods',
    property: 'properties',
    event: 'events'
  };

  return {
    $runAfter: ['ids-computed'],
    $runBefore: ['computing-paths'],
    $process: function(docs) {
      var parts;

      docs = _.filter(docs, function(doc) {

        // Is this doc a member of another doc?
        if ( doc.id.indexOf('#' ) !== -1 ) {
          doc.isMember = true;
          parts = doc.id.split('#');
          doc.memberof = parts[0];
          // remove the 'method:', 'property:', etc specifier from the id part
          doc.name = parts[1].replace(/^.*:/, '');

          log.debug('child doc found', doc.id, doc.memberof);

          var containerDocs = getDocFromAlias(doc.memberof, doc);

          if ( containerDocs.length === 0 ) {
            log.warn(createDocMessage('Missing container document: "'+ doc.memberof + '"', doc));
            return;
          }

          if ( containerDocs.length > 1 ) {
            // The memberof field was ambiguous, try prepending the module name too
            containerDocs = getDocFromAlias(_.template('${module}.${memberof}', doc), doc);
            if ( containerDocs.length !== 1 ) {
              log.warn(createDocMessage('Ambiguous container document reference: '+ doc.memberof));
              return;
            }
          }

          var containerDoc = containerDocs[0];
          doc.memberof = containerDoc.id;

          // Add this member doc to the container doc
          var containerProperty = mergeableTypes[doc.docType];
          var container = containerDoc[containerProperty] = containerDoc[containerProperty] || [];
          container.push(doc);

        } else {
          return doc;
        }
      });

      return docs;
    }
  };
};


text:
package com.microsoft.partnercatalyst.fortis.spark.sources.streamprovider

sealed trait StreamProviderException { self: Throwable =>
  // TODO
}
case class InvalidConnectorConfigException() extends Exception("Invalid connector config.") with StreamProviderException
case class UnsupportedConnectorConfigException() extends Exception("Unsupported connector config.") with StreamProviderException


text:
<templateSet group="AndroidLogKotlin">
  <template name="logd" value="android.util.Log.d(TAG, "$METHOD_NAME$: $content$")
$END$" description="Log.d(TAG, String)" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="kotlinClassName()" alwaysStopAt="false" />
    <variable name="content" expression="" defaultValue="" alwaysStopAt="true" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="logi" value="android.util.Log.i(TAG, "$METHOD_NAME$: $content$")
$END$" description="Log.i(TAG, String)" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="kotlinClassName()" alwaysStopAt="false" />
    <variable name="content" expression="" defaultValue="" alwaysStopAt="true" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="loge" value="android.util.Log.e(TAG, "$METHOD_NAME$: $content$", $exception$)
$END$" description="Log.e(TAG, String, Exception)" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="kotlinClassName()" alwaysStopAt="false" />
    <variable name="content" expression="" defaultValue="" alwaysStopAt="true" />
    <variable name="exception" expression="" defaultValue="e" alwaysStopAt="false" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="logr" value="android.util.Log.d(TAG, "$METHOD_NAME$() returned: " +  $result$)
$END$" description="Log result of this method" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="" alwaysStopAt="false" />
    <variable name="result" expression="variableOfType(methodReturnType())" defaultValue="" alwaysStopAt="true" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="logt" value="private val TAG = $className$::class.java.simpleName" description="A static logtag with your current classname" toReformat="true" toShortenFQNames="true">
    <variable name="className" expression="kotlinClassName()" defaultValue="" alwaysStopAt="false" />
    <context>
      <option name="KOTLIN_CLASS" value="true" />
    </context>
  </template>
  <template name="logw" value="android.util.Log.w(TAG, "$METHOD_NAME$: $content$", $exception$)
$END$" description="Log.w(TAG, String, Exception)" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="" alwaysStopAt="false" />
    <variable name="content" expression="" defaultValue="" alwaysStopAt="true" />
    <variable name="exception" expression="" defaultValue="e" alwaysStopAt="false" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="wtf" value="android.util.Log.wtf(TAG, "$METHOD_NAME$: $content$", $exception$)
$END$" description="Log.wtf(TAG, String, Exception)" toReformat="true" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="" alwaysStopAt="false" />
    <variable name="content" expression="" defaultValue="" alwaysStopAt="true" />
    <variable name="exception" expression="" defaultValue="e" alwaysStopAt="false" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
  <template name="logm" value="android.util.Log.d(TAG, "$METHOD_NAME$() called $content$")
$END$" description="Log with method parameters" toReformat="false" toShortenFQNames="true">
    <variable name="METHOD_NAME" expression="kotlinFunctionName()" defaultValue="kotlinClassName()" alwaysStopAt="false" />
    <variable name="content" expression="groovyScript("def params = _1.collect {it + ' = [$'+it+']'}.join(', '); return (params.empty  ? '' : ' with: ' + params)", functionParameters())" defaultValue="" alwaysStopAt="false" />
    <context>
      <option name="KOTLIN_STATEMENT" value="true" />
    </context>
  </template>
</templateSet>