Writing examples for the SEEREP API¶
To write a example it is a good idea to refer to already written ones. The focus in this documentation lies on flatbuffers type messages, because of the potential deprecation of protobuf in the project. All protobuf functionality can be replicated using flatbuffers, and flatbuffers should be used instead.
In this case the example gRPC_fb_addLabel.py will be reviewed. Service type definitions for all available flatbuffers type services can be found here. Type definitions of all flatbuffers types can be found here.
The code¶
import sys
import uuid
from typing import List, Optional, Tuple
import flatbuffers
from grpc import Channel
from seerep.fb import DatasetUuidLabel, Image, ProjectInfos
from seerep.fb import image_service_grpc_fb as imageService
from seerep.fb import meta_operations_grpc_fb as meta_ops
from seerep.util.common import get_gRPC_channel
from seerep.util.fb_helper import (
create_dataset_uuid_label,
create_label,
create_label_category,
createEmpty,
createQuery,
)
First some of the modules to interact with the servers services will be highlighted.
seerep.fb contains all python interfaces for the SEEREP services as well as the
Message types. sereep.util.fb_helper contains
helper functions related to flatbuffers, for instance functions to create a
message type directly.
Interaction with SEEREP services and handling the data¶
def add_label_raw(
target_proj_uuid: Optional[str] = None,
grpc_channel: Channel = get_gRPC_channel(),
) -> List[Tuple[str, bytearray]]:
The interaction functionality is contained within this function. With the
function definition, an option should be given to specify a target project, if
the message type allows setting it. Additionally the grpc_channel should be a
parameter in order to be able to target servers other than localhost:9090.
Both options are useful for testing later. More parameters can be added
optionally, if needed for the test cases.
stubMeta = meta_ops.MetaOperationsStub(grpc_channel)
# 3. Check if we have an existing test project, if not, one is created.
if target_proj_uuid is None:
# 2. Get all projects from the server
response = ProjectInfos.ProjectInfos.GetRootAs(
stubMeta.GetProjects(bytes(createEmpty(flatbuffers.Builder(1024))))
)
for i in range(response.ProjectsLength()):
proj_name = response.Projects(i).Name().decode()
proj_uuid = response.Projects(i).Uuid().decode()
print(proj_name + " " + proj_uuid)
if proj_name == "testproject":
target_proj_uuid = proj_uuid
if target_proj_uuid is None:
print("""
Please create a project with labeled images using
gRPC_pb_sendLabeledImage.py first."
""")
sys.exit()
At first, if target_proj_uuid is not set, the MetaOperationsStub utilizing
flatbuffers gRPC communication with the SEEREP server is used to retrieve a list
of all available projects of that server (specifically in the form of
project_infos.fbs
) and target_proj_uuid is set to the uuid of the first project with the name
testproject on that list.
stub = imageService.ImageServiceStub(grpc_channel)
builder = flatbuffers.Builder(1024)
query = createQuery(
builder,
projectUuids=[builder.CreateString(target_proj_uuid)],
withoutData=True,
)
builder.Finish(query)
buf = builder.Output()
response_ls: List = list(stub.GetImage(bytes(buf)))
if not response_ls:
print("""
No images found. Please create a project with labeled images
using gRPC_pb_sendLabeledImage.py first.
""")
sys.exit()
Following on the code requests all images from the project with the uuid of
target_proj_uuid using the ImageServiceStub. The service definition looks as
follows:
include "image.fbs";
include "query.fbs";
include "label_category.fbs";
include "dataset_uuid_label.fbs";
include "server_response.fbs";
namespace seerep.fb;
rpc_service ImageService {
GetImage(seerep.fb.Query):seerep.fb.Image (streaming: "server");
TransferImage(seerep.fb.Image):seerep.fb.ServerResponse (streaming: "client");
AddLabels(seerep.fb.DatasetUuidLabel):seerep.fb.ServerResponse (streaming: "client");
}
GetImage() takes a argument of type
seerep.fb.Query,
a more generic build query type for use in various services in SEEREP, in it's
serialized form and returns data of type
seerep.fb.Image
from the server.
msgToSend = []
label_list: List[Tuple[str, bytearray]] = []
for responseBuf in response_ls:
response = Image.Image.GetRootAs(responseBuf)
img_uuid = response.Header().UuidMsgs().decode("utf-8")
projectUuid = response.Header().UuidProject().decode("utf-8")
labelStr = ["label1", "label2"]
labels = []
for labelAct in labelStr:
labels.append(
create_label(
builder=builder,
label=labelAct,
label_id=1,
instance_uuid=str(uuid.uuid4()),
instance_id=2,
)
)
labelsCategory = []
labelsCategory.append(
create_label_category(
builder=builder,
labels=labels,
datumaro_json="a very valid datumaro json",
category="laterAddedLabel",
)
)
dataset_uuid_label = create_dataset_uuid_label(
builder=builder,
projectUuid=projectUuid,
datasetUuid=img_uuid,
labels=labelsCategory,
)
builder.Finish(dataset_uuid_label)
buf = builder.Output()
label_list.append(
(
img_uuid,
buf,
)
)
msgToSend.append(bytes(buf))
This code builds a list of DatasetUuidLabel adding some sample data into the
components of each Label Category. At the beginning two lists are defined msgToSend
is a list containing the serialized labels and label_list is a list
containing mappings where each image uuid is mapped to it's added labels.
After that the returned images from the query before are iterated. Next
Label are created and their joint header uuids are set to the
appropriate project_uuid and msg_uuid to match that specific image. At the
end the Labels are serialized and added to the lists.
The type definition of LabelCategory looks as follows:
include "label.fbs";
namespace seerep.fb;
table LabelCategory {
category:string;
labels:[Label];
datumaroJson:string;
}
root_type LabelCategory;
And the type definition of Label looks as follows:
namespace seerep.fb;
table Label {
label:string;
labelIdDatumaro:int;
instanceUuid:string;
instanceIdDatumaro:int;
}
root_type Label;
Lastly the service is called, the LabelCategory are send to the SEEREP server
and the list with the mappings is returned for further use. Note that the
flatbuffers objects are not returned in their deserialized state as the function
fb_flatc_dict defined in
here
makes use of that state.
Wrapping the raw function¶
def add_label(
target_proj_uuid: Optional[str] = None,
grpc_channel: Channel = get_gRPC_channel(),
) -> List[Tuple[str, DatasetUuidLabel.DatasetUuidLabel]]:
return [
(img_uuid, DatasetUuidLabel.DatasetUuidLabel.GetRootAs(labelbuf))
for img_uuid, labelbuf in add_label_raw(target_proj_uuid, grpc_channel)
]
This function is essentially just a wrapper for add_bb_raw() to return the
deserialized objects to be accessed through their regular flatbuffers interfaces
(in this case of type
DatasetUuidLabel.DatasetUuidLabel).
Allow for independent execution of the script¶
if __name__ == "__main__":
label_list = add_label()
for img_uuid, labelAllCat in label_list:
print(f"Added label to image with uuid {img_uuid}:")
for labelCategory in labelAllCat.Labels():
for label in labelCategory.Labels():
print(f"uuid: {label.Label().decode()}")
The last part can execute the script independently and targets the server at the
default address, which is localhost:9090. On successful execution a subset of
the sent data based on the returned mapping is printed.
Some important considerations¶
To conclude for most of the examples it is best practice to follow this structure,
namely first having a function which returns the serialized data, then wrapping
that function to return the deserialized variant and at the end the
if __name__ == "__main__" part of the script, such that the script can be
executed independently. Of course functionality can be outsourced into other
functions, when it makes sense. This structure eases the process of writing
tests later on (see writing-tests.md), especially
when fb_flatc_dict should be utilized.