kwcoco.channel_spec

This module defines the KWCOCO Channel Specification and API.

The KWCOCO Channel specification is a way to semantically express how a combination of image channels are grouped. This can specify how these channels (somtimes called bands or features) are arranged on disk or input to an algorithm. The core idea reduces to a Set[List[str]] — or a unordered set of ordered sequences of strings corresponding to channel “names”. The way these are specified is with a “,” to separate lists in an unordered set and with a “|” to separate the channel names. Other syntax exists for convinience, but a strict normalized channel spec only contains these core symbols.

Another way to think of a kwcoco channel spec is that splitting the spec by “,” gives groups of channels that should be processed together and “late-fused”. Within each group the “|” operator “early-fuses” the channels.

For instance, say we had a network and we wanted to process 3-channel rgb images in one stream and 1-channel infrared images in a second stream and then fuse them together. The kwcoco channel specification for channels labled as ‘red’, ‘green’, ‘blue’, and ‘infrared’ would be:

infrared,red|green|blue

Note, it is up to an algorithm to do any early-late fusion. KWCoco simply provides the specification as a tool to quickly access a particular combination of channels from disk.

The ChannelSpec has these simple rules:

* 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

The detailed grammar for the spec is:

?start: stream

// An identifier can contain spaces IDEN: (“_”|LETTER) ("_"|” “|LETTER|DIGIT)*

chan_single : IDEN chan_getitem : IDEN “.” INT chan_getslice_0b : IDEN “:” INT chan_getslice_ab : IDEN “.” INT “:” INT

// A channel code can just be an ID, or it can have a getitem // style syntax with a scalar or slice as an argument chan_code : chan_single | chan_getslice_0b | chan_getslice_ab | chan_getitem

// Fused channels are an ordered sequence of channel codes (without sensors) fused : chan_code (“|” chan_code)*

// Channels can be specified in a sequence but must contain parens fused_seq : “(” fused (“,” fused)* “)”

channel_rhs : fused | fused_seq

stream : channel_rhs (“,” channel_rhs)*

%import common.DIGIT %import common.LETTER %import common.INT

Note that a stream refers to a the full ChannelSpec and fused refers to FusedChannelSpec.

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

Todo

Let S1 = sensor 1 Let S2 = sensor 2 Let S3 = sensor 3

Using a <sensor>:<pure_fused_channel_spec> indicates that the fused channels belong to that sensor.

For example:

S1:red|green|blue S1:B.0:3 = S1:C.0|C.1|C.2

To specify that R|G|B channels exist in sensor 1 and sensor 2 you could do:

S1:R|G|B,S2:R|G|B

or the more concise syntax allows for a distributive law

(S1,S2):R|G|B

Notice, how the “R|G|B” channel code is distributed over the “,” in the parenthesis.

Note

• do not specify the same channel in FusedChannelSpec twice

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

BaseChannelSpec	Common code API between FusedChannelSpec and ChannelSpec
FusedChannelSpec	A specific type of channel spec with only one early fused stream.
ChannelSpec	Parse and extract information about network input channel specs for

Functions

_cached_single_fused_mapping(item_keys, parsed_items, axis=0)
_cached_single_stream_idxs(key, axis=0)
subsequence_index(oset1, oset2)	Returns a slice into the first items indicating the position of
_parse_concise_slice_syntax(v)	Helper for our slice syntax, which is may be a bit strange
oset_insert(self, index, obj)
oset_delitem(self, index)	for ubelt oset, todo contribute back to luminosoinsight

class kwcoco.channel_spec.BaseChannelSpec

Bases: ubelt.NiceRepr

Common code API between FusedChannelSpec and ChannelSpec

Todo

• [ ] Keep working on this base spec and ensure the inheriting classes

  conform to it.

property spec(self)

The string encodeing of this spec

Returns

str

abstract classmethod coerce(cls, data)

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

abstract streams(self)

Breakup this spec into individual early-fused components

Returns

List[FusedChannelSpec]

abstract normalize(self)

Expand all channel codes into their normalized long-form

Returns

BaseChannelSpec

abstract intersection(self, other)

abstract union(self, other)

abstract difference(self)

abstract issubset(self, other)

abstract issuperset(self, other)

__sub__(self, other)

__nice__(self)

__json__(self)

__and__(self, other)

__or__(self, other)

path_sanitize(self, maxlen=None)

Clean up the channel spec so it can be used in a pathname.

Parameters

maxlen (int) – if specified, and the name is longer than this length, it is shortened. Must be 8 or greater.

