:py:mod:`kwcoco.channel_spec` ============================= .. py:module:: kwcoco.channel_spec .. autoapi-nested-parse:: 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: .. 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 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:: Sensor Codes ------------ Let S1 = sensor 1 Let S2 = sensor 2 Let S3 = sensor 3 Using a : 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 .. rubric:: Example >>> import kwcoco >>> spec = kwcoco.ChannelSpec('b1|b2|b3,m.0:4|x1|x2,x.3|x.4|x.5') >>> print(spec) >>> for stream in spec.streams(): >>> print(stream) >>> # Normalization >>> normalized = spec.normalize() >>> print(normalized) >>> 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, other) :abstractmethod: .. py:method:: union(self, other) :abstractmethod: .. py:method:: difference(self) :abstractmethod: .. py:method:: issubset(self, other) :abstractmethod: .. py:method:: issuperset(self, other) :abstractmethod: .. py:method:: __sub__(self, other) .. py:method:: __nice__(self) .. py:method:: __json__(self) .. py:method:: __and__(self, other) .. py:method:: __or__(self, other) .. py:method:: 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 :rtype: str .. note:: This mapping is not invertible and should not be relied on to reconstruct the path spec. This is only a convenience. .. rubric:: 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 .. rubric:: 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 .. 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 = normed = >>> self = FusedChannelSpec.coerce('B:1:11') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = >>> self = FusedChannelSpec.coerce('B.1:11') >>> normed = self.normalize() >>> print('self = {}'.format(self)) >>> print('normed = {}'.format(normed)) self = normed = .. 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:: as_path(self) Returns a string suitable for use in a path. Note, this may no longer be a valid channel spec .. 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:: union(self, other) .. rubric:: Example >>> from kwcoco.channel_spec import * # NOQA >>> FCS = FusedChannelSpec.coerce >>> self = FCS('rgb|disparity|flowx|flowy') >>> other = FCS('r|b|XX') >>> self.union(other) .. py:method:: issubset(self, other) .. py:method:: issuperset(self, 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) >>> # single mode spec >>> ChannelSpec.coerce('rgb') >>> # early fused input spec >>> ChannelSpec.coerce('rgb|disprity') >>> # late fused input spec >>> ChannelSpec.coerce('rgb,disprity') >>> # early and late fused input spec >>> ChannelSpec.coerce('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 = normed = .. 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 = fused = .. 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:: as_path(self) Returns a string suitable for use in a path. Note, this may no longer be a valid channel spec .. 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)) .. 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()) 4 1 .. py:method:: union(self, other) Union simply tags on a second channel spec onto this one. Duplicates are maintained. .. rubric:: 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()) 10 8 .. py:method:: issubset(self, other) :abstractmethod: .. py:method:: issuperset(self, other) :abstractmethod: .. 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)