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¶
Common code API between |
|
A specific type of channel spec with only one early fused stream. |
|
Parse and extract information about network input channel specs for |
Functions¶
|
|
|
|
|
Returns a slice into the first items indicating the position of |
Helper for our slice syntax, which is may be a bit strange |
|
|
|
|
for ubelt oset, todo contribute back to luminosoinsight |
- class kwcoco.channel_spec.BaseChannelSpec[source]¶
Bases:
ubelt.NiceRepr
Common code API between
FusedChannelSpec
andChannelSpec
Todo
- [ ] Keep working on this base spec and ensure the inheriting classes
conform to it.
- abstract classmethod coerce(cls, data)[source]¶
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)[source]¶
Breakup this spec into individual early-fused components
- Returns
List[FusedChannelSpec]
- abstract normalize(self)[source]¶
Expand all channel codes into their normalized long-form
- Returns
BaseChannelSpec
- path_sanitize(self, maxlen=None)[source]¶
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
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)[source]¶
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))
- classmethod coerce(cls, data)[source]¶
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)[source]¶
Shorted the channel spec by de-normaliz slice syntax
- Returns
concise spec
- Return type
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)[source]¶
Replace aliases with explicit single-band-per-code specs
- Returns
normalize spec
- Return type
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)>
- sizes(self)[source]¶
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)[source]¶
Example
>>> FCS = FusedChannelSpec.coerce >>> 'disparity' in FCS('rgb|disparity|flowx|flowy') True >>> 'gray' in FCS('rgb|disparity|flowx|flowy') False
- as_path(self)[source]¶
Returns a string suitable for use in a path.
Note, this may no longer be a valid channel spec
- difference(self, other)[source]¶
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)[source]¶
Example
>>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b|XX') >>> self.intersection(other)
- union(self, other)[source]¶
Example
>>> from kwcoco.channel_spec import * # NOQA >>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b|XX') >>> self.union(other)
- component_indices(self, axis=2)[source]¶
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)[source]¶
Idempotence with
ChannelSpec.streams()
- fuse(self)[source]¶
Idempotence with
ChannelSpec.streams()
- class kwcoco.channel_spec.ChannelSpec(spec, parsed=None)[source]¶
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)))
- __contains__(self, key)[source]¶
Example
>>> 'disparity' in ChannelSpec('rgb,disparity,flowx|flowy') True >>> 'gray' in ChannelSpec('rgb,disparity,flowx|flowy') False
- classmethod coerce(cls, data)[source]¶
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)[source]¶
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)[source]¶
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)[source]¶
Replace aliases with explicit single-band-per-code specs
- Returns
normalized spec
- Return type
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)>
- fuse(self)[source]¶
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)[source]¶
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()))
- as_path(self)[source]¶
Returns a string suitable for use in a path.
Note, this may no longer be a valid channel spec
- difference(self, other)[source]¶
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)[source]¶
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)[source]¶
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
- sizes(self)[source]¶
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)[source]¶
Returns the unique channels that will need to be given or loaded
- _item_shapes(self, dims)[source]¶
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)[source]¶
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
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)[source]¶
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)[source]¶
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)[source]¶
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.subsequence_index(oset1, oset2)[source]¶
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)[source]¶
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.::')