:py:mod:`kwcoco.channel_spec`
=============================

.. py:module:: kwcoco.channel_spec

.. autoapi-nested-parse::

   The ChannelSpec has these simple rules:

   .. code::

       * each 1D channel is a alphanumeric string.

       * The pipe ('|') separates aligned early fused stremas (non-communative)

       * The comma (',') separates late-fused streams, (happens after pipe operations, and is communative)

       * Certain common sets of early fused channels have codenames, for example:

           rgb = r|g|b
           rgba = r|g|b|a
           dxdy = dy|dy

       * Multiple channels can be specified via a "slice" notation. For example:

           mychan.0:4

           represents 4 channels:
               mychan.0, mychan.1, mychan.2, and mychan.3

           slices after the "." work like python slices

   For single arrays, the spec is always an early fused spec.

   .. todo::

      - [X] : normalize representations? e.g: rgb = r|g|b? - OPTIONAL
      - [X] : rename to BandsSpec or SensorSpec? - REJECTED
      - [ ] : allow bands to be coerced, i.e. rgb -> gray, or gray->rgb

   .. todo::

      - [x]: Use FusedChannelSpec as a member of ChannelSpec
      - [x]: Handle special slice suffix for length calculations

   .. note:: * do not specify the same channel in FusedChannelSpec twice

   .. rubric:: Example

   >>> import kwcoco
   >>> spec = kwcoco.ChannelSpec('b1|b2|b3,m.0:4|x1|x2,x.3|x.4|x.5')
   >>> print(spec)
   <ChannelSpec(b1|b2|b3,m.0:4|x1|x2,x.3|x.4|x.5)>
   >>> for stream in spec.streams():
   >>>     print(stream)
   <FusedChannelSpec(b1|b2|b3)>
   <FusedChannelSpec(m.0:4|x1|x2)>
   <FusedChannelSpec(x.3|x.4|x.5)>
   >>> # Normalization
   >>> normalized = spec.normalize()
   >>> print(normalized)
   <ChannelSpec(b1|b2|b3,m.0|m.1|m.2|m.3|x1|x2,x.3|x.4|x.5)>
   >>> print(normalized.fuse().spec)
   b1|b2|b3|m.0|m.1|m.2|m.3|x1|x2|x.3|x.4|x.5
   >>> print(normalized.fuse().concise().spec)
   b1|b2|b3|m:4|x1|x2|x.3:6



Module Contents
---------------

Classes
~~~~~~~

.. autoapisummary::

   kwcoco.channel_spec.BaseChannelSpec
   kwcoco.channel_spec.FusedChannelSpec
   kwcoco.channel_spec.ChannelSpec



Functions
~~~~~~~~~

.. autoapisummary::

   kwcoco.channel_spec._cached_single_fused_mapping
   kwcoco.channel_spec._cached_single_stream_idxs
   kwcoco.channel_spec.subsequence_index
   kwcoco.channel_spec._parse_concise_slice_syntax
   kwcoco.channel_spec.oset_insert
   kwcoco.channel_spec.oset_delitem



.. py:class:: BaseChannelSpec

   Bases: :py:obj:`ubelt.NiceRepr`

   Common code API between :class:`FusedChannelSpec` and :class:`ChannelSpec`

   .. todo::

      - [ ] Keep working on this base spec and ensure the inheriting classes
            conform to it.

   .. py:method:: spec(self)
      :property:

      The string encodeing of this spec

      :returns: str


   .. py:method:: coerce(cls, data)
      :classmethod:
      :abstractmethod:

      Try and interpret the input data as some sort of spec

      :Parameters: **data** (*str | int | list | dict | BaseChannelSpec*) -- any input data that is known to represent a spec

      :returns: BaseChannelSpec


   .. py:method:: streams(self)
      :abstractmethod:

      Breakup this spec into individual early-fused components

      :returns: List[FusedChannelSpec]


   .. py:method:: normalize(self)
      :abstractmethod:

      Expand all channel codes into their normalized long-form

      :returns: BaseChannelSpec


   .. py:method:: intersection(self)
      :abstractmethod:


   .. py:method:: difference(self)
      :abstractmethod:


   .. py:method:: __sub__(self, other)


   .. py:method:: __nice__(self)


   .. py:method:: __json__(self)


   .. py:method:: __and__(self, other)



