#!/usr/bin/env python3
import scriptconfig as scfg
import ubelt as ub
#TODO: Either use coco_add_ignore_buffer here or use run_all_pointgen to calculate it
[docs]
class CocoAddIgnoreBufferConfig(scfg.DataConfig):
"""
Given a kwcoco file and buffer size update annotations to include ignore buffer regions around polygons
"""
src = scfg.Value(
None,
help="Specify input kwcoco filepath",
)
ignore_buffer_size = scfg.Value(
"10@10GSD",
help="give a resolved unit (e.g. 10@10mGSD) for a buffer size around each other polygon.",
)
dst = scfg.Value(None, help="Specify output kwcoco filepath")
[docs]
def main(cmdline=1, **kwargs):
r"""
IGNORE:
python $HOME/Desktop/geowatch/geowatch/cli/coco_add_ignore_buffer.py \
--src $HOME/Desktop/dvc_repos/smart_phase3_data/Aligned-Drop8-ARA/KR_R002/imganns-KR_R002-rawbands.kwcoco.zip \
--dst $HOME/Desktop/dvc_repos/smart_phase3_data/Aligned-Drop8-ARA/KR_R002/imganns-KR_R002_modified-rawbands.kwcoco.zip
CommandLine:
xdoctest -m geowatch.cli.coco_add_ignore_buffer
xdoctest $HOME/Desktop/geowatch/geowatch/cli/coco_add_ignore_buffer.py
Example:
>>> from geowatch.cli.coco_add_ignore_buffer import *
>>> import geowatch
>>> import ubelt as ub
>>> import kwcoco
>>> dpath = ub.Path.appdir('geowatch/tests/ignore_buffer')
>>> dpath.ensuredir()
>>> ignore_buffer_size = '10@10GSD'
>>> dst = dpath / 'out.kwcoco.zip'
>>> src = geowatch.coerce_kwcoco('geowatch-msi', geodata=True, dates=True)
>>> kwargs = dict(src=src.data_fpath, dst=dst)
>>> main(cmdline=0, **kwargs)
>>> result_dest = kwcoco.CocoDataset(dst)
>>> # Check non of the ignore polygons are overlaping non-ignore
>>> from shapely.ops import unary_union
>>> for video_id in result_dest.videos():
>>> images = result_dest.images(video_id=video_id)
>>> for image_id in images:
>>> src_coco_img = src.coco_image(image_id)
>>> dst_coco_img = result_dest.coco_image(image_id)
>>> src_annots = src_coco_img.annots()
>>> dst_annots = dst_coco_img.annots()
>>> new_aids = ub.oset(dst_annots) - ub.oset(src_annots)
>>> old_aids = list(src_annots)
>>> # The old polygons (new polys should not intersect these)
>>> old_polys = [p.to_shapely() for p in result_dest.annots(old_aids).detections.data['segmentations']]
>>> # The new polygons that should be the ignored regions
>>> new_polys = [p.to_shapely() for p in result_dest.annots(new_aids).detections.data['segmentations']]
>>> new_poly = unary_union(new_polys)
>>> old_poly = unary_union(old_polys)
>>> isect_poly = new_poly.intersection(old_poly)
>>> union_poly = new_poly.union(old_poly)
>>> iou = isect_poly.area / union_poly.area
>>> print(f'image_id={image_id}, iou = {ub.urepr(iou, nl=1)}')
>>> # The iou should be nearly zero (up to float errors)
>>> assert iou < 1e-5
"""
import rich
from rich.markup import escape
config = CocoAddIgnoreBufferConfig.cli(
cmdline=cmdline, data=kwargs,
# special_options=False # requires recent scriptconfig
)
rich.print("config = " + escape(ub.urepr(config)))
import kwimage
from geowatch.utils import util_resolution
from shapely.ops import unary_union
import numpy as np
import kwcoco
import kwutil
if config.src is None:
raise ValueError('must specify src kwcoco')
dset = kwcoco.CocoDataset(config.src)
dset = dset.reroot(absolute=True)
# TODO: if unit is not specified, work in videospace instead of world space
utm_gsd = util_resolution.ResolvedUnit.coerce("1GSD")
ignore_buffer = util_resolution.ResolvedScalar.coerce(config.ignore_buffer_size)
ignore_buffer_gsd = ignore_buffer.at_resolution(utm_gsd).scalar
videos = dset.videos()
pman = kwutil.util_progress.ProgressManager()
with pman:
# Ignore eff
for video_id in pman.progiter(videos, desc="looping over videos..."):
images = dset.images(video_id=video_id)
for image_id in pman.progiter(images, desc="looping over images..."):
coco_img = dset.coco_image(image_id)
_imgspace_resolution = coco_img.resolution(space="image")
image_pxl_per_meter = 1 / np.array(_imgspace_resolution["mag"])
ignore_buffer_pixel = ignore_buffer_gsd * image_pxl_per_meter
# TODO: buffer utilzing shapely method currently only accounts
# for a singular float distance need to account for several
# distances in the future, but for now the average of the
# buffer region suggested is used.
ignore_buffer_pixel = ignore_buffer_pixel.mean()
annots = coco_img.annots()
# annot_cat_ids = annots.lookup("category_id")
annot_segmenations = annots.lookup("segmentation")
# print(annot_segmenations)
# annot_cat_names = dset.categories(annot_cat_ids).lookup("name")
annot_polys = [
kwimage.MultiPolygon.coerce(s).to_shapely()
for s in annot_segmenations
]
# We do not want to ignore any existing annotation region.
do_not_ignore_poly = unary_union(annot_polys)
# For each existing annotation
new_ignore_polys = []
SANITY_CHECK = 0
if SANITY_CHECK == 1:
for poly in annot_polys:
expanded_poly = poly.buffer(ignore_buffer_pixel)
# SANITY_CHECK = 0
# TODO: dont use big whiles
# SANITY_CHECK
iou = 1
while iou > 0.0001 and not expanded_poly.is_empty:
# Expand the region around it
expanded_poly = poly.buffer(ignore_buffer_pixel)
# Remove any regions touching existing annotation
new_ignore_geom = expanded_poly - do_not_ignore_poly
for nonignore_poly in annot_polys:
isect_poly = nonignore_poly.intersection(
new_ignore_geom
)
union_poly = nonignore_poly.union(new_ignore_geom)
iou = isect_poly.area / union_poly.area
# print(iou)
expanded_poly = new_ignore_geom
if not new_ignore_geom.is_empty:
new_ignore_polys.append(expanded_poly)
else:
for poly in annot_polys:
expanded_poly = poly.buffer(ignore_buffer_pixel)
# Expand the region around it
# Remove any regions touching existing annotation
new_ignore_geom = expanded_poly - do_not_ignore_poly
if not new_ignore_geom.is_empty:
new_ignore_polys.append(new_ignore_geom)
if 0:
# kwimage.MultiPolygon.coerce(do_not_ignore_poly).draw(setlim=1,color='kitware_red')
kwimage.MultiPolygon.coerce(annot_polys[0]).draw(
color="kitware_green"
)
for poly in new_ignore_polys:
kwimage.MultiPolygon.coerce(poly).draw(color="kitware_blue")
kwimage.MultiPolygon.coerce(do_not_ignore_poly).draw(
setlim=1, color="kitware_red"
)
for poly in new_ignore_polys:
_poly = kwimage.MultiPolygon.from_shapely(poly)
dset.add_annotation(
image_id=image_id,
category_id=dset.ensure_category("ignore"),
bbox=_poly.bounding_box().to_xywh(),
segmentation=_poly,
)
# g.plot() #Shows base polys
# kwplot.kwplot.plt.show()
# f.plot() #Shows the aftermath of do_not_ignore_poly
# kwplot.plt.show()
# p.plot() # Shows all polys with their buffer
# kwplot.plt.show()
out_path = config.dst
# Write to the compressed path
dset.dump(out_path)
rich.print(f"Wrote modified kwcoco to: [link={out_path}]{out_path}[/link]")
__config__ = CocoAddIgnoreBufferConfig
if __name__ == "__main__":
"""
python -m geowatch.cli.coco_add_ignore_buffer
"""
main()