@@ -62,7 +62,7 @@ def test_setup():
6262 # print(cmd_resp)
6363
6464
65- def test_subscribe_and_command ():
65+ def subscribe_and_command ():
6666 mqtt_client = MQTTCommClient (url = 'localhost' )
6767
6868 control_streams = ControlChannels .list_all_control_streams (server_url ).json ()
@@ -74,12 +74,17 @@ def on_message_command(client, userdata, msg):
7474 print ("Received Command" )
7575 print (f'{ msg .payload .decode ("utf-8" )} )
7676 control_stream_id = control_id
77+
78+ payload = msg .payload .decode ("utf-8" )
79+ p_dict = json .loads (payload )
80+ id = p_dict ["id" ]
81+
7782 resp = {
78- 'id' : '*******' ,
83+ 'id' : id ,
7984 'command@id' : control_stream_id ,
8085 'statusCode' : 'COMPLETED'
8186 }
82- # client.publish(f'/api/controls/{control_stream_id}/status', payload=json.dumps(resp), qos=1)
87+ client .publish (f'/api/controls/{ control_stream_id } , payload = json .dumps (resp ), qos = 1 )
8388
8489 def on_message_all (client , userdata , msg ):
8590 print (f'\n Received Message:{ msg } )
@@ -93,36 +98,105 @@ def on_message_all(client, userdata, msg):
9398 mqtt_client .start ()
9499
95100 time .sleep (2 )
96- command_json = CommandJSON (control_id = control_streams ["items" ][0 ]["id" ],
97- issue_time = datetime .now ().isoformat () + 'Z' ,
98- params = {"timestamp" : datetime .now ().timestamp () * 1000 , "testcount" : 1 })
99101
100- print (f'Issuing Command: { command_json .model_dump_json (exclude_none = True , by_alias = True )} )
101- cmd_resp = Commands .send_commands_to_specific_control_stream (server_url , control_streams ["items" ][0 ]["id" ],
102- command_json .model_dump_json (exclude_none = True ,
103- by_alias = True ),
104- headers = json_headers )
105- # try issuing a command from the MQTT client
102+ # print(f'Issuing Command: {command_json.model_dump_json(exclude_none=True, by_alias=True)}')
103+ # cmd_resp = Commands.send_commands_to_specific_control_stream(server_url, control_streams["items"][0]["id"],
104+ # command_json.model_dump_json(exclude_none=True,
105+ # by_alias=True),
106+ # headers=json_headers)
107+ # # try issuing a command from the MQTT client
106108 # mqtt_client.publish(f'/api/controls/{control_id}/commands', command_json.model_dump_json(exclude_none=True,
107109 # by_alias=True),
108110 # 1)
109111 # print(f'\n*****Command Response: {cmd_resp}*****')
110- status_resp = {
111- 'id' : '*******' ,
112- 'command@id' : "unknown" ,
113- 'statusCode' : 'COMPLETED'
114- }
112+ # status_resp = {
113+ # 'id': '*******',
114+ # 'command@id': "unknown",
115+ # 'statusCode': 'COMPLETED'
116+ # }
115117 # Commands.add_command_status_reports(server_url, "0", json.dumps(status_resp))
116118
117119
118- def test_command_dahua ():
120+ def command_dahua ():
119121 system_id = "tstk16o31es4m"
120122 control_stream_id = "k08p16h6k4a6c"
121123 control_input = CommandJSON (control_id = control_stream_id , issue_time = datetime .now ().isoformat () + 'Z' ,
122124 params = {"pan" : 180 })
123125 print (f'Issuing Command: { control_input .model_dump_json (exclude_none = True , by_alias = True )} )
124126 cmd_resp = Commands .send_commands_to_specific_control_stream (server_url , control_stream_id ,
125127 control_input .model_dump_json (exclude_none = True ,
126- by_alias = True ),
128+ by_alias = True ),
127129 headers = json_headers )
128- print (f'\n *****Command Response: { cmd_resp } )
130+ print (f'\n *****Command Response: { cmd_resp } )
131+
132+
133+ def create_status_listener_client (control_id : str ):
134+ def on_connect (client , userdata , flags , rc , props = None ):
135+ print (f"Status Updater Client connected with result code { rc } )
136+
137+ def on_command (client , userdata , msg ):
138+ payload = msg .payload .decode ("utf-8" )
139+ print (f"Received Command: { payload } )
140+ p_dict = json .loads (payload )
141+ id = p_dict ["id" ]
142+
143+ resp = {
144+ 'id' : id ,
145+ # 'command@id': control_id,
146+ 'statusCode' : 'COMPLETED'
147+ }
148+
149+ print (f'Issuing Status: { resp } )
150+
151+ client .publish (f'/api/controls/{ control_id } , payload = json .dumps (resp ), qos = 1 )
152+
153+ mqtt_client = MQTTCommClient (url = 'localhost' )
154+ mqtt_client .set_on_connect (on_connect )
155+ mqtt_client .connect (keepalive = 60 )
156+ mqtt_client .set_on_message_callback (f'/api/controls/{ control_id } , on_command )
157+ mqtt_client .subscribe (f'/api/controls/{ control_id } )
158+ mqtt_client .start ()
159+
160+ return mqtt_client
161+
162+
163+ def create_command_client (control_id : str ):
164+ def on_connect (client , userdata , flags , rc , props = None ):
165+ print (f"Command Client connected with result code { rc } )
166+
167+ def on_status (client , userdata , msg ):
168+ print ("" )
169+ print (f"Received Status: { msg .payload .decode ('utf-8' )} )
170+ print ("" )
171+
172+ mqtt_client = MQTTCommClient (url = 'localhost' )
173+ mqtt_client .set_on_connect (on_connect )
174+ mqtt_client .connect ()
175+ mqtt_client .set_on_message_callback (f'/api/controls/o1l72d8md66a0/status' , on_status )
176+ mqtt_client .subscribe (f'/api/controls/o1l72d8md66a0/status' )
177+ mqtt_client .start ()
178+
179+ return mqtt_client
180+
181+
182+ def create_test_command (control_id ):
183+ command_json = CommandJSON (control_id = control_id ,
184+ issue_time = datetime .now ().isoformat () + 'Z' ,
185+ params = {"timestamp" : datetime .now ().timestamp () * 1000 , "testcount" : 1 })
186+
187+ return command_json .model_dump_json (exclude_none = True , by_alias = True )
188+
189+
190+ def test_command_with_status_updates ():
191+ control_streams = ControlChannels .list_all_control_streams (server_url ).json ()
192+ control_id = control_streams ["items" ][0 ]["id" ]
193+
194+ # Create a Command Client For Status Updates
195+ status_updater = create_status_listener_client (control_id )
196+ command_sender = create_command_client (control_id )
197+
198+ # Wait for a bit
199+ time .sleep (1 )
200+
201+ # Send a Command
202+ command_sender .publish (f'/api/controls/{ control_id } , create_test_command (control_id ), 0 )
0 commit comments