.. py:class:: FusedChannelSpec(parsed, _is_normalized=False)

   Bases: :py:obj:`BaseChannelSpec`

   A specific type of channel spec with only one early fused stream.

   The channels in this stream are non-communative

   Behaves like a list of atomic-channel codes
   (which may represent more than 1 channel), normalized codes always
   represent exactly 1 channel.

   .. note:: This class name and API is in flux and subject to change.

   .. todo::

      A special code indicating a name and some number of bands that that
      names contains, this would primarilly be used for large numbers of
      channels produced by a network. Like:
      
          resnet_d35d060_L5:512
      
          or
      
          resnet_d35d060_L5[:512]
      
      might refer to a very specific (hashed) set of resnet parameters
      with 512 bands
      
      maybe we can do something slicly like:
      
          resnet_d35d060_L5[A:B]
          resnet_d35d060_L5:A:B
      
      Do we want to "just store the code" and allow for parsing later?
      
      Or do we want to ensure the serialization is parsed before we
      construct the data structure?

   .. rubric:: Example

   >>> from kwcoco.channel_spec import *  # NOQA
   >>> import pickle
   >>> self = FusedChannelSpec.coerce(3)
   >>> recon = pickle.loads(pickle.dumps(self))
   >>> self = ChannelSpec.coerce('a|b,c|d')
   >>> recon = pickle.loads(pickle.dumps(self))

   .. py:attribute:: _alias_lut
      

      

   .. py:attribute:: _memo
      

      

   .. py:attribute:: _size_lut
      

      

   .. py:method:: __len__(self)


   .. py:method:: __getitem__(self, index)


   .. py:method:: concat(cls, items)
      :classmethod:


   .. py:method:: spec(self)

      The string encodeing of this spec

      :returns: str


   .. py:method:: unique(self)


   .. py:method:: parse(cls, spec)
      :classmethod:


   .. py:method:: coerce(cls, data)
      :classmethod:

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> FusedChannelSpec.coerce(['a', 'b', 'c'])
      >>> FusedChannelSpec.coerce('a|b|c')
      >>> FusedChannelSpec.coerce(3)
      >>> FusedChannelSpec.coerce(FusedChannelSpec(['a']))
      >>> assert FusedChannelSpec.coerce('').numel() == 0


   .. py:method:: concise(self)

      Shorted the channel spec by de-normaliz slice syntax

      :returns: concise spec
      :rtype: FusedChannelSpec

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> self = FusedChannelSpec.coerce(
      >>>      'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0:3|c.0')
      >>> short = self.concise()
      >>> long = short.normalize()
      >>> numels = [c.numel() for c in [self, short, long]]
      >>> print('self.spec  = {!r}'.format(self.spec))
      >>> print('short.spec = {!r}'.format(short.spec))
      >>> print('long.spec  = {!r}'.format(long.spec))
      >>> print('numels = {!r}'.format(numels))
      self.spec  = 'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0:3|c.0'
      short.spec = 'b|a|a:3|a.5|c|a.8:10|b:3|c.0'
      long.spec  = 'b|a|a.0|a.1|a.2|a.5|c|a.8|a.9|b.0|b.1|b.2|c.0'
      numels = [13, 13, 13]
      >>> assert long.concise().spec == short.spec


   .. py:method:: normalize(self)

      Replace aliases with explicit single-band-per-code specs

      :returns: normalize spec
      :rtype: FusedChannelSpec

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> self = FusedChannelSpec.coerce('b1|b2|b3|rgb')
      >>> normed = self.normalize()
      >>> print('self = {}'.format(self))
      >>> print('normed = {}'.format(normed))
      self = <FusedChannelSpec(b1|b2|b3|rgb)>
      normed = <FusedChannelSpec(b1|b2|b3|r|g|b)>
      >>> self = FusedChannelSpec.coerce('B:1:11')
      >>> normed = self.normalize()
      >>> print('self = {}'.format(self))
      >>> print('normed = {}'.format(normed))
      self = <FusedChannelSpec(B:1:11)>
      normed = <FusedChannelSpec(B.1|B.2|B.3|B.4|B.5|B.6|B.7|B.8|B.9|B.10)>
      >>> self = FusedChannelSpec.coerce('B.1:11')
      >>> normed = self.normalize()
      >>> print('self = {}'.format(self))
      >>> print('normed = {}'.format(normed))
      self = <FusedChannelSpec(B.1:11)>
      normed = <FusedChannelSpec(B.1|B.2|B.3|B.4|B.5|B.6|B.7|B.8|B.9|B.10)>


   .. py:method:: numel(self)

      Total number of channels in this spec


   .. py:method:: sizes(self)

      Returns a list indicating the size of each atomic code

      :returns: List[int]

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> self = FusedChannelSpec.coerce('b1|Z:3|b2|b3|rgb')
      >>> self.sizes()
      [1, 3, 1, 1, 3]
      >>> assert(FusedChannelSpec.parse('a.0').numel()) == 1
      >>> assert(FusedChannelSpec.parse('a:0').numel()) == 0
      >>> assert(FusedChannelSpec.parse('a:1').numel()) == 1


   .. py:method:: __contains__(self, key)

      .. rubric:: Example

      >>> FCS = FusedChannelSpec.coerce
      >>> 'disparity' in FCS('rgb|disparity|flowx|flowy')
      True
      >>> 'gray' in FCS('rgb|disparity|flowx|flowy')
      False


   .. py:method:: code_list(self)

      Return the expanded code list


   .. py:method:: as_list(self)


   .. py:method:: as_oset(self)


   .. py:method:: as_set(self)


   .. py:method:: __set__(self)


   .. py:method:: difference(self, other)

      Set difference

      .. rubric:: Example

      >>> FCS = FusedChannelSpec.coerce
      >>> self = FCS('rgb|disparity|flowx|flowy')
      >>> other = FCS('r|b')
      >>> self.difference(other)
      >>> other = FCS('flowx')
      >>> self.difference(other)
      >>> FCS = FusedChannelSpec.coerce
      >>> assert len((FCS('a') - {'a'}).parsed) == 0
      >>> assert len((FCS('a.0:3') - {'a.0'}).parsed) == 2


   .. py:method:: intersection(self, other)

      .. rubric:: Example

      >>> FCS = FusedChannelSpec.coerce
      >>> self = FCS('rgb|disparity|flowx|flowy')
      >>> other = FCS('r|b|XX')
      >>> self.intersection(other)


   .. py:method:: component_indices(self, axis=2)

      Look up component indices within this stream

      .. rubric:: Example

      >>> FCS = FusedChannelSpec.coerce
      >>> self = FCS('disparity|rgb|flowx|flowy')
      >>> component_indices = self.component_indices()
      >>> print('component_indices = {}'.format(ub.repr2(component_indices, nl=1)))
      component_indices = {
          'disparity': (slice(...), slice(...), slice(0, 1, None)),
          'flowx': (slice(...), slice(...), slice(4, 5, None)),
          'flowy': (slice(...), slice(...), slice(5, 6, None)),
          'rgb': (slice(...), slice(...), slice(1, 4, None)),
      }


   .. py:method:: streams(self)

      Idempotence with :func:`ChannelSpec.streams`


   .. py:method:: fuse(self)

      Idempotence with :func:`ChannelSpec.streams`



