kwcoco.util.delayed_ops.delayed_nodes module¶
Intermediate operations
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedStack(parts, axis)[source]¶
Bases:
DelayedNaryOperation
Stacks multiple arrays together.
- property shape¶
Returns: None | Tuple[int | None, …]
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedConcat(parts, axis)[source]¶
Bases:
DelayedNaryOperation
Stacks multiple arrays together.
- property shape¶
Returns: None | Tuple[int | None, …]
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedFrameStack(parts)[source]¶
Bases:
DelayedStack
Stacks multiple arrays together.
- class kwcoco.util.delayed_ops.delayed_nodes.ImageOpsMixin[source]¶
Bases:
object
- crop(space_slice=None, chan_idxs=None, clip=True, wrap=True, pad=0)[source]¶
Crops an image along integer pixel coordinates.
- Parameters
space_slice (Tuple[slice, slice]) – y-slice and x-slice.
chan_idxs (List[int]) – indexes of bands to take
clip (bool) – if True, the slice is interpreted normally, where it won’t go past the image extent, otherwise slicing into negative regions or past the image bounds will result in padding. Defaults to True.
wrap (bool) – if True, negative indexes “wrap around”, otherwise they are treated as is. Defaults to True.
pad (int | List[Tuple[int, int]]) – if specified, applies extra padding
- Returns
DelayedImage
Example
>>> from kwcoco.util.delayed_ops import DelayedLoad >>> import kwimage >>> self = DelayedLoad.demo().prepare() >>> self = self.dequantize({'quant_max': 255}) >>> self = self.warp({'scale': 1 / 2}) >>> pad = 0 >>> h, w = space_dims = self.dsize[::-1] >>> grid = list(ub.named_product({ >>> 'left': [0, -64], 'right': [0, 64], >>> 'top': [0, -64], 'bot': [0, 64],})) >>> grid += [ >>> {'left': 64, 'right': -64, 'top': 0, 'bot': 0}, >>> {'left': 64, 'right': 64, 'top': 0, 'bot': 0}, >>> {'left': 0, 'right': 0, 'top': 64, 'bot': -64}, >>> {'left': 64, 'right': -64, 'top': 64, 'bot': -64}, >>> ] >>> crops = [] >>> for pads in grid: >>> space_slice = (slice(pads['top'], h + pads['bot']), >>> slice(pads['left'], w + pads['right'])) >>> delayed = self.crop(space_slice) >>> crop = delayed.finalize() >>> yyxx = kwimage.Boxes.from_slice(space_slice, wrap=False, clip=0).toformat('_yyxx').data[0] >>> title = '[{}:{}, {}:{}]'.format(*yyxx) >>> crop_canvas = kwimage.draw_header_text(crop, title, fit=True, bg_color='kw_darkgray') >>> crops.append(crop_canvas) >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> canvas = kwimage.stack_images_grid(crops, pad=16, bg_value='kw_darkgreen') >>> canvas = kwimage.fill_nans_with_checkers(canvas) >>> kwplot.imshow(canvas, title='Normal Slicing: Cropped Images With Wrap+Clipped Slices', doclf=1, fnum=1) >>> kwplot.show_if_requested()
Example
>>> # Demo the case with pads / no-clips / no-wraps >>> from kwcoco.util.delayed_ops import DelayedLoad >>> import kwimage >>> self = DelayedLoad.demo().prepare() >>> self = self.dequantize({'quant_max': 255}) >>> self = self.warp({'scale': 1 / 2}) >>> pad = [(64, 128), (32, 96)] >>> pad = [(0, 20), (0, 0)] >>> pad = 0 >>> pad = 8 >>> h, w = space_dims = self.dsize[::-1] >>> grid = list(ub.named_product({ >>> 'left': [0, -64], 'right': [0, 64], >>> 'top': [0, -64], 'bot': [0, 64],})) >>> grid += [ >>> {'left': 64, 'right': -64, 'top': 0, 'bot': 0}, >>> {'left': 64, 'right': 64, 'top': 0, 'bot': 0}, >>> {'left': 0, 'right': 0, 'top': 64, 'bot': -64}, >>> {'left': 64, 'right': -64, 'top': 64, 'bot': -64}, >>> ] >>> crops = [] >>> for pads in grid: >>> space_slice = (slice(pads['top'], h + pads['bot']), >>> slice(pads['left'], w + pads['right'])) >>> delayed = self._padded_crop(space_slice, pad=pad) >>> crop = delayed.finalize(optimize=1) >>> yyxx = kwimage.Boxes.from_slice(space_slice, wrap=False, clip=0).toformat('_yyxx').data[0] >>> title = '[{}:{}, {}:{}]'.format(*yyxx) >>> if pad: >>> title += f'{chr(10)}pad={pad}' >>> crop_canvas = kwimage.draw_header_text(crop, title, fit=True, bg_color='kw_darkgray') >>> crops.append(crop_canvas) >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> canvas = kwimage.stack_images_grid(crops, pad=16, bg_value='kw_darkgreen', resize='smaller') >>> canvas = kwimage.fill_nans_with_checkers(canvas) >>> kwplot.imshow(canvas, title='Negative Slicing: Cropped Images With clip=False wrap=False', doclf=1, fnum=2) >>> kwplot.show_if_requested()
- warp(transform, dsize='auto', antialias=True, interpolation='linear', border_value='auto')[source]¶
Applys an affine transformation to the image
- Parameters
transform (ndarray | dict | kwimage.Affine) – a coercable affine matrix. See
kwimage.Affine
for details on what can be coerced.dsize (Tuple[int, int] | str) – The width / height of the output canvas. If ‘auto’, dsize is computed such that the positive coordinates of the warped image will fit in the new canvas. In this case, any pixel that maps to a negative coordinate will be clipped. This has the property that the input transformation is not modified.
antialias (bool) – if True determines if the transform is downsampling and applies antialiasing via gaussian a blur. Defaults to False
interpolation (str) – interpolation code or cv2 integer. Interpolation codes are linear, nearest, cubic, lancsoz, and area. Defaults to “linear”.
border_value (int | float | str) – if auto will be nan for float and 0 for int.
- Returns
DelayedImage
- scale(scale, dsize='auto', antialias=True, interpolation='linear', border_value='auto')[source]¶
An alias for self.warp({“scale”: scale}, …)
- dequantize(quantization)[source]¶
Rescales image intensities from int to floats.
- Parameters
quantization (Dict[str, Any]) – see
kwcoco.util.delayed_ops.helpers.dequantize()
- Returns
DelayedDequantize
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedChannelConcat(parts, dsize=None)[source]¶
Bases:
ImageOpsMixin
,DelayedConcat
Stacks multiple arrays together.
CommandLine
xdoctest -m /home/joncrall/code/kwcoco/kwcoco/util/delayed_ops/delayed_nodes.py DelayedChannelConcat:1
Example
>>> from kwcoco.util.delayed_ops import * # NOQA >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedLoad >>> import kwcoco >>> dsize = (307, 311) >>> c1 = DelayedNans(dsize=dsize, channels='foo') >>> c2 = DelayedLoad.demo('astro', dsize=dsize, channels='R|G|B').prepare() >>> cat = DelayedChannelConcat([c1, c2]) >>> warped_cat = cat.warp({'scale': 1.07}, dsize=(328, 332)) >>> warped_cat._validate() >>> warped_cat.finalize()
Example
>>> # Test case that failed in initial implementation >>> # Due to incorrectly pushing channel selection under the concat >>> from kwcoco.util.delayed_ops import * # NOQA >>> import kwimage >>> fpath = kwimage.grab_test_image_fpath() >>> base1 = DelayedLoad(fpath, channels='r|g|b').prepare() >>> base2 = DelayedLoad(fpath, channels='x|y|z').prepare().scale(2) >>> base3 = DelayedLoad(fpath, channels='i|j|k').prepare().scale(2) >>> bands = [base2, base1[:, :, 0].scale(2).evaluate(), >>> base1[:, :, 1].evaluate().scale(2), >>> base1[:, :, 2].evaluate().scale(2), base3] >>> delayed = DelayedChannelConcat(bands) >>> delayed = delayed.warp({'scale': 2}) >>> delayed = delayed[0:100, 0:55, [0, 2, 4]] >>> delayed.write_network_text() >>> delayed.optimize()
- property channels¶
Returns: None | kwcoco.FusedChannelSpec
- property shape¶
Returns: Tuple[int | None, int | None, int | None]
- take_channels(channels)[source]¶
This method returns a subset of the vision data with only the specified bands / channels.
- Parameters
channels (List[int] | slice | channel_spec.FusedChannelSpec) – List of integers indexes, a slice, or a channel spec, which is typically a pipe (|) delimited list of channel codes. See
kwcoco.ChannelSpec
for more detials.- Returns
a delayed vision operation that only operates on the following channels.
- Return type
Example
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> import kwcoco >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> self = delayed = dset.coco_image(1).delay(mode=1) >>> channels = 'B11|B8|B1|B10' >>> new = self.take_channels(channels)
Example
>>> # Complex case >>> import kwcoco >>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedLoad >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral') >>> delayed = dset.coco_image(1).delay(mode=1) >>> astro = DelayedLoad.demo('astro', channels='r|g|b').prepare() >>> aligned = astro.warp(kwimage.Affine.scale(600 / 512), dsize='auto') >>> self = combo = DelayedChannelConcat(delayed.parts + [aligned]) >>> channels = 'B1|r|B8|g' >>> new = self.take_channels(channels) >>> new_cropped = new.crop((slice(10, 200), slice(12, 350))) >>> new_opt = new_cropped.optimize() >>> datas = new_opt.finalize() >>> if 1: >>> new_cropped.write_network_text(with_labels='name') >>> new_opt.write_network_text(with_labels='name') >>> vizable = kwimage.normalize_intensity(datas, axis=2) >>> self._validate() >>> new._validate() >>> new_cropped._validate() >>> new_opt._validate() >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> stacked = kwimage.stack_images(vizable.transpose(2, 0, 1)) >>> kwplot.imshow(stacked)
Example
>>> # Test case where requested channel does not exist >>> import kwcoco >>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> dset = kwcoco.CocoDataset.demo('vidshapes8-multispectral', use_cache=1, verbose=100) >>> self = delayed = dset.coco_image(1).delay(mode=1) >>> channels = 'B1|foobar|bazbiz|B8' >>> new = self.take_channels(channels) >>> new_cropped = new.crop((slice(10, 200), slice(12, 350))) >>> fused = new_cropped.finalize() >>> assert fused.shape == (190, 338, 4) >>> assert np.all(np.isnan(fused[..., 1:3])) >>> assert not np.any(np.isnan(fused[..., 0])) >>> assert not np.any(np.isnan(fused[..., 3]))
- property num_overviews¶
Returns: int
- undo_warps(remove=None, retain=None, squash_nans=False, return_warps=False)[source]¶
Attempts to “undo” warping for each concatenated channel and returns a list of delayed operations that are cropped to the right regions.
Typically you will retrain offset, theta, and shear to remove scale. This ensures the data is spatially aligned up to a scale factor.
- Parameters
remove (List[str]) – if specified, list components of the warping to remove. Can include: “offset”, “scale”, “shearx”, “theta”. Typically set this to [“scale”].
retain (List[str]) – if specified, list components of the warping to retain. Can include: “offset”, “scale”, “shearx”, “theta”. Mutually exclusive with “remove”. If neither remove or retain is specified, retain is set to
[]
.squash_nans (bool) – if True, pure nan channels are squashed into a 1x1 array as they do not correspond to a real source.
return_warps (bool) – if True, return the transforms we applied. This is useful when you need to warp objects in the original space into the jagged space.
Example
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedLoad >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedNans >>> import ubelt as ub >>> import kwimage >>> import kwarray >>> import numpy as np >>> # Demo case where we have different channels at different resolutions >>> base = DelayedLoad.demo(channels='r|g|b').prepare().dequantize({'quant_max': 255}) >>> bandR = base[:, :, 0].scale(100 / 512)[:, :-50].evaluate() >>> bandG = base[:, :, 1].scale(300 / 512).warp({'theta': np.pi / 8, 'about': (150, 150)}).evaluate() >>> bandB = base[:, :, 2].scale(600 / 512)[:150, :].evaluate() >>> bandN = DelayedNans((600, 600), channels='N') >>> # Make a concatenation of images of different underlying native resolutions >>> delayed_vidspace = DelayedChannelConcat([ >>> bandR.scale(6, dsize=(600, 600)).optimize(), >>> bandG.warp({'theta': -np.pi / 8, 'about': (150, 150)}).scale(2, dsize=(600, 600)).optimize(), >>> bandB.scale(1, dsize=(600, 600)).optimize(), >>> bandN, >>> ]).warp({'scale': 0.7}).optimize() >>> vidspace_box = kwimage.Boxes([[100, 10, 270, 160]], 'ltrb') >>> vidspace_poly = vidspace_box.to_polygons()[0] >>> vidspace_slice = vidspace_box.to_slices()[0] >>> self = delayed_vidspace[vidspace_slice].optimize() >>> print('--- Aligned --- ') >>> self.write_network_text() >>> squash_nans = True >>> undone_all_parts, tfs1 = self.undo_warps(squash_nans=squash_nans, return_warps=True) >>> undone_scale_parts, tfs2 = self.undo_warps(remove=['scale'], squash_nans=squash_nans, return_warps=True) >>> stackable_aligned = self.finalize().transpose(2, 0, 1) >>> stackable_undone_all = [] >>> stackable_undone_scale = [] >>> print('--- Undone All --- ') >>> for undone in undone_all_parts: ... undone.write_network_text() ... stackable_undone_all.append(undone.finalize()) >>> print('--- Undone Scale --- ') >>> for undone in undone_scale_parts: ... undone.write_network_text() ... stackable_undone_scale.append(undone.finalize()) >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> canvas0 = kwimage.stack_images(stackable_aligned, axis=1) >>> canvas1 = kwimage.stack_images(stackable_undone_all, axis=1) >>> canvas2 = kwimage.stack_images(stackable_undone_scale, axis=1) >>> canvas0 = kwimage.draw_header_text(canvas0, 'Rescaled Aligned Channels') >>> canvas1 = kwimage.draw_header_text(canvas1, 'Unwarped Channels') >>> canvas2 = kwimage.draw_header_text(canvas2, 'Unscaled Channels') >>> canvas = kwimage.stack_images([canvas0, canvas1, canvas2], axis=0) >>> canvas = kwimage.fill_nans_with_checkers(canvas) >>> kwplot.imshow(canvas)
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedArray(subdata=None)[source]¶
Bases:
DelayedUnaryOperation
A generic NDArray.
- property shape¶
Returns: None | Tuple[int | None, …]
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedImage(subdata=None, dsize=None, channels=None)[source]¶
Bases:
ImageOpsMixin
,DelayedArray
For the case where an array represents a 2D image with multiple channels
- property shape¶
Returns: None | Tuple[int | None, int | None, int | None]
- property num_channels¶
Returns: None | int
- property dsize¶
Returns: None | Tuple[int | None, int | None]
- property channels¶
Returns: None | kwcoco.FusedChannelSpec
- property num_overviews¶
Returns: int
- take_channels(channels)[source]¶
This method returns a subset of the vision data with only the specified bands / channels.
- Parameters
channels (List[int] | slice | channel_spec.FusedChannelSpec) – List of integers indexes, a slice, or a channel spec, which is typically a pipe (|) delimited list of channel codes. See kwcoco.ChannelSpec for more detials.
- Returns
a new delayed load with a fused take channel operation
- Return type
Note
The channel subset must exist here or it will raise an error. A better implementation (via pymbolic) might be able to do better
Example
>>> # >>> # Test Channel Select Via Code >>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> self = DelayedLoad.demo(dsize=(16, 16), channels='r|g|b').prepare() >>> channels = 'r|b' >>> new = self.take_channels(channels)._validate() >>> new2 = new[:, :, [1, 0]]._validate() >>> new3 = new2[:, :, [1]]._validate()
Example
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> import kwcoco >>> self = DelayedLoad.demo('astro').prepare() >>> channels = [2, 0] >>> new = self.take_channels(channels) >>> new3 = new.take_channels([1, 0]) >>> new._validate() >>> new3._validate()
>>> final1 = self.finalize() >>> final2 = new.finalize() >>> final3 = new3.finalize() >>> assert np.all(final1[..., 2] == final2[..., 0]) >>> assert np.all(final1[..., 0] == final2[..., 1]) >>> assert final2.shape[2] == 2
>>> assert np.all(final1[..., 2] == final3[..., 1]) >>> assert np.all(final1[..., 0] == final3[..., 0]) >>> assert final3.shape[2] == 2
- undo_warp(remove=None, retain=None, squash_nans=False, return_warp=False)[source]¶
Attempts to “undo” warping for each concatenated channel and returns a list of delayed operations that are cropped to the right regions.
Typically you will retrain offset, theta, and shear to remove scale. This ensures the data is spatially aligned up to a scale factor.
- Parameters
remove (List[str]) – if specified, list components of the warping to remove. Can include: “offset”, “scale”, “shearx”, “theta”. Typically set this to [“scale”].
retain (List[str]) – if specified, list components of the warping to retain. Can include: “offset”, “scale”, “shearx”, “theta”. Mutually exclusive with “remove”. If neither remove or retain is specified, retain is set to
[]
.squash_nans (bool) – if True, pure nan channels are squashed into a 1x1 array as they do not correspond to a real source.
return_warp (bool) – if True, return the transform we applied. This is useful when you need to warp objects in the original space into the jagged space.
- SeeAlso:
DelayedChannelConcat.undo_warps
Example
>>> # Test similar to undo_warps, but on each channel separately >>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedLoad >>> from kwcoco.util.delayed_ops.delayed_leafs import DelayedNans >>> import ubelt as ub >>> import kwimage >>> import kwarray >>> import numpy as np >>> # Demo case where we have different channels at different resolutions >>> base = DelayedLoad.demo(channels='r|g|b').prepare().dequantize({'quant_max': 255}) >>> bandR = base[:, :, 0].scale(100 / 512)[:, :-50].evaluate() >>> bandG = base[:, :, 1].scale(300 / 512).warp({'theta': np.pi / 8, 'about': (150, 150)}).evaluate() >>> bandB = base[:, :, 2].scale(600 / 512)[:150, :].evaluate() >>> bandN = DelayedNans((600, 600), channels='N') >>> B0 = bandR.scale(6, dsize=(600, 600)).optimize() >>> B1 = bandG.warp({'theta': -np.pi / 8, 'about': (150, 150)}).scale(2, dsize=(600, 600)).optimize() >>> B2 = bandB.scale(1, dsize=(600, 600)).optimize() >>> vidspace_box = kwimage.Boxes([[-10, -10, 270, 160]], 'ltrb').scale(1 / .7).quantize() >>> vidspace_poly = vidspace_box.to_polygons()[0] >>> vidspace_slice = vidspace_box.to_slices()[0] >>> # Test with the padded crop >>> self0 = B0.crop(vidspace_slice, wrap=0, clip=0, pad=10).optimize() >>> self1 = B1.crop(vidspace_slice, wrap=0, clip=0, pad=10).optimize() >>> self2 = B2.crop(vidspace_slice, wrap=0, clip=0, pad=10).optimize() >>> parts = [self0, self1, self2] >>> # Run the undo on each channel >>> undone_scale_parts = [d.undo_warp(remove=['scale']) for d in parts] >>> print('--- Aligned --- ') >>> stackable_aligned = [] >>> for d in parts: >>> d.write_network_text() >>> stackable_aligned.append(d.finalize()) >>> print('--- Undone Scale --- ') >>> stackable_undone_scale = [] >>> for undone in undone_scale_parts: ... undone.write_network_text() ... stackable_undone_scale.append(undone.finalize()) >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> canvas0 = kwimage.stack_images(stackable_aligned, axis=1, pad=5, bg_value='kw_darkgray') >>> canvas2 = kwimage.stack_images(stackable_undone_scale, axis=1, pad=5, bg_value='kw_darkgray') >>> canvas0 = kwimage.draw_header_text(canvas0, 'Rescaled Channels') >>> canvas2 = kwimage.draw_header_text(canvas2, 'Native Scale Channels') >>> canvas = kwimage.stack_images([canvas0, canvas2], axis=0, bg_value='kw_darkgray') >>> canvas = kwimage.fill_nans_with_checkers(canvas) >>> kwplot.imshow(canvas)
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedAsXarray(subdata=None, dsize=None, channels=None)[source]¶
Bases:
DelayedImage
Casts the data to an xarray object in the finalize step
- Example;
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> # without channels >>> base = DelayedLoad.demo(dsize=(16, 16)).prepare() >>> self = base.as_xarray() >>> final = self._validate().finalize() >>> assert len(final.coords) == 0 >>> assert final.dims == ('y', 'x', 'c') >>> # with channels >>> base = DelayedLoad.demo(dsize=(16, 16), channels='r|g|b').prepare() >>> self = base.as_xarray() >>> final = self._validate().finalize() >>> assert final.coords.indexes['c'].tolist() == ['r', 'g', 'b'] >>> assert final.dims == ('y', 'x', 'c')
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedWarp(subdata, transform, dsize='auto', antialias=True, interpolation='linear', border_value='auto')[source]¶
Bases:
DelayedImage
Applies an affine transform to an image.
Example
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> self = DelayedLoad.demo(dsize=(16, 16)).prepare() >>> warp1 = self.warp({'scale': 3}) >>> warp2 = warp1.warp({'theta': 0.1}) >>> warp3 = warp2._opt_fuse_warps() >>> warp3._validate() >>> print(ub.repr2(warp2.nesting(), nl=-1, sort=0)) >>> print(ub.repr2(warp3.nesting(), nl=-1, sort=0))
- property transform¶
Returns: kwimage.Affine
- optimize()[source]¶
- Returns
DelayedImage
Example
>>> # Demo optimization that removes a noop warp >>> from kwcoco.util.delayed_ops import DelayedLoad >>> import kwimage >>> base = DelayedLoad.demo(channels='r|g|b').prepare() >>> self = base.warp(kwimage.Affine.eye()) >>> new = self.optimize() >>> assert len(self.as_graph().nodes) == 2 >>> assert len(new.as_graph().nodes) == 1
Example
>>> # Test optimize nans >>> from kwcoco.util.delayed_ops import DelayedNans >>> import kwimage >>> base = DelayedNans(dsize=(100, 100), channels='a|b|c') >>> self = base.warp(kwimage.Affine.scale(0.1)) >>> # Should simply return a new nan generator >>> new = self.optimize() >>> assert len(new.as_graph().nodes) == 1
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedDequantize(subdata, quantization)[source]¶
Bases:
DelayedImage
Rescales image intensities from int to floats.
The output is usually between 0 and 1. This also handles transforming nodata into nan values.
- optimize()[source]¶
- Returns
DelayedImage
Example
>>> # Test a case that caused an error in development >>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> fpath = kwimage.grab_test_image_fpath() >>> base = DelayedLoad(fpath, channels='r|g|b').prepare() >>> quantization = {'quant_max': 255, 'nodata': 0} >>> self = base.get_overview(1).dequantize(quantization) >>> self.write_network_text() >>> opt = self.optimize()
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedCrop(subdata, space_slice=None, chan_idxs=None)[source]¶
Bases:
DelayedImage
Crops an image along integer pixel coordinates.
Example
>>> from kwcoco.util.delayed_ops.delayed_nodes import * # NOQA >>> from kwcoco.util.delayed_ops import DelayedLoad >>> base = DelayedLoad.demo(dsize=(16, 16)).prepare() >>> # Test Fuse Crops Space Only >>> crop1 = base[4:12, 0:16] >>> self = crop1[2:6, 0:8] >>> opt = self._opt_fuse_crops() >>> self.write_network_text() >>> opt.write_network_text() >>> # >>> # Test Channel Select Via Index >>> self = base[:, :, [0]] >>> self.write_network_text() >>> final = self._finalize() >>> assert final.shape == (16, 16, 1) >>> assert base[:, :, [0, 1]].finalize().shape == (16, 16, 2) >>> assert base[:, :, [2, 0, 1]].finalize().shape == (16, 16, 3)
- optimize()[source]¶
- Returns
DelayedImage
Example
>>> # Test optimize nans >>> from kwcoco.util.delayed_ops import DelayedNans >>> import kwimage >>> base = DelayedNans(dsize=(100, 100), channels='a|b|c') >>> self = base[0:10, 0:5] >>> # Should simply return a new nan generator >>> new = self.optimize() >>> self.write_network_text() >>> new.write_network_text() >>> assert len(new.as_graph().nodes) == 1
- class kwcoco.util.delayed_ops.delayed_nodes.DelayedOverview(subdata, overview)[source]¶
Bases:
DelayedImage
Downsamples an image by a factor of two.
If the underlying image being loaded has precomputed overviews it simply loads these instead of downsampling the original image, which is more efficient.
Example
>>> # xdoctest: +REQUIRES(module:osgeo) >>> # Make a complex chain of operations and optimize it >>> from kwcoco.util.delayed_ops import * # NOQA >>> import kwimage >>> fpath = kwimage.grab_test_image_fpath(overviews=3) >>> dimg = DelayedLoad(fpath, channels='r|g|b').prepare() >>> dimg = dimg.get_overview(1) >>> dimg = dimg.get_overview(1) >>> dimg = dimg.get_overview(1) >>> dopt = dimg.optimize() >>> if 1: >>> import networkx as nx >>> dimg.write_network_text() >>> dopt.write_network_text() >>> print(ub.repr2(dopt.nesting(), nl=-1, sort=0)) >>> final0 = dimg._finalize()[:] >>> final1 = dopt._finalize()[:] >>> assert final0.shape == final1.shape >>> # xdoctest: +REQUIRES(--show) >>> import kwplot >>> kwplot.autompl() >>> kwplot.imshow(final0, pnum=(1, 2, 1), fnum=1, title='raw') >>> kwplot.imshow(final1, pnum=(1, 2, 2), fnum=1, title='optimized')
- property num_overviews¶
Returns: int
- kwcoco.util.delayed_ops.delayed_nodes.DelayedOverview2¶
alias of
DelayedOverview
- kwcoco.util.delayed_ops.delayed_nodes.DelayedCrop2¶
alias of
DelayedCrop
- kwcoco.util.delayed_ops.delayed_nodes.DelayedDequantize2¶
alias of
DelayedDequantize
- kwcoco.util.delayed_ops.delayed_nodes.DelayedWarp2¶
alias of
DelayedWarp
- kwcoco.util.delayed_ops.delayed_nodes.DelayedAsXarray2¶
alias of
DelayedAsXarray
- kwcoco.util.delayed_ops.delayed_nodes.DelayedImage2¶
alias of
DelayedImage
- kwcoco.util.delayed_ops.delayed_nodes.DelayedArray2¶
alias of
DelayedArray
- kwcoco.util.delayed_ops.delayed_nodes.DelayedChannelConcat2¶
alias of
DelayedChannelConcat
- kwcoco.util.delayed_ops.delayed_nodes.DelayedFrameStack2¶
alias of
DelayedFrameStack
- kwcoco.util.delayed_ops.delayed_nodes.DelayedConcat2¶
alias of
DelayedConcat
- kwcoco.util.delayed_ops.delayed_nodes.DelayedStack2¶
alias of
DelayedStack