in iotthingshadow/src-mapthingname/app.py [0:0]
def lambda_handler(event, context):
""" Determines a thing name based on an attribute of LoRAWAN message (e.g. WirelessDeviceId)
Parameters (as event attributes)
----------
searchvalue : str
value that will be used for a lookup of thing name
Environment variable
----------------
SEARCH_TYPE : str
Can have values "ASSOCIATED_THING" or "THING_INDEX"
If value "ASSOCIATED_THING" is provided, input must be a AWS IoT Core for LoRaWAN WirelessDeviceId The function will return the name of the thing associated to a specified WirelessDeviceId.
If value "THING_INDEX" is provided, an additional environment variable SEARCH_THING_ATTRIBUTENAME must be specified. This function will perform a search in the IoT registry for an attribute named as SEARCH_THING_ATTRIBUTENAME, and match for the value of the input "searchvalue".
Returns
-------
This function returns a JSON object with the following keys:
ThingName : str
Name of AWS IoT Thing
Error handling
-------
This function will raise an Exception in case of an error
"""
global thing_name_cache
logger.info("Received event: %s" % json.dumps(event))
# Get the search value (for example, if using 'WirelessDeviceId' as a search attribute,
# the search value would be like 8b00de4a-0fac-407b-93e6-8c59fd411f16")
if not "searchvalue" in event:
raise Exception("Missing event attribute 'searchvalue'")
search_value = event.get("searchvalue")
thing_name = None
# If thing name is cached, return the value stored in cache
if thing_name_cache.get(search_value, None) != None:
thing_name = thing_name_cache.get(search_value)
logger.info("Found value [%s] for key [%s] in cache" % (
thing_name, search_value))
# If search type is ASSOCIATED_THING, invoke AWS IoT Core for LoRaWAN API to retrieve a thing associated
# with this device (see https://docs.aws.amazon.com/iot-wireless/2020-11-22/apireference/API_AssociateWirelessDeviceWithThing.html)
elif PARAM_SEARCH_TYPE == SEARCH_TYPE_ASSOCIATED_THING:
logger.info("Query associated thing for id: %s" % search_value)
thing_name = client_iotwireless.get_wireless_device(
Identifier=search_value,
IdentifierType='WirelessDeviceId'
).get("ThingName")
thing_name_cache[search_value] = thing_name
# If search type is THING_INDEX, invoke AWS IoT Core API to retrieve a Thing based on attribute name
# (see https://docs.aws.amazon.com/iot/latest/apireference/API_SearchIndex.html)
elif PARAM_SEARCH_TYPE == SEARCH_TYPE_THING_INDEX:
# Build a query string
# (see details at https://docs.aws.amazon.com/iot/latest/developerguide/example-queries.html)
query_string = "attributes."+PARAM_SEARCH_THING_ATTRIBUTE+":" + search_value
logger.info("Query string: %s" % query_string)
# Invoke the index search
search_result = client_iot.search_index(
indexName='AWS_Things',
queryString=query_string
).get("things")
# Error handling
if (len(search_result) == 0):
raise Exception(
"Error, query [%s] did not return any results" % query_string)
if (len(search_result) > 1):
raise Exception(
"Error, query [%s] returned more the one result" % query_string)
# Extract the name of AWS IoT Thing
thing_name = search_result[0].get("thingName")
thing_name_cache[search_value] = thing_name
else:
raise Exception("Unsupported search type: [%s]" % PARAM_SEARCH_TYPE)
logger.info("Thing name is: %s" % thing_name)
return {PARAM_OUTPUT_ATTRIBUTE_NAME: thing_name}