Custom Writer
Learning Objectives
Replicator comes with a basicWriter which provides labeled output. Although the basicWriter is generic enough, Replicator provides a way to create a custom writer to address custom output format requirements. Through this sample shows the process to create a custom writer.
Making a custom writer
To implement a custom writer we will need to implement the class Writer. For the specifics of the implementation, a detailed description is below.
Implementing a Writer
To create a custom Write, first we have to implement a Writer class detailed in API documentation .
In this example we are creating a 2D bounding box writer that is only writing the bounding box information for our desired class, in this case worker, for your use case, you might have different classes you are interested in.
First let’s initialize the writer. In this case the only mandatory parameter is the output directory. Notice here the use of annotators. Annotators are the types of ground truth annotations created by Replicator.
Also note that controlling and filtering semantics can be further explored in Adding Semantics to a Scene.
def __init__(
self,
output_dir,
rgb: bool = True,
bounding_box_2d_tight: bool = False,
image_output_format="png",
):
self._output_dir = output_dir
self._backend = BackendDispatch({"paths": {"out_dir": output_dir}})
self._frame_id = 0
self._image_output_format = image_output_format
self.annotators = []
# RGB
if rgb:
self.annotators.append(AnnotatorRegistry.get_annotator("rgb"))
# Bounding Box 2D
if bounding_box_2d_tight:
self.annotators.append(AnnotatorRegistry.get_annotator("bounding_box_2d_tight",
init_params={"semanticTypes": ["class"]}))
Then we have a helper function. Here we check that the bounding box meets minimum size criteria.
def check_bbox_area(self, bbox_data, size_limit):
length = abs(bbox_data['x_min'] - bbox_data['x_max'])
width = abs(bbox_data['y_min'] - bbox_data['y_max'])
area = length * width
if area > size_limit:
return True
else:
return False
Lastly we have the most important function write. In this case it takes in the data, checks if in data dictionary is rgb and bounding_box_2d_tight.
If it is in there then we extract the bounding box information and the labels. We then go through each of the labels, if the label is worker, then we write the data in coco like format after we check the size.
def write(self, data):
if "rgb" in data and "bounding_box_2d_tight" in data:
bbox_data = data["bounding_box_2d_tight"]["data"]
id_to_labels = data["bounding_box_2d_tight"]["info"]["idToLabels"]
for id, labels in id_to_labels.items():
id = int(id)
if 'worker' in labels:
target_bbox_data = {'x_min': bbox_data['x_min'], 'y_min': bbox_data['y_min'],
'x_max': bbox_data['x_max'], 'y_max': bbox_data['y_max']}
if self.check_bbox_area(target_bbox_data, 0.5):
width = int(abs(target_bbox_data["x_max"][0] - target_bbox_data["x_min"][0]))
height = int(abs(target_bbox_data["y_max"][0] - target_bbox_data["y_min"][0]))
if width != 2147483647 and height != 2147483647:
filepath = f"rgb_{self._frame_id}.{self._image_output_format}"
self._backend.write_image(filepath, data["rgb"])
bbox_filepath = f"bbox_{self._frame_id}.json"
coco_bbox_data = {"x": int(target_bbox_data["x_max"][0]),
"y": int(target_bbox_data["y_max"][0]),
"width": width,
"height": height}
buf = io.BytesIO()
buf.write(json.dumps(coco_bbox_data).encode())
self._backend.write_blob(bbox_filepath, buf.getvalue())
Here is the complete class implemented.
import time
import asyncio
import json
import io
import omni.kit
import omni.usd
import omni.replicator.core as rep
from omni.replicator.core import Writer, AnnotatorRegistry, BackendDispatch
class WorkerWriter(Writer):
def __init__(
self,
output_dir,
rgb: bool = True,
bounding_box_2d_tight: bool = False,
image_output_format="png",
):
self._output_dir = output_dir
self._backend = BackendDispatch({"paths": {"out_dir": output_dir}})
self._frame_id = 0
self._image_output_format = image_output_format
self.annotators = []
# RGB
if rgb:
self.annotators.append(AnnotatorRegistry.get_annotator("rgb"))
# Bounding Box 2D
if bounding_box_2d_tight:
self.annotators.append(AnnotatorRegistry.get_annotator("bounding_box_2d_tight",
init_params={"semanticTypes": ["class"]}))
def check_bbox_area(self, bbox_data, size_limit):
length = abs(bbox_data['x_min'] - bbox_data['x_max'])
width = abs(bbox_data['y_min'] - bbox_data['y_max'])
area = length * width
if area > size_limit:
return True
else:
return False
def write(self, data):
if "rgb" in data and "bounding_box_2d_tight" in data:
bbox_data = data["bounding_box_2d_tight"]["data"]
id_to_labels = data["bounding_box_2d_tight"]["info"]["idToLabels"]
for id, labels in id_to_labels.items():
id = int(id)
if 'worker' in labels:
target_bbox_data = {'x_min': bbox_data['x_min'], 'y_min': bbox_data['y_min'],
'x_max': bbox_data['x_max'], 'y_max': bbox_data['y_max']}
if self.check_bbox_area(target_bbox_data, 0.5):
width = int(abs(target_bbox_data["x_max"][0] - target_bbox_data["x_min"][0]))
height = int(abs(target_bbox_data["y_max"][0] - target_bbox_data["y_min"][0]))
if width != 2147483647 and height != 2147483647:
filepath = f"rgb_{self._frame_id}.{self._image_output_format}"
self._backend.write_image(filepath, data["rgb"])
bbox_filepath = f"bbox_{self._frame_id}.json"
coco_bbox_data = {"x": int(target_bbox_data["x_max"][0]),
"y": int(target_bbox_data["y_max"][0]),
"width": width,
"height": height}
buf = io.BytesIO()
buf.write(json.dumps(coco_bbox_data).encode())
self._backend.write_blob(bbox_filepath, buf.getvalue())
self._frame_id += 1
End to end example
Following the set up steps in Setting up the Script Editor, you can copy the script below in the editor. After running the script from the editor run replicator following Running and Previewing Replicator. For a deeper understanding on the randomization piece of the script check out Randomizing appearance, placement and orientation of an existing 3D assets with a built-in writer. For more examples of writers, you can check the script defining the BasicWriter and other writers in replicators script folder. To find that folder follow the steps shown in Scripts for Replicator.
import time
import asyncio
import json
import io
import omni.kit
import omni.usd
import omni.replicator.core as rep
from omni.replicator.core import Writer, AnnotatorRegistry, BackendDispatch
WORKER = 'omniverse://localhost/NVIDIA/Assets/Characters/Reallusion/Worker/Worker.usd'
PROPS = 'omniverse://localhost/NVIDIA/Assets/Vegetation/Shrub'
ENVS = 'omniverse://localhost/NVIDIA/Assets/Scenes/Templates/Outdoor/Puddles.usd'
SURFACE = 'omniverse://localhost/NVIDIA/Assets/Scenes/Templates/Basic/display_riser.usd'
class WorkerWriter(Writer):
def __init__(
self,
output_dir,
rgb: bool = True,
bounding_box_2d_tight: bool = False,
image_output_format="png",
):
self._output_dir = output_dir
self._backend = BackendDispatch({"paths": {"out_dir": output_dir}})
self._frame_id = 0
self._image_output_format = image_output_format
self.annotators = []
# RGB
if rgb:
self.annotators.append(AnnotatorRegistry.get_annotator("rgb"))
# Bounding Box 2D
if bounding_box_2d_tight:
self.annotators.append(AnnotatorRegistry.get_annotator("bounding_box_2d_tight",
init_params={"semanticTypes": ["class"]}))
def check_bbox_area(self, bbox_data, size_limit):
length = abs(bbox_data['x_min'] - bbox_data['x_max'])
width = abs(bbox_data['y_min'] - bbox_data['y_max'])
area = length * width
if area > size_limit:
return True
else:
return False
def write(self, data):
if "rgb" in data and "bounding_box_2d_tight" in data:
bbox_data = data["bounding_box_2d_tight"]["data"]
id_to_labels = data["bounding_box_2d_tight"]["info"]["idToLabels"]
for id, labels in id_to_labels.items():
id = int(id)
if 'worker' in labels:
target_bbox_data = {'x_min': bbox_data['x_min'], 'y_min': bbox_data['y_min'],
'x_max': bbox_data['x_max'], 'y_max': bbox_data['y_max']}
if self.check_bbox_area(target_bbox_data, 0.5):
width = int(abs(target_bbox_data["x_max"][0] - target_bbox_data["x_min"][0]))
height = int(abs(target_bbox_data["y_max"][0] - target_bbox_data["y_min"][0]))
if width != 2147483647 and height != 2147483647:
filepath = f"rgb_{self._frame_id}.{self._image_output_format}"
self._backend.write_image(filepath, data["rgb"])
bbox_filepath = f"bbox_{self._frame_id}.json"
coco_bbox_data = {"x": int(target_bbox_data["x_max"][0]),
"y": int(target_bbox_data["y_max"][0]),
"width": width,
"height": height}
buf = io.BytesIO()
buf.write(json.dumps(coco_bbox_data).encode())
self._backend.write_blob(bbox_filepath, buf.getvalue())
self._frame_id += 1
rep.WriterRegistry.register(WorkerWriter)
with rep.new_layer():
def env_props(size=50):
instances = rep.randomizer.instantiate(rep.utils.get_usd_files(PROPS), size=size, mode='point_instance')
with instances:
rep.modify.pose(
position=rep.distribution.uniform((-500, 0, -500), (500, 0, 500)),
rotation=rep.distribution.uniform((-90, -180, 0), (-90, 180, 0)),
)
return instances.node
def worker():
worker = rep.create.from_usd(WORKER, semantics=[('class', 'worker')])
with worker:
rep.modify.semantics([('class', 'worker')])
rep.modify.pose(
position=rep.distribution.uniform((-500, 0, -500), (500, 0, 500)),
rotation=rep.distribution.uniform((-90, -45, 0), (-90, 45, 0)),
)
return worker
rep.randomizer.register(env_props)
rep.randomizer.register(worker)
# Add Default Light
distance_light = rep.create.light(rotation=(315,0,0), intensity=3000, light_type="distant")
# Setup the static elements
env = rep.create.from_usd(ENVS)
table = rep.create.from_usd(SURFACE)
# Setup camera and attach it to render product
camera = rep.create.camera(
focus_distance=800,
f_stop=0.5
)
# Setup randomization
with rep.trigger.on_frame(num_frames=10):
rep.randomizer.env_props(10)
rep.randomizer.worker()
with camera:
rep.modify.pose(position=rep.distribution.uniform((-500, 200, 1000), (500, 500, 1500)), look_at=table)
render_product = rep.create.render_product(camera, resolution=(1024, 1024))
# Initialize and attach writer
# writer = rep.WriterRegistry.get("OmniWriter")
writer = rep.WriterRegistry.get("WorkerWriter")
out_dir = "custom_writer_output"
writer.initialize(output_dir=out_dir, rgb=True, bounding_box_2d_tight=True)
writer.attach([render_product])