def apply_transform(vertices, transform):
  """
  Scales mesh vertices from stored coordinate 
  space to physical coordinates.

  vertices: a Nx3 array of X,Y,Z floats
  transform: a 4x4 affine transform matrix 
    from stored to physical coordinates.
  """
  if (transform == np.eye(4)).all():
    return vertices
  
  n_verts = len(vertices)
  vertices = np.hstack([vertices, np.ones((n_verts,1)) ]).T
  return np.matmul(transform[:3,:], vertices).T class UnshardedLegacyPrecomputedMeshSource(object):
  def __init__(self, meta, cache, config, readonly=False):
    self.meta = meta
    self.cache = cache
    self.config = config

    self.readonly = bool(readonly)

    if 'transform' not in self.meta.info:
      self.transform = np.eye(4)
    else:
      self.transform = np.array(self.meta.info['transform'] + [0,0,0,1]).reshape(4,4)

    self.spatial_index = None
    if self.meta.spatial_index:
      mip = self.meta.mip or 0
      self.spatial_index = CachedSpatialIndex(
        self.cache, self.config,
        cloudpath=self.meta.layerpath, 
        bounds=self.meta.meta.bounds(mip), 
        resolution=self.meta.info['spatial_index'].get(
          'resolution', self.meta.meta.resolution(mip)
        ),
        chunk_size=self.meta.info['spatial_index']['chunk_size'],
      )

  @property
  def path(self):
    return self.meta.mesh_path

  def manifest_path(self, segid):
    mesh_json_file_name = str(segid) + ':0'
    return self.meta.join(self.path, mesh_json_file_name)

  def _get_manifests(self, segids):
    segids = toiter(segids)    
    paths = [ self.manifest_path(segid) for segid in segids ]
    fragments = self.cache.download(paths)

    contents = {}
    for filename, content in fragments.items():
      content = content.decode('utf8')
      content = json.loads(content)
      segid = filename_to_segid(filename)
      contents[segid] = content['fragments']

    return contents

  def _get_mesh_fragments(self, paths):
    paths = [ self.meta.join(self.path, path) for path in paths ]

    compress = self.config.compress
    if compress is None:
      compress = True

    fragments = self.cache.download(paths, compress=compress)
    fragments = [ (filename, content) for filename, content in fragments.items() ]
    fragments = sorted(fragments, key=lambda frag: frag[0]) # make decoding deterministic
    return fragments

  def exists(self, segids, progress=None):
    """
    Checks if the mesh exists.

    Returns: { label: path or None, ... }
    """
    manifest_paths = [ self.manifest_path(segid) for segid in segids ]
    progress = progress if progress is not None else self.config.progress

    cf = CloudFiles(
      self.meta.cloudpath, 
      progress=progress, 
      green=self.config.green, 
      secrets=self.config.secrets
    )
    exists = cf.exists(manifest_paths)

    segid_regexp = re.compile(r'(\d+):0$')

    output = {}
    for path, there in exists.items():
      (segid,) = re.search(segid_regexp, path).groups()
      output[segid] = path if there else None
  
    return output
  
  def get(
      self, segids, 
      remove_duplicate_vertices=True, 
      fuse=True,
      chunk_size=None
    ):
    """
    Merge fragments derived from these segids into a single vertex and face list.

    Why merge multiple segids into one mesh? For example, if you have a set of
    segids that belong to the same neuron.

    segids: (iterable or int) segids to render into a single mesh

    Optional:
      remove_duplicate_vertices: bool, fuse exactly matching vertices
      fuse: bool, merge all downloaded meshes into a single mesh
      chunk_size: [chunk_x, chunk_y, chunk_z] if passed only merge at chunk boundaries
    
    Returns: Mesh object if fused, else { segid: Mesh, ... }
    """
    segids = toiter(segids)
    dne = self.exists(segids)
    dne = [ label for label, path in dne.items() if path is None ]

    if dne:
      missing = ', '.join([ str(segid) for segid in dne ])
      raise ValueError(red(
        'Segment ID(s) {} are missing corresponding mesh manifests.\nAborted.' \
        .format(missing)
      ))

    fragments = self._get_manifests(segids)
    fragments = fragments.values()
    fragments = list(itertools.chain.from_iterable(fragments)) # flatten
    fragments = self._get_mesh_fragments(fragments)

    # decode all the fragments
    meshdata = defaultdict(list)
    for frag in tqdm(fragments, disable=(not self.config.progress), desc="Decoding Mesh Buffer"):
      segid = filename_to_segid(frag[0])
      try:
        mesh = Mesh.from_precomputed(frag[1])
      except Exception:
        print(frag[0], 'had a problem.')
        raise
      meshdata[segid].append(mesh)

    if not fuse:
      return { segid: Mesh.concatenate(*meshes) for segid, meshes in six.iteritems(meshdata) }

    meshdata = [ (segid, mesh) for segid, mesh in six.iteritems(meshdata) ]
    meshdata = sorted(meshdata, key=lambda sm: sm[0])
    meshdata = [ mesh for segid, mesh in meshdata ]
    meshdata = list(itertools.chain.from_iterable(meshdata)) # flatten
    mesh = Mesh.concatenate(*meshdata)
    mesh.vertices = apply_transform(mesh.vertices, self.transform)


    if not remove_duplicate_vertices:
      return mesh 

    if not chunk_size:
      return mesh.consolidate()

    if self.meta.mip is not None:
      mip = self.meta.mip
    else:
      # This will usually be wrong, but it's backwards compatible.
      # Throwing an exception instead would probably break too many
      # things.
      mip = self.config.mip

    if mip not in self.meta.meta.available_mips:
      raise exceptions.ScaleUnavailableError("mip {} is not available.".format(mip))

    resolution = self.meta.meta.resolution(mip)
    chunk_offset = self.meta.meta.voxel_offset(mip)

    return mesh.deduplicate_chunk_boundaries(
      chunk_size * resolution, is_draco=False,
      offset=(chunk_offset * resolution)
    )

  def save(self, segids, filepath=None, file_format='ply'):
    """
    Save one or more segids into a common mesh format as a single file.

    segids: int, string, or list thereof
    filepath: string, file-like, or None (optional)
    file_format: string (optional)
    
    Supported Formats: 'obj', 'ply', 'precomputed'
    """
    segids = toiter(segids)

    mesh = self.get(segids, fuse=True, remove_duplicate_vertices=True)

    if file_format == 'obj':
      data = mesh.to_obj()
    elif file_format == 'ply':
      data = mesh.to_ply()
    elif file_format == 'precomputed':
      data = mesh.to_precomputed()
    else:
      raise NotImplementedError('Only .obj, .ply, and precomputed are currently supported.')

    if not filepath:
      filepath = str(segids[0]) + "." + file_format
      if len(segids) > 1:
        filepath = "{}_{}.{}".format(segids[0], segids[-1], file_format)

    try:
      filepath.write(data)
    except AttributeError:
      with open(filepath, 'wb') as f:
        f.write(data)

  def get_bbox(self, bbox):
    if self.spatial_index is None:
      raise IndexError("A spatial index has not been created.")

    segids = self.spatial_index.query(bbox)
    return self.get(segids, fuse=False) class ShardingSpecification(object):
  def __init__(
    self, type, preshift_bits, 
    hash, minishard_bits, 
    shard_bits, 
    minishard_index_encoding='raw', 
    data_encoding='raw'
  ):

    self.type = type 
    self.preshift_bits = uint64(preshift_bits)
    self.hash = hash 
    self.minishard_bits = uint64(minishard_bits)
    self.shard_bits = uint64(shard_bits)
    self.minishard_index_encoding = minishard_index_encoding
    self.data_encoding = data_encoding

    self.minishard_mask = self.compute_minishard_mask(self.minishard_bits)
    self.shard_mask = self.compute_shard_mask(self.shard_bits, self.minishard_bits)              

    self.validate()

  def clone(self):
    return ShardingSpecification.from_dict(self.to_dict())

  def index_length(self):
    return int((2 ** self.minishard_bits) * 16)

  @property
  def hash(self):
    return self._hash

  @hash.setter
  def hash(self, val):
    if val == 'identity':
      self.hashfn = lambda x: uint64(x)
    elif val == 'murmurhash3_x86_128':
      self.hashfn = lambda x: uint64(mmh3.hash64(uint64(x).tobytes(), x64arch=False)[0]) 
    else:
      raise SpecViolation("hash {} must be either 'identity' or 'murmurhash3_x86_128'".format(val))

    self._hash = val

  @property
  def preshift_bits(self):
    return self._preshift_bits
  
  @preshift_bits.setter
  def preshift_bits(self, val):
    self._preshift_bits = uint64(val) 

  @property
  def shard_bits(self):
    return self._shard_bits
  
  @shard_bits.setter
  def shard_bits(self, val):
    self._shard_bits = uint64(val) 

  @property
  def minishard_bits(self):
    return self._minishard_bits
  
  @minishard_bits.setter
  def minishard_bits(self, val):
    val = uint64(val)
    self.minishard_mask = self.compute_minishard_mask(val)
    self._minishard_bits = uint64(val)

  def compute_minishard_mask(self, val):
    if val < 0:
      raise ValueError(str(val) + " must be greater or equal to than zero.")
    elif val == 0:
      return uint64(0)

    minishard_mask = uint64(1)
    for i in range(val - uint64(1)):
      minishard_mask <<= uint64(1)
      minishard_mask |= uint64(1)
    return uint64(minishard_mask)

  def compute_shard_mask(self, shard_bits, minishard_bits):
    ones64 = uint64(0xffffffffffffffff)
    movement = uint64(minishard_bits + shard_bits)
    shard_mask = ~((ones64 >> movement) << movement)
    minishard_mask = self.compute_minishard_mask(minishard_bits)
    return shard_mask & (~minishard_mask)

  @classmethod
  def from_json(cls, vals):
    dct = json.loads(vals.decode('utf8'))
    return cls.from_dict(dct)

  def to_json(self):
    return jsonify(self.to_dict())

  @classmethod
  def from_dict(cls, vals):
    vals = copy.deepcopy(vals)
    vals['type'] = vals['@type']
    del vals['@type']
    return cls(**vals)

  def to_dict(self):
    return {
      '@type': self.type,
      'preshift_bits': self.preshift_bits,
      'hash': self.hash,
      'minishard_bits': self.minishard_bits,
      'shard_bits': self.shard_bits,
      'minishard_index_encoding': self.minishard_index_encoding,
      'data_encoding': self.data_encoding,
    }

  def compute_shard_location(self, key):
    chunkid = uint64(key) >> uint64(self.preshift_bits)
    chunkid = self.hashfn(chunkid)
    minishard_number = uint64(chunkid & self.minishard_mask)
    shard_number = uint64((chunkid & self.shard_mask) >> uint64(self.minishard_bits))
    shard_number = format(shard_number, 'x').zfill(int(np.ceil(self.shard_bits / 4.0)))
    remainder = chunkid >> uint64(self.minishard_bits + self.shard_bits)

    return ShardLocation(shard_number, minishard_number, remainder)

  def synthesize_shards(self, data, data_offset=None, progress=False):
    """
    Given this specification and a comprehensive listing of
    all the items that could be combined into a given shard,
    synthesize the shard files for this set of labels.

    data: { label: binary, ... }

    e.g. { 5: b'...', 7: b'...' }

    data_offset: { label: offset, ... }

    e.g. { 5: 1234, 7: 5678...' }

    Returns: {
      $filename: binary data,
    }
    """
    return synthesize_shard_files(self, data, data_offset, progress)

  def synthesize_shard(self, labels, data_offset=None, progress=False, presorted=False):
    """
    Assemble a shard file from a group of labels that all belong in the same shard.

    Assembles the .shard file like:
    [ shard index; minishards; all minishard indices ]

    label_group: 
      If presorted is True:
        { minishardno: { label: binary, ... }, ... }
      If presorted is False:
        { label: binary }
    progress: show progress bars

    Returns: binary representing a shard file 
    """
    return synthesize_shard_file(self, labels, data_offset, progress, presorted)

  def validate(self):
    if self.type not in ('neuroglancer_uint64_sharded_v1',):
      raise SpecViolation(
        "@type ({}) must be 'neuroglancer_uint64_sharded_v1'." \
        .format(self.type)
      )

    if not (64 > self.preshift_bits >= 0):
      raise SpecViolation("preshift_bits must be a whole number less than 64: {}".format(self.preshift_bits))

    if not (64 >= self.minishard_bits >= 0):
      raise SpecViolation("minishard_bits must be between 0 and 64 inclusive: {}".format(self.minishard_bits))

    if not (64 >= self.shard_bits >= 0):
      raise SpecViolation("shard_bits must be between 0 and 64 inclusive: {}".format(self.shard_bits))

    if self.minishard_bits + self.shard_bits > 64:
      raise SpecViolation(
        "minishard_bits and shard_bits must sum to less than or equal to 64: minishard_bits<{}> + shard_bits<{}> = {}".format(
        self.minishard_bits, self.shard_bits, self.minishard_bits + self.shard_bits
      ))

    if self.hash not in ('identity', 'murmurhash3_x86_128'):
      raise SpecViolation("hash {} must be either 'identity' or 'murmurhash3_x86_128'".format(self.hash))

    if self.minishard_index_encoding not in ('raw', 'gzip'):
      raise SpecViolation("minishard_index_encoding only supports values 'raw' or 'gzip'.")

    if self.data_encoding not in ('raw', 'gzip'):
      raise SpecViolation("data_encoding only supports values 'raw' or 'gzip'.")
    
  def __str__(self):
    return "ShardingSpecification::" + str(self.to_dict()) class ShardReader(object):
  def __init__(
    self, meta, cache, spec,
    shard_index_cache_size=512,
    minishard_index_cache_size=128,
    green=False
  ):
    """
    Reads standard Precomputed shard files. 

    meta: a PrecomputedMetadata class
    cache: a CacheService instance
    spec: a ShardingSpecification instance

    shard_index_cache_size: size of LRU cache for fixed indices 
    minishard_index_cache_size: size of LRU cache for minishard indices
    """
    self.meta = meta
    self.cache = cache
    self.spec = spec
    self.green = green

    self.shard_index_cache = LRU(shard_index_cache_size)
    self.minishard_index_cache = LRU(minishard_index_cache_size)

  def get_filename(self, label):
    return self.compute_shard_location(label)[0]

  def compute_shard_location(self, label):
    """
    Returns (filename, shard_number) for meshes and skeletons. 
    Images require a different scheme.
    """
    shard_loc = self.spec.compute_shard_location(label)
    filename = str(shard_loc.shard_number) + '.shard'
    return (filename, shard_loc.minishard_number)

  def get_index(self, filename, path=""):
    """
    Retrieves the shard index which is used for 
    locating the appropriate minishard index.

    Returns: 2^minishard_bits entries of a uint64 
      array of [[ byte start, byte end ], ... ] 
    """
    indices = self.get_indices([ filename ], path, progress=False)
    index = list(indices.values())[0]
    if index is None:
      raise EmptyFileException(filename + " was zero bytes.")
    return index

  def get_indices(self, filenames, path="", progress=None):
    """
    For all given files, retrieves the shard index which 
    is used for locating the appropriate minishard indices.

    Returns: { 
      path_to_/filename.shard: 2^minishard_bits entries of a uint64 
            array of [[ byte start, byte end ], ... ],
      ...
    } 
    """
    filenames = toiter(filenames)
    filenames = [ self.meta.join(path, fname) for fname in filenames ]
    fufilled = { 
      fname: self.shard_index_cache[fname] \
      for fname in filenames \
      if fname in self.shard_index_cache  
    }

    requests = []
    for fname in filenames:
      if fname in fufilled:
        continue
      requests.append({
        'path': fname, 
        'local_alias': fname + '.index', 
        'start': 0, 
        'end': self.spec.index_length(),
      })

    progress = 'Shard Indices' if progress else False
    binaries = self.cache.download_as(requests, progress=progress)
    for (fname, start, end), content in binaries.items():
      try:
        index = self.decode_index(content, fname)
        self.shard_index_cache[fname] = index
        fufilled[fname] = index
      except EmptyFileException:
        self.shard_index_cache[fname] = None
        fufilled[fname] = None

    return fufilled

  def decode_index(self, binary, filename='Shard'):
    if binary is None or len(binary) == 0:
      raise EmptyFileException(filename + " was zero bytes.")
    elif len(binary) != self.spec.index_length():
      raise SpecViolation(
        filename + ": shard index was an incorrect length ({}) for this specification ({}).".format(
          len(binary), self.spec.index_length()
        ))
    
    index = np.frombuffer(binary, dtype=np.uint64)
    index = index.reshape( (index.size // 2, 2), order='C' )
    return index + self.spec.index_length()

  def decode_minishard_index(self, minishard_index, filename=''):
    """Returns [[label, offset, size], ... ] where offset and size are in bytes."""

    if self.spec.minishard_index_encoding != 'raw':
      minishard_index = compression.decompress(
        minishard_index, encoding=self.spec.minishard_index_encoding, filename=filename
      )

    minishard_index = np.copy(np.frombuffer(minishard_index, dtype=np.uint64))
    minishard_index = minishard_index.reshape( (3, len(minishard_index) // 3), order='C' ).T

    minishard_index[:,0] = np.cumsum(minishard_index[:,0])
    minishard_index[:,1] = np.cumsum(minishard_index[:,1])
    minishard_index[1:,1] += np.cumsum(minishard_index[:-1,2])
    minishard_index[:,1] += self.spec.index_length()

    return minishard_index 

  def get_minishard_index(self, filename, index, minishard_no, path=""):
    """
    Retrieves the minishard index for a given minishard number.

    Returns: uint64 Nx3 array with multiple rows of [segid, byte start, byte end]
    """
    res = self.get_minishard_indices(filename, index, minishard_no, path)
    return res[minishard_no]

  def get_minishard_indices(self, filename, index, minishard_nos, path=""):
    """
    Retrieves the minishard indices for a set of minishard numbers.

    Returns: { minishard_no: uint64 Nx3 array of [segid, byte start, byte end], ... }
    """
    res = self.get_minishard_indices_for_files(( (filename, index, minishard_nos), ), path)
    return res[basename(filename)]

  def get_minishard_indices_for_files(self, requests, path="", progress=None):
    """
    Fetches the specified minishard indices for all the specified files
    at once. This is required to get high performance as opposed to fetching
    the all minishard indices for a single file.

    requests: iterable of tuples
      [  (filename, index, minishard_numbers), ... ]

    Returns: map of filename -> minishard numbers -> minishard indices

    e.g. 
    {
      filename_1: {
          0: uint64 Nx3 array of [segid, byte start, byte end],
          1: ...,
      }
      filename_2: ...
    }
    """
    fufilled_by_filename = defaultdict(dict)
    msn_map = {}

    download_requests = []
    for filename, index, minishard_nos in requests:
      fufilled_requests, pending_requests = self.compute_minishard_index_requests(
        filename, index, minishard_nos, path
      ) 
      fufilled_by_filename[filename] = fufilled_requests
      for msn, start, end in pending_requests:
        msn_map[(basename(filename), start, end)] = msn

        filepath = self.meta.join(path, filename)

        download_requests.append({
          'path': filepath,
          'local_alias': '{}-{}.msi'.format(filepath, msn),
          'start': start,
          'end': end,
        })

    progress = 'Minishard Indices' if progress else False
    results = self.cache.download_as(download_requests, progress=progress)
  
    for (filename, start, end), content in results.items():
      filename = basename(filename)
      cache_key = (filename, start, end)
      msn = msn_map[cache_key]
      minishard_index = self.decode_minishard_index(content, filename)
      self.minishard_index_cache[cache_key] = minishard_index
      fufilled_by_filename[filename][msn] = minishard_index

    return fufilled_by_filename

  def compute_minishard_index_requests(self, filename, index, minishard_nos, path=""):
    """
    Helper method for get_minishard_indices_for_files. 
    Computes which requests must be made over the network vs can be fufilled from LRU cache.
    """
    minishard_nos = toiter(minishard_nos)

    if index is None:
      return ({ msn: None for msn in minishard_nos }, [])

    fufilled_requests = {}

    byte_ranges = {}
    for msn in minishard_nos:
      bytes_start, bytes_end = index[msn]

      # most typically: [0,0] for an incomplete shard
      if bytes_start == bytes_end:
        fufilled_requests[msn] = None
        continue

      bytes_start, bytes_end = int(bytes_start), int(bytes_end)
      byte_ranges[msn] = (bytes_start, bytes_end)

    full_path = self.meta.join(self.meta.cloudpath, path)

    pending_requests = []
    for msn, (bytes_start, bytes_end) in byte_ranges.items():
      cache_key = (filename, bytes_start, bytes_end)
      if cache_key in self.minishard_index_cache:
        fufilled_requests[msn] = self.minishard_index_cache[cache_key]
      else:
        pending_requests.append((msn, bytes_start, bytes_end))

    return (fufilled_requests, pending_requests)

  def exists(self, labels, path="", return_byte_range=False, progress=None):
    """
    Checks a shard's minishard index for whether a file exists.

    If return_byte_range = False:
      OUTPUT = SHARD_FILEPATH or None if not exists
    Else:
      OUTPUT = [ SHARD_FILEPATH or None, byte_start, num_bytes ]

    Returns:
      If labels is not an iterable:
        return OUTPUT
      Else:
        return { label_1: OUTPUT, label_2: OUTPUT, ... }
    """
    return_one = False

    try:
      iter(labels)
    except TypeError:
      return_one = True

    to_labels = defaultdict(list)
    to_all_labels = defaultdict(list)
    filename_to_minishard_num = defaultdict(list)

    for label in set(toiter(labels)):
      filename, minishard_number = self.compute_shard_location(label)
      to_labels[(filename, minishard_number)].append(label)
      to_all_labels[filename].append(label)
      filename_to_minishard_num[filename].append(minishard_number)

    indices = self.get_indices(to_all_labels.keys(), path, progress=progress)

    all_minishards = self.get_minishard_indices_for_files([ 
      (basename(filepath), index, filename_to_minishard_num[basename(filepath)]) \
      for filepath, index in indices.items()
    ], path, progress=progress)

    results = {}
    for filename, file_minishards in all_minishards.items():
      filepath = self.meta.join(path, filename)
      for mini_no, msi in file_minishards.items():
        labels = to_labels[(filename, mini_no)]

        for label in labels:
          if msi is None:
            results[label] = None
            continue

          idx = np.where(msi[:,0] == label)[0]
          if len(idx) == 0:
            results[label] = None
          else:
            if return_byte_range:
              _, offset, size = msi[idx,:][0]
              results[label] = [ filepath, int(offset), int(size) ]
            else:
              results[label] = filepath

    if return_one:
      return(list(results.values())[0])
    return results

  def disassemble_shard(self, shard):
    """
    Given an entire shard as a bytestring, convert 
    it into a dict of { label: byte content }.
    """
    index = self.decode_index(shard[:self.spec.index_length()])
    shattered = {}
    for start, end in index:
      start, end = int(start), int(end)
      if start == end:
        continue

      msi = self.decode_minishard_index(shard[start:end])
      for label, offset, size in msi:
        offset, size = int(offset), int(size)
        binary = shard[offset:offset+size]
        
        if self.spec.data_encoding != 'raw':
          binary = compression.decompress(binary, encoding=self.spec.data_encoding)
        
        shattered[label] = binary

    return shattered

  def get_data(
    self, label, path="", 
    progress=None, parallel=1
  ):
    """Fetches data from shards.

    label: one or more segment ids
    path: subdirectory path
    progress: display progress bars
    parallel: (int >= 0) use multiple processes

    Return: 
      if label is a scalar:
        a byte string
      else: (label is an iterable)
        {
          label_1: byte string,
          ....
        }
    """
    label, return_multiple = toiter(label, is_iter=True)
    label = set(( int(l) for l in label))
    if not label:
      return {}

    cached = {}
    if self.cache.enabled:
      cached = self.cache.get([ 
        self.meta.join(path, str(lbl)) for lbl in label
      ], progress=progress)

    results = {}
    for cloudpath, content in cached.items():
      lbl = int(basename(cloudpath))
      if content is not None:
        label.remove(lbl)
        
      results[lbl] = cached[cloudpath]

    del cached

    # { label: [ filename, byte start, num_bytes ] }
    exists = self.exists(label, path, return_byte_range=True, progress=progress)
    for k in list(exists.keys()):
      if exists[k] is None:
        results[k] = None
        del exists[k]

    key_label = { (basename(v[0]), v[1], v[2]): k for k,v in exists.items() }

    files = ( 
      { 'path': basename(ext[0]), 'start': int(ext[1]), 'end': int(ext[1]) + int(ext[2]) }
      for ext in exists.values()
    )

    # Requesting many individual shard chunks is slow, but due to z-ordering
    # we might be able to combine adjacent byte ranges. Especially helpful
    # when downloading entire shards!
    bundles = []
    for chunk in sorted(files, key=itemgetter("path", "start")):
      if not bundles or (chunk['path'] != bundles[-1]['path']) or (chunk['start'] != bundles[-1]['end']):
        bundles.append(dict(content=None, subranges=[], **chunk))
      else:
        bundles[-1]['end'] = chunk['end']

      bundles[-1]['subranges'].append({
          'start': chunk['start'],
          'length': chunk['end'] - chunk['start'],
          'slices': slice(chunk['start'] - bundles[-1]['start'], chunk['end'] - bundles[-1]['start'])
      })

    full_path = self.meta.join(self.meta.cloudpath, path)
    bundles_resp = CloudFiles(
      full_path, 
      progress=progress, 
      green=self.green,
      parallel=parallel,
    ).get(bundles)

    # Responses are not guaranteed to be in order of requests
    bundles_resp = { (r['path'], r['byte_range']): r for r in bundles_resp }

    binaries = {}
    for bundle_req in bundles:
      bundle_resp = bundles_resp[(bundle_req['path'], (bundle_req['start'], bundle_req['end']))]
      if bundle_resp['error']:
        raise bundle_resp['error']

      for chunk in bundle_req['subranges']:
        key = (bundle_req['path'], chunk['start'], chunk['length'])
        lbl = key_label[key]
        binaries[lbl] = bundle_resp['content'][chunk['slices']]

    del bundles
    del bundles_resp

    if self.spec.data_encoding != 'raw':
      for filepath, binary in tqdm(binaries.items(), desc="Decompressing", disable=(not progress)):
        if binary is None:
          continue
        binaries[filepath] = compression.decompress(
          binary, encoding=self.spec.data_encoding, filename=filepath
        )
    
    if self.cache.enabled:
      self.cache.put([ 
        (self.meta.join(path, str(filepath)), binary) for filepath, binary in binaries.items()
      ], progress=progress)

    results.update(binaries)

    if return_multiple:
      return results
    return first(results.values())

  def list_labels(self, filename, path="", size=False):
    """
    List all the labels in the index of a given shard file.

    size: (bool) if True, list the size in bytes of each label

    Returns: 
      if not size:
        np.uint64 array of labels 
      else:
        [ (label, size in bytes), ... ] in descending order of size
    """
    index = self.get_index(filename, path)
    all_minishard_nos = list(range(len(index)))
    minishard_indices = self.get_minishard_indices(filename, index, all_minishard_nos, path)
    minishard_indices = [  
      msi for msi in minishard_indices.values() if msi is not None
    ]
    if not size:
      labels = np.concatenate([  
        msi[:,0] for msi in minishard_indices
      ])
      return np.sort(labels)
    else:
      labels = np.concatenate([  
        msi[:,:]
        for msi in minishard_indices
      ])
      labels = [ (row[0], row[2]) for row in labels[:] ]
      return sorted(labels, key=lambda x: x[1], reverse=True) class Mesh(object):
  """
  Represents the vertices, faces, and normals of a mesh
  as numpy arrays.

  class Mesh:
    ndarray[float32, ndim=2] self.vertices: [ [x,y,z], ... ]
    ndarray[uint32,  ndim=2] self.faces:    [ [v1,v2,v3], ... ]
    ndarray[float32, ndim=2] self.normals:  [ [nx,ny,nz], ... ]
  """
  def __init__(
    self, vertices, faces, normals=None, 
    segid=None, encoding_type=None, encoding_options=None
  ):
    self.vertices = np.array(vertices, dtype=np.float32)
    self.faces = np.array(faces, dtype=np.uint32)

    if normals is None:
      self.normals = np.array([], dtype=np.float32).reshape((0,3))
    else:
      self.normals = np.array(normals, dtype=np.float32)

    self.segid = segid
    self.encoding_type = encoding_type
    self.encoding_options = encoding_options 

  def __len__(self):
    return self.vertices.shape[0]

  def __eq__(self, other):
    """Tests strict equality between two meshes."""

    no_self_normals = self.normals is None or self.normals.size == 0
    no_other_normals = other.normals is None or other.normals.size == 0

    if no_self_normals != no_other_normals:
      return False
       
    equality = np.all(self.vertices == other.vertices) \
      and np.all(self.faces == other.faces)

    if no_self_normals:
      return equality

    return (equality and np.all(self.normals == other.normals))

  def __repr__(self):
    return "Mesh(vertices<{}>, faces<{}>, normals<{}>, segid={}, encoding_type=<{}>)".format(
      self.vertices.shape[0], self.faces.shape[0], self.normals.shape[0],
      self.segid, self.encoding_type
    )

  def __getitem__(self, key):
    val = None 
    if key == 'vertices':
      val = self.vertices
    elif key == 'num_vertices':
      val = len(self)
    elif key == 'faces':
      val = self.faces
    else:
      raise KeyError("{} not found.".format(key))

    deprecation_notice(key)
    return val

  def empty(self):
    return self.vertices.size == 0 or self.faces.size == 0

  def clone(self):
    return Mesh(
      np.copy(self.vertices), np.copy(self.faces), np.copy(self.normals),
      self.segid, 
      encoding_type=copy.deepcopy(self.encoding_type),
      encoding_options=copy.deepcopy(self.encoding_options),
    )

  def edges(self):
    """
    Generate an edge list from the faces. 
    edges are not guaranteed to be unique.
    """
    srt = lambda x,y: (x,y) if x < y else (y,x)
    for face in self.faces:
      yield srt(face[0], face[1])
      yield srt(face[1], face[2])
      yield srt(face[0], face[2])

  def triangles(self):
    """
    Faces are numbered using the index of vertices,
    but sometimes it is convenient to have a list 
    of triangles in their proper coordinate space.
    """
    Nf = self.faces.shape[0]
    tris = np.zeros( (Nf, 3, 3), dtype=np.float32, order='C' ) # triangle, vertices, (x,y,z)

    for i in range(Nf):
      for j in range(3):
        tris[i,j,:] = self.vertices[ self.faces[i,j] ]

    return tris

  @classmethod
  def concatenate(cls, *meshes):
    vertex_ct = np.zeros(len(meshes) + 1, np.uint32)
    vertex_ct[1:] = np.cumsum([ len(mesh) for mesh in meshes ])

    vertices = np.concatenate([ mesh.vertices for mesh in meshes ])
    
    faces = np.concatenate([ 
      mesh.faces + vertex_ct[i] for i, mesh in enumerate(meshes) 
    ])

    normals = np.concatenate([ mesh.normals for mesh in meshes ])

    encoding_type = list(set([ mesh.encoding_type for mesh in meshes ]))
    if len(encoding_type) == 1:
      encoding_type = encoding_type[0]

    return Mesh(vertices, faces, normals, encoding_type=encoding_type)

  def consolidate(self):
    """Remove duplicate vertices and faces. Returns a new mesh object."""
    if self.empty():
      return Mesh([], [], normals=None)

    vertices = self.vertices
    faces = self.faces
    normals = self.normals

    eff_verts, uniq_idx, idx_representative = np.unique(
      vertices, axis=0, return_index=True, return_inverse=True
    )

    face_vector_map = np.vectorize(lambda x: idx_representative[x])
    eff_faces = face_vector_map(faces)
    eff_faces = np.unique(eff_faces, axis=0)

    # normal_vector_map = np.vectorize(lambda idx: normals[idx])
    # eff_normals = normal_vector_map(uniq_idx)

    return Mesh(eff_verts, eff_faces, None, 
      segid=self.segid,
      encoding_type=copy.deepcopy(self.encoding_type),
      encoding_options=copy.deepcopy(self.encoding_options),
    )

  @classmethod
  def from_precomputed(self, binary, segid=None):
    """
    Mesh from_precomputed(self, binary)

    Decode a Precomputed format mesh from a byte string.
    
    Format:
      uint32        Nv * float32 * 3   uint32 * 3 until end
      Nv            (x,y,z)            (v1,v2,v2)
      N Vertices    Vertices           Faces
    """
    num_vertices = struct.unpack("=I", binary[0:4])[0]
    try:
      # count=-1 means all data in buffer
      vertices = np.frombuffer(binary, dtype=np.float32, count=3*num_vertices, offset=4)
      faces = np.frombuffer(binary, dtype=np.uint32, count=-1, offset=(4 + 12 * num_vertices)) 
    except ValueError:
      raise MeshDecodeError("""
        The input buffer is too small for the Precomputed format.
        Minimum Bytes: {} 
        Actual Bytes: {}
      """.format(4 + 4 * num_vertices, len(binary)))

    vertices = vertices.reshape(num_vertices, 3)
    faces = faces.reshape(faces.size // 3, 3)

    return Mesh(
      vertices, faces, 
      segid=segid, 
      normals=None, 
      encoding_type='precomputed'
    )

  def to_precomputed(self):
    """
    bytes to_precomputed(self)

    Convert mesh into binary format compatible with Neuroglancer.
    Does not preserve normals.
    """
    vertex_index_format = [
      np.uint32(self.vertices.shape[0]), # Number of vertices (3 coordinates)
      self.vertices,
      self.faces
    ]
    return b''.join([ array.tobytes('C') for array in vertex_index_format ])

  @classmethod
  def from_obj(self, text, segid=None):
    """Given a string representing a Wavefront OBJ file, decode to a Mesh."""

    vertices = []
    faces = []
    normals = []

    if type(text) is bytes:
      text = text.decode('utf8')

    for line in text.split('\n'):
      line = line.strip()
      if len(line) == 0:
        continue
      elif line[0] == '#':
        continue
      elif line[0] == 'f':
        if line.find('/') != -1:
          # e.g. f 6092/2095/6079 6087/2092/6075 6088/2097/6081
          (v1, vt1, vn1, v2, vt2, vn2, v3, vt3, vn3) = re.match(r'f\s+(\d+)/(\d*)/(\d+)\s+(\d+)/(\d*)/(\d+)\s+(\d+)/(\d*)/(\d+)', line).groups()
        else:
          (v1, v2, v3) = re.match(r'f\s+(\d+)\s+(\d+)\s+(\d+)', line).groups()
        faces.append( (int(v1), int(v2), int(v3)) )
      elif line[0] == 'v':
        if line[1] == 't': # vertex textures not supported
          # e.g. vt 0.351192 0.337058
          continue 
        elif line[1] == 'n': # vertex normals
          # e.g. vn 0.992266 -0.033290 -0.119585
          (n1, n2, n3) = re.match(r'vn\s+([-\d\.]+)\s+([-\d\.]+)\s+([-\d\.]+)', line).groups()
          normals.append( (float(n1), float(n2), float(n3)) )
        else:
          # e.g. v -0.317868 -0.000526 -0.251834
          (v1, v2, v3) = re.match(r'v\s+([-\d\.]+)\s+([-\d\.]+)\s+([-\d\.]+)', line).groups()
          vertices.append( (float(v1), float(v2), float(v3)) )

    vertices = np.array(vertices, dtype=np.float32)
    faces = np.array(faces, dtype=np.uint32)
    normals = np.array(normals, dtype=np.float32)

    return Mesh(vertices, faces - 1, normals, segid=segid)

  def to_obj(self):
    """Return a string representing a .obj file."""
    objdata = []
    objdata += [ 'v {:.5f} {:.5f} {:.5f}'.format(*vertex) for vertex in self.vertices ]
    objdata += [ 'f {} {} {}'.format(*face) for face in (self.faces+1) ] # obj is 1 indexed
    objdata = '\n'.join(objdata) + '\n'
    return objdata.encode('utf8')

  def to_ply(self):
    """
    Return a bytearray in .ply format, 
    a more compact format than .obj that's still widely readable.
    """
    vertexct = self.vertices.shape[0]
    trianglect = self.faces.shape[0]

    # Header
    plydata = bytearray("""ply
format binary_little_endian 1.0
element vertex {}
property float x
property float y
property float z
element face {}
property list int int vertex_indices
end_header
""".format(vertexct, trianglect).encode('utf8'))

    # Vertex data (x y z): "fff" 
    plydata.extend(self.vertices.tobytes('C'))

    # Faces (3 f1 f2 f3): "3iii" 
    plydata.extend(
      np.insert(self.faces, 0, 3, axis=1).tobytes('C')
    )

    return plydata

  @classmethod
  def from_draco(cls, binary, segid=None):
    import DracoPy

    try:
      mesh = DracoPy.decode(binary)
    except ValueError:
      raise MeshDecodeError("Not a valid draco mesh.")

    return Mesh(
      mesh.points, mesh.faces, 
      segid=segid,
      normals=None,
      encoding_type='draco', 
      encoding_options=mesh.encoding_options
    )

  def deduplicate_vertices(self, is_chunk_aligned):
    faces = self.faces
    verts = self.vertices
    # find all vertices that have exactly 2 duplicates
    unique_vertices, unique_inverse, counts = np.unique(
      verts, return_inverse=True, return_counts=True, axis=0
    )
    
    only_double = np.where(counts == 2)[0]
    is_doubled = np.isin(unique_inverse, only_double)
    # this stores whether each vertex should be merged or not
    do_merge = np.array(is_doubled & is_chunk_aligned)

    # setup an artificial 4th coordinate for vertex positions
    # which will be unique in general, 
    # but then repeated for those that are merged
    new_vertices = np.hstack((verts, np.arange(verts.shape[0])[:, np.newaxis]))
    new_vertices[do_merge, 3] = -1
  
    faces = faces.flatten()

    # use unique to make the artificial vertex list unique and reindex faces
    vertices, newfaces = np.unique(new_vertices[faces], return_inverse=True, axis=0)
    newfaces = newfaces.astype(np.uint32).reshape( (len(newfaces) // 3, 3) )

    return Mesh(vertices[:,0:3], newfaces, None, segid=self.segid, 
      encoding_type=self.encoding_type, encoding_options=self.encoding_options
    )

  def deduplicate_chunk_boundaries(self, chunk_size, is_draco=False, draco_grid_size=None, offset=(0,0,0)):
    offset = Vec(*offset)
    verts = self.vertices - offset
    # find all vertices that are exactly on chunk_size boundaries
    if is_draco:
      if draco_grid_size is None:
        raise ValueError('Must specify draco grid size to dedup draco meshes')
      is_chunk_aligned = is_draco_chunk_aligned(verts, chunk_size, draco_grid_size=draco_grid_size)
    else:
      is_chunk_aligned = np.any(np.mod(verts, chunk_size) == 0, axis=1)

    return self.deduplicate_vertices(is_chunk_aligned)

  def viewer(self):
    try:
      import vtk
    except ImportError:
      print("The mesh viewer requires the OpenGL based vtk. Try: pip install vtk --upgrade")
      return

    colors = vtk.vtkNamedColors()
    # Set the background color.
    bkg = map(lambda x: x / 255.0, [40, 40, 40, 255])
    colors.SetColor("BkgColor", *bkg)

    # This creates a point cloud model
    points = vtk.vtkPoints()
    verts = vtk.vtkCellArray()
    polydata = vtk.vtkPolyData()
    polydata.SetPoints(points)
    polydata.SetVerts(verts)
    for vertex in self.vertices:
      pid = points.InsertNextPoint(vertex)
      verts.InsertNextCell(1)
      verts.InsertCellPoint(pid)

    # The mapper is responsible for pushing the geometry into the graphics
    # library. It may also do color mapping, if scalars or other
    # attributes are defined.
    mapper = vtk.vtkPolyDataMapper()
    mapper.SetInputData(polydata)

    # The actor is a grouping mechanism: besides the geometry (mapper), it
    # also has a property, transformation matrix, and/or texture map.
    # Here we set its color and rotate it -22.5 degrees.
    cylinderActor = vtk.vtkActor()
    cylinderActor.SetMapper(mapper)
    cylinderActor.GetProperty().SetColor(colors.GetColor3d("Mint"))
    cylinderActor.RotateX(30.0)
    cylinderActor.RotateY(-45.0)

    # Create the graphics structure. The renderer renders into the render
    # window. The render window interactor captures mouse events and will
    # perform appropriate camera or actor manipulation depending on the
    # nature of the events.
    ren = vtk.vtkRenderer()
    renWin = vtk.vtkRenderWindow()
    renWin.AddRenderer(ren)
    iren = vtk.vtkRenderWindowInteractor()
    iren.SetRenderWindow(renWin)

    # Add the actors to the renderer, set the background and size
    ren.AddActor(cylinderActor)
    ren.SetBackground(colors.GetColor3d("BkgColor"))
    renWin.SetSize(800, 800)

    text = "Point Cloud Rendering of Mesh Object"
    if self.segid is not None:
      renWin.SetWindowName(text + " (Label {})".format(self.segid))
    else:
      renWin.SetWindowName(text)

    # This allows the interactor to initalize itself. It has to be
    # called before an event loop.
    iren.Initialize()

    # We'll zoom in a little by accessing the camera and invoking a "Zoom"
    # method on it.
    ren.ResetCamera()
    ren.GetActiveCamera().Zoom(1.5)
    renWin.Render()

    # Start the event loop.
    iren.Start()     def yellow(text):
  return colorize('yellow', text) def red(text):
  return colorize('red', text) def toiter(obj, is_iter=False):
  if isinstance(obj, str) or isinstance(obj, dict):
    if is_iter:
      return [ obj ], False
    return [ obj ]

  try:
    iter(obj)
    if is_iter:
      return obj, True
    return obj 
  except TypeError:
    if is_iter:
      return [ obj ], False
    return [ obj ] def first(lst):
  if isinstance(lst, types.GeneratorType):
    try:
      return next(lst)
    except StopIteration:
      return None
  try:
    return lst[0]
  except TypeError:
    try:
      return next(iter(lst))
    except StopIteration:
      return None
  except IndexError:
    return None class AuthenticationError(BaseException):
class DimensionError(BaseException):
class InfoError(Exception):
class InfoUnavailableError(InfoError):
class RedirectError(InfoError):
class TooManyRedirects(RedirectError):
class CyclicRedirect(RedirectError):
class ScaleUnavailableError(IndexError):
class ReadOnlyException(Exception):
class AlignmentError(ValueError):
class EmptyVolumeException(Exception):
class EmptyFileException(Exception):
class EmptyRequestException(ValueError):
class DecodingError(Exception):
class EncodingError(Exception):
class OutOfBoundsError(ValueError):
class UnsupportedCompressionType(ValueError):
class MeshDecodeError(ValueError):
class SkeletonUnassignedEdgeError(Exception):
class SkeletonDecodeError(Exception):
class SkeletonEncodeError(Exception):
class SkeletonTransformError(Exception):
class SkeletonAttributeMixingError(Exception):
class UnsupportedFormatError(Exception):
class UnsupportedProtocolError(ValueError):
class UnsupportedGrapheneAPIVersionError(Exception):
class SpecViolation(Exception):
class SpatialIndexGapError(Exception):
class WriteLockAcquisitionError(Exception):
class WriteLockReleaseError(Exception):
from collections import defaultdict
from cloudfiles import CloudFiles
from .common import apply_transform
from .unsharded import UnshardedLegacyPrecomputedMeshSource
from ..sharding import ShardingSpecification, ShardReader
from ....mesh import Mesh
from ....lib import yellow, red, toiter, first
from .... import exceptions
import re
import struct
import numpy as np
import fastremap
  return meshdata

class UnshardedMultiLevelPrecomputedMeshSource(UnshardedLegacyPrecomputedMeshSource):
  def __init__(self, meta, cache, config, readonly=False):
    super().__init__(meta, cache, config, readonly)

    self.vertex_quantization_bits = self.meta.info['vertex_quantization_bits']
    self.lod_scale_multiplier = self.meta.info['lod_scale_multiplier']
    self.transform = np.array(self.meta.info['transform'] + [0,0,0,1]).reshape(4,4)
  
  @property
  def path(self):
    return self.meta.mesh_path

  def get_manifest(self, segid, progress=None):
    """Retrieve the manifest for one or more segments."""
    segid, multiple_return = toiter(segid, is_iter=True)
    progress = progress if progress is not None else self.config.progress

    cloudpath = self.meta.join(self.meta.cloudpath, self.path)
    cf = CloudFiles(cloudpath, progress=progress)
    results = cf.get((f"{sid}.index" for sid in segid ), total=len(segid))

    if not multiple_return:
      if not results:
        return None
      binary = results[0]["content"]
      if binary is None:
        return None
      return MultiLevelPrecomputedMeshManifest.from_binary(binary, segment_id=first(segid), shard_offset=0)

    regexp = re.compile(r'(\d+)\.index$')
    manifests = []
    for res in results:
      key = res["path"]
      sid = int(re.match(regexp, key).groups()[0])
      binary = res["content"]
      if binary is None:
        manifests.append(None)
      manifest = MultiLevelPrecomputedMeshManifest.from_binary(binary, segment_id=sid, shard_offset=0)
      manifests.append(manifest)

    return manifests

  def exists(self, segids, progress=None):
    """
    Checks if the mesh exists

    Returns: { MultiLevelPrecomputedMeshManifest or None, ... }
    """
    cf = CloudFiles(self.path)
    return cf.exists(( f"{segid}.index" for segid in segids ))

  def get(self, segids, lod=0, concat=True, progress=None):
    """Fetch meshes at a given level of detail (lod).

    Parameters:
    segids: (iterable or int) segids to render

    lod: int, default 0
      Level of detail to retrieve.  0 is highest level of detail.

    Optional:
      concat: bool, concatenate fragments (per segment per lod)

    Returns:
    { segid: { Mesh } }
    ... or if concatenate=False: { segid: { Mesh, ... } }

    Reference:
      https://github.com/google/neuroglancer/blob/master/src/neuroglancer/datasource/precomputed/meshes.md
    """
    if lod < 0:
      raise exceptions.ValueError(red(f'lod ({lod}) must be >= 0.'))

    progress = progress if progress is not None else self.config.progress
    segids = toiter(segids)

    # decode all the fragments
    manifests = self.get_manifest(segids)
    for manifest in manifests:
      if manifest is None:
        raise exceptions.MeshDecodeError(red(
          f'Manifest not found for segment {manifest.segment_id}.'
        ))
      if lod >= manifest.num_lods:
        raise exceptions.MeshDecodeError(red(
          f'LOD value ({lod}) out of range (0 - {manifest.num_lods - 1}) for segment {manifest.segment_id}.'
        ))

    full_path = self.meta.join(self.meta.cloudpath, self.path)

    meshdata = defaultdict(list)
    for manifest in manifests:
      # Read the manifest (with a tweak to sharding.py to get the offset)
      fragment_sizes = [ 
        np.sum(lod_fragment_sizes) for lod_fragment_sizes in manifest.fragment_offsets 
      ]
      
      lod_binary = CloudFiles(
        full_path, progress=progress, 
        green=self.config.green, secrets=self.config.secrets
      ).get({
        'path': str(manifest.segment_id),
        'start': int(np.sum(fragment_sizes[0:lod])),
        'end': int(np.sum(fragment_sizes[0:lod+1])),  
      })

      meshes = extract_lod_meshes(
        manifest, lod, lod_binary, 
        self.vertex_quantization_bits, self.transform
      )
      meshdata.update(meshes)

    if concat:
      for segid in meshdata:
        meshdata[segid] = Mesh.concatenate(*meshdata[segid])

    return meshdata


class ShardedMultiLevelPrecomputedMeshSource(UnshardedLegacyPrecomputedMeshSource):