temp_controller_with_thermostats.py

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
importos
importasyncio
importrandom
importlogging

fromazure.iot.device.aioimportIoTHubDeviceClient
fromazure.iot.device.aioimportProvisioningDeviceClient
fromazure.iot.deviceimportMethodResponse
fromdatetimeimporttimedelta, datetime
importpnp_helper

logging.basicConfig(level=logging.ERROR)

# the interfaces that are pulled in to implement the device.
# User has to know these values as these may change and user can
# choose to implement different interfaces.
thermostat_digital_twin_model_identifier="dtmi:com:example:Thermostat;1"
device_info_digital_twin_model_identifier="dtmi:azure:DeviceManagement:DeviceInformation;1"

# The device "TemperatureController" that is getting implemented using the above interfaces.
# This id can change according to the company the user is from
# and the name user wants to call this Plug and Play device
model_id="dtmi:com:example:TemperatureController;2"

# the components inside this Plug and Play device.
# there can be multiple components from 1 interface
# component names according to interfaces following pascal case.
device_information_component_name="deviceInformation"
thermostat_1_component_name="thermostat1"
thermostat_2_component_name="thermostat2"
serial_number="some_serial_number"
#####################################################
# COMMAND HANDLERS : User will define these handlers
# depending on what commands the component defines

#####################################################
# GLOBAL VARIABLES
THERMOSTAT_1=None
THERMOSTAT_2=None


classThermostat(object):
def__init__(self, name, moving_win=10):

