in source/consumer/lambda_handler.py [0:0]
def process_label_detection(asset, workflow, results):
# Rekognition label detection puts labels on an inner array in its JSON result, but for ease of search in Elasticsearch we need those results as a top level json array. So this function does that.
metadata = json.loads(results)
es = connect_es(es_endpoint)
extracted_items = []
# We can tell if json results are paged by checking to see if the json results are an instance of the list type.
if isinstance(metadata, list):
# handle paged results
for page in metadata:
if "Labels" in page:
for item in page["Labels"]:
try:
item["Operator"] = "label_detection"
item["Workflow"] = workflow
if "Label" in item:
# Flatten the inner Label array
item["Confidence"] = item["Label"]["Confidence"]
item["Name"] = item["Label"]["Name"]
item["Instances"] = ''
if 'Instances' in item["Label"]:
item["Instances"] = item["Label"]["Instances"]
item["Parents"] = ''
if 'Parents' in item["Label"]:
item["Parents"] = item["Label"]["Parents"]
# Delete the flattened array
del item["Label"]
extracted_items.append(item)
except KeyError as e:
print("KeyError: " + str(e))
print("Item: " + json.dumps(item))
else:
# these results are not paged
if "Labels" in metadata:
for item in metadata["Labels"]:
try:
item["Operator"] = "label_detection"
item["Workflow"] = workflow
if "Label" in item:
# Flatten the inner Label array
item["Confidence"] = item["Label"]["Confidence"]
item["Name"] = item["Label"]["Name"]
item["Instances"] = ''
if 'Instances' in item["Label"]:
item["Instances"] = item["Label"]["Instances"]
item["Parents"] = ''
if 'Parents' in item["Label"]:
item["Parents"] = item["Label"]["Parents"]
# Delete the flattened array
del item["Label"]
extracted_items.append(item)
except KeyError as e:
print("KeyError: " + str(e))
print("Item: " + json.dumps(item))
bulk_index(es, asset, "labels", extracted_items)