.. py:class:: ChannelSpec(spec, parsed=None)

   Bases: :py:obj:`BaseChannelSpec`

   Parse and extract information about network input channel specs for
   early or late fusion networks.

   Behaves like a dictionary of FusedChannelSpec objects

   .. todo::

      - [ ] Rename to something that indicates this is a collection of
          FusedChannelSpec? MultiChannelSpec?

   .. note:: This class name and API is in flux and subject to change.

   .. note::

      The pipe ('|') character represents an early-fused input stream, and
      order matters (it is non-communative).
      
      The comma (',') character separates different inputs streams/branches
      for a multi-stream/branch network which will be lated fused. Order does
      not matter

   .. rubric:: Example

   >>> from kwcoco.channel_spec import *  # NOQA
   >>> # Integer spec
   >>> ChannelSpec.coerce(3)
   <ChannelSpec(u0|u1|u2) ...>

   >>> # single mode spec
   >>> ChannelSpec.coerce('rgb')
   <ChannelSpec(rgb) ...>

   >>> # early fused input spec
   >>> ChannelSpec.coerce('rgb|disprity')
   <ChannelSpec(rgb|disprity) ...>

   >>> # late fused input spec
   >>> ChannelSpec.coerce('rgb,disprity')
   <ChannelSpec(rgb,disprity) ...>

   >>> # early and late fused input spec
   >>> ChannelSpec.coerce('rgb|ir,disprity')
   <ChannelSpec(rgb|ir,disprity) ...>

   .. rubric:: Example

   >>> self = ChannelSpec('gray')
   >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1)))
   >>> self = ChannelSpec('rgb')
   >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1)))
   >>> self = ChannelSpec('rgb|disparity')
   >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1)))
   >>> self = ChannelSpec('rgb|disparity,disparity')
   >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1)))
   >>> self = ChannelSpec('rgb,disparity,flowx|flowy')
   >>> print('self.info = {}'.format(ub.repr2(self.info, nl=1)))

   .. rubric:: Example

   >>> specs = [
   >>>     'rgb',              # and rgb input
   >>>     'rgb|disprity',     # rgb early fused with disparity
   >>>     'rgb,disprity',     # rgb early late with disparity
   >>>     'rgb|ir,disprity',  # rgb early fused with ir and late fused with disparity
   >>>     3,                  # 3 unknown channels
   >>> ]
   >>> for spec in specs:
   >>>     print('=======================')
   >>>     print('spec = {!r}'.format(spec))
   >>>     #
   >>>     self = ChannelSpec.coerce(spec)
   >>>     print('self = {!r}'.format(self))
   >>>     sizes = self.sizes()
   >>>     print('sizes = {!r}'.format(sizes))
   >>>     print('self.info = {}'.format(ub.repr2(self.info, nl=1)))
   >>>     #
   >>>     item = self._demo_item((1, 1), rng=0)
   >>>     inputs = self.encode(item)
   >>>     components = self.decode(inputs)
   >>>     input_shapes = ub.map_vals(lambda x: x.shape, inputs)
   >>>     component_shapes = ub.map_vals(lambda x: x.shape, components)
   >>>     print('item = {}'.format(ub.repr2(item, precision=1)))
   >>>     print('inputs = {}'.format(ub.repr2(inputs, precision=1)))
   >>>     print('input_shapes = {}'.format(ub.repr2(input_shapes)))
   >>>     print('components = {}'.format(ub.repr2(components, precision=1)))
   >>>     print('component_shapes = {}'.format(ub.repr2(component_shapes, nl=1)))

   .. py:method:: spec(self)
      :property:

      The string encodeing of this spec

      :returns: str


   .. py:method:: __contains__(self, key)

      .. rubric:: Example

      >>> 'disparity' in ChannelSpec('rgb,disparity,flowx|flowy')
      True
      >>> 'gray' in ChannelSpec('rgb,disparity,flowx|flowy')
      False


   .. py:method:: info(self)
      :property:


   .. py:method:: coerce(cls, data)
      :classmethod:

      Attempt to interpret the data as a channel specification

      :returns: ChannelSpec

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> data = FusedChannelSpec.coerce(3)
      >>> assert ChannelSpec.coerce(data).spec == 'u0|u1|u2'
      >>> data = ChannelSpec.coerce(3)
      >>> assert data.spec == 'u0|u1|u2'
      >>> assert ChannelSpec.coerce(data).spec == 'u0|u1|u2'
      >>> data = ChannelSpec.coerce('u:3')
      >>> assert data.normalize().spec == 'u.0|u.1|u.2'


   .. py:method:: parse(self)

      Build internal representation

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> self = ChannelSpec('b1|b2|b3|rgb,B:3')
      >>> print(self.parse())
      >>> print(self.normalize().parse())
      >>> ChannelSpec('').parse()

      .. rubric:: Example

      >>> base = ChannelSpec('rgb|disparity,flowx|r|flowy')
      >>> other = ChannelSpec('rgb')
      >>> self = base.intersection(other)
      >>> assert self.numel() == 4


   .. py:method:: concise(self)

      .. rubric:: Example

      >>> self = ChannelSpec('b1|b2,b3|rgb|B.0,B.1|B.2')
      >>> print(self.concise().spec)
      b1|b2,b3|r|g|b|B.0,B.1:3


   .. py:method:: normalize(self)

      Replace aliases with explicit single-band-per-code specs

      :returns: normalized spec
      :rtype: ChannelSpec

      .. rubric:: Example

      >>> self = ChannelSpec('b1|b2,b3|rgb,B:3')
      >>> normed = self.normalize()
      >>> print('self   = {}'.format(self))
      >>> print('normed = {}'.format(normed))
      self   = <ChannelSpec(b1|b2,b3|rgb,B:3)>
      normed = <ChannelSpec(b1|b2,b3|r|g|b,B.0|B.1|B.2)>


   .. py:method:: keys(self)


   .. py:method:: values(self)


   .. py:method:: items(self)


   .. py:method:: fuse(self)

      Fuse all parts into an early fused channel spec

      :returns: FusedChannelSpec

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> self = ChannelSpec.coerce('b1|b2,b3|rgb,B:3')
      >>> fused = self.fuse()
      >>> print('self  = {}'.format(self))
      >>> print('fused = {}'.format(fused))
      self  = <ChannelSpec(b1|b2,b3|rgb,B:3)>
      fused = <FusedChannelSpec(b1|b2|b3|rgb|B:3)>


   .. py:method:: streams(self)

      Breaks this spec up into one spec for each early-fused input stream

      .. rubric:: Example

      self = ChannelSpec.coerce('r|g,B1|B2,fx|fy')
      list(map(len, self.streams()))


   .. py:method:: code_list(self)


   .. py:method:: difference(self, other)

      Set difference. Remove all instances of other channels from
      this set of channels.

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *
      >>> self = ChannelSpec('rgb|disparity,flowx|r|flowy')
      >>> other = ChannelSpec('rgb')
      >>> print(self.difference(other))
      >>> other = ChannelSpec('flowx')
      >>> print(self.difference(other))
      <ChannelSpec(disparity,flowx|flowy)>
      <ChannelSpec(r|g|b|disparity,r|flowy)>

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *
      >>> self = ChannelSpec('a|b,c|d')
      >>> new = self - {'a', 'b'}
      >>> len(new.sizes()) == 1
      >>> empty = new - 'c|d'
      >>> assert empty.numel() == 0


   .. py:method:: intersection(self, other)

      Set difference. Remove all instances of other channels from
      this set of channels.

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *
      >>> self = ChannelSpec('rgb|disparity,flowx|r|flowy')
      >>> other = ChannelSpec('rgb')
      >>> new = self.intersection(other)
      >>> print(new)
      >>> print(new.numel())
      >>> other = ChannelSpec('flowx')
      >>> new = self.intersection(other)
      >>> print(new)
      >>> print(new.numel())
      <ChannelSpec(r|g|b,r)>
      4
      <ChannelSpec(flowx)>
      1


   .. py:method:: numel(self)

      Total number of channels in this spec


   .. py:method:: sizes(self)

      Number of dimensions for each fused stream channel

      IE: The EARLY-FUSED channel sizes

      .. rubric:: Example

      >>> self = ChannelSpec('rgb|disparity,flowx|flowy,B:10')
      >>> self.normalize().concise()
      >>> self.sizes()


   .. py:method:: unique(self, normalize=False)

      Returns the unique channels that will need to be given or loaded


   .. py:method:: _item_shapes(self, dims)

      Expected shape for an input item

      :Parameters: **dims** (*Tuple[int, int]*) -- the spatial dimension

      :returns: Dict[int, tuple]


   .. py:method:: _demo_item(self, dims=(4, 4), rng=None)

      Create an input that satisfies this spec

      :returns:

                an item like it might appear when its returned from the
                    `__getitem__` method of a :class:`torch...Dataset`.
      :rtype: dict

      .. rubric:: Example

      >>> dims = (1, 1)
      >>> ChannelSpec.coerce(3)._demo_item(dims, rng=0)
      >>> ChannelSpec.coerce('r|g|b|disaprity')._demo_item(dims, rng=0)
      >>> ChannelSpec.coerce('rgb|disaprity')._demo_item(dims, rng=0)
      >>> ChannelSpec.coerce('rgb,disaprity')._demo_item(dims, rng=0)
      >>> ChannelSpec.coerce('rgb')._demo_item(dims, rng=0)
      >>> ChannelSpec.coerce('gray')._demo_item(dims, rng=0)


   .. py:method:: encode(self, item, axis=0, mode=1)

      Given a dictionary containing preloaded components of the network
      inputs, build a concatenated (fused) network representations of each
      input stream.

      :Parameters: * **item** (*Dict[str, Tensor]*) -- a batch item containing unfused parts.
                     each key should be a single-stream (optionally early fused)
                     channel key.
                   * **axis** (*int, default=0*) -- concatenation dimension

      :returns:     mapping between input stream and its early fused tensor input.
      :rtype: Dict[str, Tensor]

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> import numpy as np
      >>> dims = (4, 4)
      >>> item = {
      >>>     'rgb': np.random.rand(3, *dims),
      >>>     'disparity': np.random.rand(1, *dims),
      >>>     'flowx': np.random.rand(1, *dims),
      >>>     'flowy': np.random.rand(1, *dims),
      >>> }
      >>> # Complex Case
      >>> self = ChannelSpec('rgb,disparity,rgb|disparity|flowx|flowy,flowx|flowy')
      >>> fused = self.encode(item)
      >>> input_shapes = ub.map_vals(lambda x: x.shape, fused)
      >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1)))
      >>> # Simpler case
      >>> self = ChannelSpec('rgb|disparity')
      >>> fused = self.encode(item)
      >>> input_shapes = ub.map_vals(lambda x: x.shape, fused)
      >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1)))

      .. rubric:: Example

      >>> # Case where we have to break up early fused data
      >>> import numpy as np
      >>> dims = (40, 40)
      >>> item = {
      >>>     'rgb|disparity': np.random.rand(4, *dims),
      >>>     'flowx': np.random.rand(1, *dims),
      >>>     'flowy': np.random.rand(1, *dims),
      >>> }
      >>> # Complex Case
      >>> self = ChannelSpec('rgb,disparity,rgb|disparity,rgb|disparity|flowx|flowy,flowx|flowy,flowx,disparity')
      >>> inputs = self.encode(item)
      >>> input_shapes = ub.map_vals(lambda x: x.shape, inputs)
      >>> print('input_shapes = {}'.format(ub.repr2(input_shapes, nl=1)))

      >>> # xdoctest: +REQUIRES(--bench)
      >>> #self = ChannelSpec('rgb|disparity,flowx|flowy')
      >>> import timerit
      >>> ti = timerit.Timerit(100, bestof=10, verbose=2)
      >>> for timer in ti.reset('mode=simple'):
      >>>     with timer:
      >>>         inputs = self.encode(item, mode=0)
      >>> for timer in ti.reset('mode=minimize-concat'):
      >>>     with timer:
      >>>         inputs = self.encode(item, mode=1)


   .. py:method:: decode(self, inputs, axis=1)

      break an early fused item into its components

      :Parameters: * **inputs** (*Dict[str, Tensor]*) -- dictionary of components
                   * **axis** (*int, default=1*) -- channel dimension

      .. rubric:: Example

      >>> from kwcoco.channel_spec import *  # NOQA
      >>> import numpy as np
      >>> dims = (4, 4)
      >>> item_components = {
      >>>     'rgb': np.random.rand(3, *dims),
      >>>     'ir': np.random.rand(1, *dims),
      >>> }
      >>> self = ChannelSpec('rgb|ir')
      >>> item_encoded = self.encode(item_components)
      >>> batch = {k: np.concatenate([v[None, :], v[None, :]], axis=0)
      ...          for k, v in item_encoded.items()}
      >>> components = self.decode(batch)

      .. rubric:: Example

      >>> # xdoctest: +REQUIRES(module:netharn, module:torch)
      >>> import torch
      >>> import numpy as np
      >>> dims = (4, 4)
      >>> components = {
      >>>     'rgb': np.random.rand(3, *dims),
      >>>     'ir': np.random.rand(1, *dims),
      >>> }
      >>> components = ub.map_vals(torch.from_numpy, components)
      >>> self = ChannelSpec('rgb|ir')
      >>> encoded = self.encode(components)
      >>> from netharn.data import data_containers
      >>> item = {k: data_containers.ItemContainer(v, stack=True)
      >>>         for k, v in encoded.items()}
      >>> batch = data_containers.container_collate([item, item])
      >>> components = self.decode(batch)


   .. py:method:: component_indices(self, axis=2)

      Look up component indices within fused streams

      .. rubric:: Example

      >>> dims = (4, 4)
      >>> inputs = ['flowx', 'flowy', 'disparity']
      >>> self = ChannelSpec('disparity,flowx|flowy')
      >>> component_indices = self.component_indices()
      >>> print('component_indices = {}'.format(ub.repr2(component_indices, nl=1)))
      component_indices = {
          'disparity': ('disparity', (slice(None, None, None), slice(None, None, None), slice(0, 1, None))),
          'flowx': ('flowx|flowy', (slice(None, None, None), slice(None, None, None), slice(0, 1, None))),
          'flowy': ('flowx|flowy', (slice(None, None, None), slice(None, None, None), slice(1, 2, None))),
      }