Returns

path suitable for usage in a filename

Return type

str

Note

This mapping is not invertible and should not be relied on to reconstruct the path spec. This is only a convenience.

Example

>>> import kwcoco
>>> print(kwcoco.FusedChannelSpec.coerce('a chan with space|bar|baz').path_sanitize())
a chan with space_bar_baz
>>> print(kwcoco.ChannelSpec.coerce('foo|bar|baz,biz').path_sanitize())
foo_bar_baz,biz

Example

>>> import kwcoco
>>> print(kwcoco.ChannelSpec.coerce('foo.0:3').normalize().path_sanitize(24))
foo.0_foo.1_foo.2
>>> print(kwcoco.ChannelSpec.coerce('foo.0:256').normalize().path_sanitize(24))
tuuxtfnrsvdhezkdndysxo_256

class kwcoco.channel_spec.FusedChannelSpec(parsed, _is_normalized=False)

Bases: 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?

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))

_alias_lut

_memo

_size_lut

__len__(self)

__getitem__(self, index)

classmethod concat(cls, items)

spec(self)

The string encodeing of this spec

Returns

str

unique(self)

classmethod parse(cls, spec)

classmethod coerce(cls, data)

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

concise(self)

Shorted the channel spec by de-normaliz slice syntax

Returns

concise spec

Return type

FusedChannelSpec

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

normalize(self)

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

Returns

normalize spec

Return type

FusedChannelSpec

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)>

numel(self)

Total number of channels in this spec

sizes(self)

Returns a list indicating the size of each atomic code

Returns

List[int]

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

__contains__(self, key)

Example

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

code_list(self)

Return the expanded code list

as_list(self)

as_oset(self)

as_set(self)

as_path(self)

Returns a string suitable for use in a path.

Note, this may no longer be a valid channel spec

__set__(self)

difference(self, other)

Set difference

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

intersection(self, other)

Example

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

union(self, other)

Example

>>> from kwcoco.channel_spec import *  # NOQA
>>> FCS = FusedChannelSpec.coerce
>>> self = FCS('rgb|disparity|flowx|flowy')
>>> other = FCS('r|b|XX')
>>> self.union(other)

issubset(self, other)

issuperset(self, other)

component_indices(self, axis=2)

Look up component indices within this stream

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)),
}

streams(self)

Idempotence with ChannelSpec.streams()

fuse(self)

Idempotence with ChannelSpec.streams()

class kwcoco.channel_spec.ChannelSpec(spec, parsed=None)

Bases: 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

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) ...>

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)))

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)))

property spec(self)

The string encodeing of this spec

Returns

str

__contains__(self, key)

Example

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

property info(self)

classmethod coerce(cls, data)

Attempt to interpret the data as a channel specification

Returns

ChannelSpec

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'

parse(self)

Build internal representation

Example

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

Example

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

concise(self)

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

normalize(self)

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

Returns

normalized spec

Return type

ChannelSpec

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)>

keys(self)

values(self)

items(self)

fuse(self)

Fuse all parts into an early fused channel spec

Returns

FusedChannelSpec

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)>

streams(self)

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

Example

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

code_list(self)

as_path(self)

Returns a string suitable for use in a path.

Note, this may no longer be a valid channel spec

difference(self, other)

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

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)>

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

intersection(self, other)

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

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

union(self, other)

Union simply tags on a second channel spec onto this one. Duplicates are maintained.

Example

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

abstract issubset(self, other)

abstract issuperset(self, other)

numel(self)

Total number of channels in this spec

sizes(self)

Number of dimensions for each fused stream channel

IE: The EARLY-FUSED channel sizes

Example

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

unique(self, normalize=False)

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

_item_shapes(self, dims)

Expected shape for an input item

Parameters

dims (Tuple[int, int]) – the spatial dimension

Returns

Dict[int, tuple]

_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 torch...Dataset.

Return type

dict

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)

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.

Return type

Dict[str, Tensor]

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)))

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)

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

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)

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)

component_indices(self, axis=2)

Look up component indices within fused streams

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))),
}

kwcoco.channel_spec._cached_single_fused_mapping(item_keys, parsed_items, axis=0)

kwcoco.channel_spec._cached_single_stream_idxs(key, axis=0)

kwcoco.channel_spec.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

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

kwcoco.channel_spec._parse_concise_slice_syntax(v)

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

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.::')

kwcoco.channel_spec.oset_insert(self, index, obj)

kwcoco.channel_spec.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)