self.moving_window=moving_win
self.records= [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
self.index=0

self.cur=0
self.max=0
self.min=0
self.avg=0

self.name=name

defrecord(self, current_temp):
self.cur=current_temp
self.records[self.index] =current_temp
self.max=self.calculate_max(current_temp)
self.min=self.calculate_min(current_temp)
self.avg=self.calculate_average()

self.index= (self.index+1) %self.moving_window

defcalculate_max(self, current_temp):
ifnotself.max:
returncurrent_temp
elifcurrent_temp>self.max:
returnself.max

defcalculate_min(self, current_temp):
ifnotself.min:
returncurrent_temp
elifcurrent_temp<self.min:
returnself.min

defcalculate_average(self):
returnsum(self.records) /self.moving_window

defcreate_report(self):
response_dict= {}
response_dict["maxTemp"] =self.max
response_dict["minTemp"] =self.min
response_dict["avgTemp"] =self.avg
response_dict["startTime"] = (
 (datetime.now() -timedelta(0, self.moving_window*8)).astimezone().isoformat()
 )
response_dict["endTime"] =datetime.now().astimezone().isoformat()
returnresponse_dict


asyncdefreboot_handler(values):
ifvalues:
print("Rebooting after delay of {delay} secs".format(delay=values))
print("Done rebooting")


asyncdefmax_min_handler(values):
ifvalues:
print(
"Will return the max, min and average temperature from the specified time {since} to the current time".format(
since=values
 )
 )
print("Done generating")


# END COMMAND HANDLERS
#####################################################

#####################################################
# CREATE RESPONSES TO COMMANDS


defcreate_max_min_report_response(thermostat_name):
"""
 An example function that can create a response to the "getMaxMinReport" command request the way the user wants it.
 Most of the times response is created by a helper function which follows a generic pattern.
 This should be only used when the user wants to give a detailed response back to the Hub.
 :param values: The values that were received as part of the request.
 """
if"Thermostat;1"inthermostat_nameandTHERMOSTAT_1:
response_dict=THERMOSTAT_1.create_report()
elifTHERMOSTAT_2:
response_dict=THERMOSTAT_2.create_report()
else: # This is done to pass certification.
response_dict= {}
response_dict["maxTemp"] =0
response_dict["minTemp"] =0
response_dict["avgTemp"] =0
response_dict["startTime"] =datetime.now().astimezone().isoformat()
response_dict["endTime"] =datetime.now().astimezone().isoformat()

print(response_dict)
returnresponse_dict


# END CREATE RESPONSES TO COMMANDS
#####################################################

#####################################################
# TELEMETRY TASKS


asyncdefsend_telemetry_from_temp_controller(device_client, telemetry_msg, component_name=None):
msg=pnp_helper.create_telemetry(telemetry_msg, component_name)
awaitdevice_client.send_message(msg)
print("Sent message")
print(msg)
awaitasyncio.sleep(5)


#####################################################
# COMMAND TASKS


asyncdefexecute_command_listener(
device_client,
component_name=None,
method_name=None,
user_command_handler=None,
create_user_response_handler=None,
):
"""
 Coroutine for executing listeners. These will listen for command requests.
 They will take in a user provided handler and call the user provided handler
 according to the command request received.
 :param device_client: The device client
 :param component_name: The name of the device like "sensor"
 :param method_name: (optional) The specific method name to listen for. Eg could be "blink", "turnon" etc.
 If not provided the listener will listen for all methods.
 :param user_command_handler: (optional) The user provided handler that needs to be executed after receiving "command requests".
 If not provided nothing will be executed on receiving command.
 :param create_user_response_handler: (optional) The user provided handler that will create a response.
 If not provided a generic response will be created.
 :return:
 """
whileTrue:
ifcomponent_nameandmethod_name:
command_name=component_name+"*"+method_name
elifmethod_name:
command_name=method_name
else:
command_name=None

command_request=awaitdevice_client.receive_method_request(command_name)
print("Command request received with payload")
values=command_request.payload
print(values)

ifuser_command_handler:
awaituser_command_handler(values)
else:
print("No handler provided to execute")

 (response_status, response_payload) =pnp_helper.create_response_payload_with_status(
command_request, method_name, create_user_response=create_user_response_handler
 )

command_response=MethodResponse.create_from_method_request(
command_request, response_status, response_payload
 )

try:
awaitdevice_client.send_method_response(command_response)
exceptException:
print("responding to the {command} command failed".format(command=method_name))


#####################################################
# PROPERTY TASKS


asyncdefexecute_property_listener(device_client):
whileTrue:
patch=awaitdevice_client.receive_twin_desired_properties_patch() # blocking call
print(patch)
properties_dict=pnp_helper.create_reported_properties_from_desired(patch)

awaitdevice_client.patch_twin_reported_properties(properties_dict)


#####################################################
# An # END KEYBOARD INPUT LISTENER to quit application


defstdin_listener():
"""
 Listener for quitting the sample
 """
whileTrue:
selection=input("Press Q to quit\n")
ifselection=="Q"orselection=="q":
print("Quitting...")
break


# END KEYBOARD INPUT LISTENER
#####################################################


#####################################################
# MAIN STARTS
asyncdefprovision_device(provisioning_host, id_scope, registration_id, symmetric_key, model_id):
provisioning_device_client=ProvisioningDeviceClient.create_from_symmetric_key(
provisioning_host=provisioning_host,
registration_id=registration_id,
id_scope=id_scope,
symmetric_key=symmetric_key,
 )

provisioning_device_client.provisioning_payload= {"modelId": model_id}
returnawaitprovisioning_device_client.register()


asyncdefmain():
switch=os.getenv("IOTHUB_DEVICE_SECURITY_TYPE")
ifswitch=="DPS":
provisioning_host= (
os.getenv("IOTHUB_DEVICE_DPS_ENDPOINT")
ifos.getenv("IOTHUB_DEVICE_DPS_ENDPOINT")
else"global.azure-devices-provisioning.net"
 )
id_scope=os.getenv("IOTHUB_DEVICE_DPS_ID_SCOPE")
registration_id=os.getenv("IOTHUB_DEVICE_DPS_DEVICE_ID")
symmetric_key=os.getenv("IOTHUB_DEVICE_DPS_DEVICE_KEY")

registration_result=awaitprovision_device(
provisioning_host, id_scope, registration_id, symmetric_key, model_id
 )

ifregistration_result.status=="assigned":
print("Device was assigned")
print(registration_result.registration_state.assigned_hub)
print(registration_result.registration_state.device_id)
device_client=IoTHubDeviceClient.create_from_symmetric_key(
symmetric_key=symmetric_key,
hostname=registration_result.registration_state.assigned_hub,
device_id=registration_result.registration_state.device_id,
product_info=model_id,
 )
else:
raiseRuntimeError(
"Could not provision device. Aborting Plug and Play device connection."
 )

elifswitch=="connectionString":
conn_str=os.getenv("IOTHUB_DEVICE_CONNECTION_STRING")
print("Connecting using Connection String "+conn_str)
device_client=IoTHubDeviceClient.create_from_connection_string(
conn_str, product_info=model_id
 )
else:
raiseRuntimeError(
"At least one choice needs to be made for complete functioning of this sample."
 )

# Connect the client.
awaitdevice_client.connect()

################################################
# Update readable properties from various components

properties_root=pnp_helper.create_reported_properties(serialNumber=serial_number)
properties_thermostat1=pnp_helper.create_reported_properties(
thermostat_1_component_name, maxTempSinceLastReboot=98.34
 )
properties_thermostat2=pnp_helper.create_reported_properties(
thermostat_2_component_name, maxTempSinceLastReboot=48.92
 )
properties_device_info=pnp_helper.create_reported_properties(
device_information_component_name,
swVersion="5.5",
manufacturer="Contoso Device Corporation",
model="Contoso 4762B-turbo",
osName="Mac Os",
processorArchitecture="x86-64",
processorManufacturer="Intel",
totalStorage=1024,
totalMemory=32,
 )

property_updates=asyncio.gather(
device_client.patch_twin_reported_properties(properties_root),
device_client.patch_twin_reported_properties(properties_thermostat1),
device_client.patch_twin_reported_properties(properties_thermostat2),
device_client.patch_twin_reported_properties(properties_device_info),
 )

################################################
# Get all the listeners running
print("Listening for command requests and property updates")

globalTHERMOSTAT_1
globalTHERMOSTAT_2
THERMOSTAT_1=Thermostat(thermostat_1_component_name, 10)
THERMOSTAT_2=Thermostat(thermostat_2_component_name, 10)

listeners=asyncio.gather(
execute_command_listener(
device_client, method_name="reboot", user_command_handler=reboot_handler
 ),
execute_command_listener(
device_client,
thermostat_1_component_name,
method_name="getMaxMinReport",
user_command_handler=max_min_handler,
create_user_response_handler=create_max_min_report_response,
 ),
execute_command_listener(
device_client,
thermostat_2_component_name,
method_name="getMaxMinReport",
user_command_handler=max_min_handler,
create_user_response_handler=create_max_min_report_response,
 ),
execute_property_listener(device_client),
 )

################################################
# Function to send telemetry every 8 seconds

asyncdefsend_telemetry():
print("Sending telemetry from various components")

whileTrue:
curr_temp_ext=random.randrange(10, 50)
THERMOSTAT_1.record(curr_temp_ext)

temperature_msg1= {"temperature": curr_temp_ext}
awaitsend_telemetry_from_temp_controller(
device_client, temperature_msg1, thermostat_1_component_name
 )

curr_temp_int=random.randrange(10, 50) # Current temperature in Celsius
THERMOSTAT_2.record(curr_temp_int)

temperature_msg2= {"temperature": curr_temp_int}

awaitsend_telemetry_from_temp_controller(
device_client, temperature_msg2, thermostat_2_component_name
 )

workingset_msg3= {"workingSet": random.randrange(1, 100)}
awaitsend_telemetry_from_temp_controller(device_client, workingset_msg3)

send_telemetry_task=asyncio.ensure_future(send_telemetry())

# Run the stdin listener in the event loop
loop=asyncio.get_running_loop()
user_finished=loop.run_in_executor(None, stdin_listener)
# # Wait for user to indicate they are done listening for method calls
awaituser_finished

ifnotlisteners.done():
listeners.set_result("DONE")

ifnotproperty_updates.done():
property_updates.set_result("DONE")

listeners.cancel()
property_updates.cancel()

send_telemetry_task.cancel()

# Finally, shut down the client
awaitdevice_client.shutdown()


#####################################################
# EXECUTE MAIN

if__name__=="__main__":
asyncio.run(main())