.. py:function:: _cached_single_fused_mapping(item_keys, parsed_items, axis=0)


.. py:function:: _cached_single_stream_idxs(key, axis=0)


.. py:function:: subsequence_index(oset1, oset2)

   Returns a slice into the first items indicating the position of
   the second items if they exist.

   This is a variant of the substring problem.

   :returns: None | slice

   .. rubric:: Example

   >>> oset1 = ub.oset([1, 2, 3, 4, 5, 6])
   >>> oset2 = ub.oset([2, 3, 4])
   >>> index = subsequence_index(oset1, oset2)
   >>> assert index

   >>> oset1 = ub.oset([1, 2, 3, 4, 5, 6])
   >>> oset2 = ub.oset([2, 4, 3])
   >>> index = subsequence_index(oset1, oset2)
   >>> assert not index


.. py:function:: _parse_concise_slice_syntax(v)

   Helper for our slice syntax, which is may be a bit strange

   .. rubric:: Example

   >>> print(_parse_concise_slice_syntax('B:10'))
   >>> print(_parse_concise_slice_syntax('B.0:10:3'))
   >>> print(_parse_concise_slice_syntax('B.:10:3'))
   >>> print(_parse_concise_slice_syntax('B::10:3'))
   >>> # Careful, this next one is quite different
   >>> print(_parse_concise_slice_syntax('B:10:3'))
   >>> print(_parse_concise_slice_syntax('B:3:10:3'))
   >>> print(_parse_concise_slice_syntax('B.:10'))
   >>> print(_parse_concise_slice_syntax('B.:3:'))
   >>> print(_parse_concise_slice_syntax('B.:3:2'))
   >>> print(_parse_concise_slice_syntax('B::2:3'))
   >>> print(_parse_concise_slice_syntax('B.0:10:3'))
   >>> print(_parse_concise_slice_syntax('B.:10:3'))
   ('B', 0, 10, 1)
   ('B', 0, 10, 3)
   ('B', 0, 10, 3)
   ('B', 0, 10, 3)
   ('B', 10, 3, 1)
   ('B', 3, 10, 3)
   ('B', 0, 10, 1)
   ('B', 0, 3, 1)
   ('B', 0, 3, 2)
   ('B', 0, 2, 3)
   ('B', 0, 10, 3)
   ('B', 0, 10, 3)
   >>> import pytest
   >>> with pytest.raises(ValueError):
   >>>     _parse_concise_slice_syntax('B.0')
   >>> with pytest.raises(ValueError):
   >>>     _parse_concise_slice_syntax('B0')
   >>> with pytest.raises(ValueError):
   >>>     _parse_concise_slice_syntax('B:')
   >>> with pytest.raises(ValueError):
   >>>     _parse_concise_slice_syntax('B:0.10')
   >>> with pytest.raises(ValueError):
   >>>     _parse_concise_slice_syntax('B.::')


.. py:function:: oset_insert(self, index, obj)


.. py:function:: oset_delitem(self, index)

   for ubelt oset, todo contribute back to luminosoinsight

   >>> self = ub.oset([1, 2, 3, 4, 5, 6, 7, 8, 9])
   >>> index = slice(3, 5)
   >>> oset_delitem(self, index)