diff options
Diffstat (limited to 'sleekxmpp/plugins/xep_0323/sensordata.py')
| -rw-r--r-- | sleekxmpp/plugins/xep_0323/sensordata.py | 430 |
1 files changed, 215 insertions, 215 deletions
diff --git a/sleekxmpp/plugins/xep_0323/sensordata.py b/sleekxmpp/plugins/xep_0323/sensordata.py index 30c28504..2e2f2470 100644 --- a/sleekxmpp/plugins/xep_0323/sensordata.py +++ b/sleekxmpp/plugins/xep_0323/sensordata.py @@ -29,12 +29,12 @@ log = logging.getLogger(__name__) class XEP_0323(BasePlugin): """ - XEP-0323: IoT Sensor Data + XEP-0323: IoT Sensor Data This XEP provides the underlying architecture, basic operations and data structures for sensor data communication over XMPP networks. It includes - a hardware abstraction model, removing any technical detail implemented + a hardware abstraction model, removing any technical detail implemented in underlying technologies. Also see <http://xmpp.org/extensions/xep-0323.html> @@ -55,10 +55,10 @@ class XEP_0323(BasePlugin): Sensordata Event:Rejected -- Received a reject from sensor for a request Sensordata Event:Cancelled -- Received a cancel confirm from sensor Sensordata Event:Fields -- Received fields from sensor for a request - This may be triggered multiple times since + This may be triggered multiple times since the sensor can split up its response in multiple messages. - Sensordata Event:Failure -- Received a failure indication from sensor + Sensordata Event:Failure -- Received a failure indication from sensor for a request. Typically a comm timeout. Attributes: @@ -69,7 +69,7 @@ class XEP_0323(BasePlugin): relevant to a request's session. This dictionary is used both by the client and sensor side. On client side, seqnr is used as key, while on sensor side, a session_id is used - as key. This ensures that the two will not collide, so + as key. This ensures that the two will not collide, so one instance can be both client and sensor. Sensor side ----------- @@ -89,12 +89,12 @@ class XEP_0323(BasePlugin): Sensor side ----------- - register_node -- Register a sensor as available from this XMPP + register_node -- Register a sensor as available from this XMPP instance. Client side ----------- - request_data -- Initiates a request for data from one or more + request_data -- Initiates a request for data from one or more sensors. Non-blocking, a callback function will be called when data is available. @@ -102,7 +102,7 @@ class XEP_0323(BasePlugin): name = 'xep_0323' description = 'XEP-0323 Internet of Things - Sensor Data' - dependencies = set(['xep_0030']) + dependencies = set(['xep_0030']) stanza = stanza @@ -155,11 +155,11 @@ class XEP_0323(BasePlugin): self._handle_event_started)) # Server side dicts - self.nodes = {} - self.sessions = {} + self.nodes = {}; + self.sessions = {}; - self.last_seqnr = 0 - self.seqnr_lock = Lock() + self.last_seqnr = 0; + self.seqnr_lock = Lock(); ## For testning only self.test_authenticated_from = "" @@ -182,7 +182,7 @@ class XEP_0323(BasePlugin): def plugin_end(self): """ Stop the XEP-0323 plugin """ - self.sessions.clear() + self.sessions.clear(); self.xmpp.remove_handler('Sensordata Event:Req') self.xmpp.remove_handler('Sensordata Event:Accepted') self.xmpp.remove_handler('Sensordata Event:Rejected') @@ -198,9 +198,9 @@ class XEP_0323(BasePlugin): def register_node(self, nodeId, device, commTimeout, sourceId=None, cacheType=None): """ Register a sensor/device as available for serving of data through this XMPP - instance. + instance. - The device object may by any custom implementation to support + The device object may by any custom implementation to support specific devices, but it must implement the functions: has_field request_fields @@ -212,25 +212,25 @@ class XEP_0323(BasePlugin): commTimeout -- Time in seconds to wait between each callback from device during a data readout. Float. sourceId -- [optional] identifying the data source controlling the device - cacheType -- [optional] narrowing down the search to a specific kind of node + cacheType -- [optional] narrowing down the search to a specific kind of node """ - self.nodes[nodeId] = {"device": device, + self.nodes[nodeId] = {"device": device, "commTimeout": commTimeout, - "sourceId": sourceId, - "cacheType": cacheType} + "sourceId": sourceId, + "cacheType": cacheType}; def _set_authenticated(self, auth=''): """ Internal testing function """ - self.test_authenticated_from = auth + self.test_authenticated_from = auth; def _handle_event_req(self, iq): """ Event handler for reception of an Iq with req - this is a request. - Verifies that + Verifies that - all the requested nodes are available - - at least one of the requested fields is available from at least + - at least one of the requested fields is available from at least one of the nodes If the request passes verification, an accept response is sent, and @@ -238,42 +238,42 @@ class XEP_0323(BasePlugin): If the verification fails, a reject message is sent. """ - seqnr = iq['req']['seqnr'] - error_msg = '' - req_ok = True + seqnr = iq['req']['seqnr']; + error_msg = ''; + req_ok = True; # Authentication if len(self.test_authenticated_from) > 0 and not iq['from'] == self.test_authenticated_from: # Invalid authentication - req_ok = False - error_msg = "Access denied" + req_ok = False; + error_msg = "Access denied"; # Nodes - process_nodes = [] + process_nodes = []; if len(iq['req']['nodes']) > 0: for n in iq['req']['nodes']: if not n['nodeId'] in self.nodes: - req_ok = False - error_msg = "Invalid nodeId " + n['nodeId'] - process_nodes = [n['nodeId'] for n in iq['req']['nodes']] + req_ok = False; + error_msg = "Invalid nodeId " + n['nodeId']; + process_nodes = [n['nodeId'] for n in iq['req']['nodes']]; else: - process_nodes = self.nodes.keys() + process_nodes = self.nodes.keys(); # Fields - if we just find one we are happy, otherwise we reject - process_fields = [] + process_fields = []; if len(iq['req']['fields']) > 0: found = False for f in iq['req']['fields']: for node in self.nodes: if self.nodes[node]["device"].has_field(f['name']): - found = True - break + found = True; + break; if not found: - req_ok = False - error_msg = "Invalid field " + f['name'] - process_fields = [f['name'] for n in iq['req']['fields']] + req_ok = False; + error_msg = "Invalid field " + f['name']; + process_fields = [f['name'] for n in iq['req']['fields']]; - req_flags = iq['req']._get_flags() + req_flags = iq['req']._get_flags(); request_delay_sec = None if 'when' in req_flags: @@ -283,7 +283,7 @@ class XEP_0323(BasePlugin): try: dt = datetime.datetime.strptime(req_flags['when'], "%Y-%m-%dT%H:%M:%S") except ValueError: - req_ok = False + req_ok = False; error_msg = "Invalid datetime in 'when' flag, please use ISO format (i.e. 2013-04-05T15:00:03)." if not dt is None: @@ -292,30 +292,30 @@ class XEP_0323(BasePlugin): dtdiff = dt - dtnow request_delay_sec = dtdiff.seconds + dtdiff.days * 24 * 3600 if request_delay_sec <= 0: - req_ok = False - error_msg = "Invalid datetime in 'when' flag, cannot set a time in the past. Current time: " + dtnow.isoformat() + req_ok = False; + error_msg = "Invalid datetime in 'when' flag, cannot set a time in the past. Current time: " + dtnow.isoformat(); if req_ok: - session = self._new_session() - self.sessions[session] = {"from": iq['from'], "to": iq['to'], "seqnr": seqnr} - self.sessions[session]["commTimers"] = {} - self.sessions[session]["nodeDone"] = {} + session = self._new_session(); + self.sessions[session] = {"from": iq['from'], "to": iq['to'], "seqnr": seqnr}; + self.sessions[session]["commTimers"] = {}; + self.sessions[session]["nodeDone"] = {}; #print("added session: " + str(self.sessions)) - iq.reply() - iq['accepted']['seqnr'] = seqnr + iq.reply(); + iq['accepted']['seqnr'] = seqnr; if not request_delay_sec is None: iq['accepted']['queued'] = "true" - iq.send(block=False) + iq.send(block=False); - self.sessions[session]["node_list"] = process_nodes + self.sessions[session]["node_list"] = process_nodes; if not request_delay_sec is None: # Delay request to requested time timer = Timer(request_delay_sec, self._event_delayed_req, args=(session, process_fields, req_flags)) - self.sessions[session]["commTimers"]["delaytimer"] = timer - timer.start() + self.sessions[session]["commTimers"]["delaytimer"] = timer; + timer.start(); return if self.threaded: @@ -324,19 +324,19 @@ class XEP_0323(BasePlugin): tr_req.start() #print("started thread") else: - self._threaded_node_request(session, process_fields, req_flags) + self._threaded_node_request(session, process_fields, req_flags); else: - iq.reply() - iq['type'] = 'error' - iq['rejected']['seqnr'] = seqnr - iq['rejected']['error'] = error_msg - iq.send(block=False) + iq.reply(); + iq['type'] = 'error'; + iq['rejected']['seqnr'] = seqnr; + iq['rejected']['error'] = error_msg; + iq.send(block=False); def _threaded_node_request(self, session, process_fields, flags): - """ + """ Helper function to handle the device readouts in a separate thread. - + Arguments: session -- The request session id process_fields -- The fields to request from the devices @@ -344,41 +344,41 @@ class XEP_0323(BasePlugin): Formatted as a dictionary like { "flag name": "flag value" ... } """ for node in self.sessions[session]["node_list"]: - self.sessions[session]["nodeDone"][node] = False + self.sessions[session]["nodeDone"][node] = False; for node in self.sessions[session]["node_list"]: - timer = TimerReset(self.nodes[node]['commTimeout'], self._event_comm_timeout, args=(session, node)) - self.sessions[session]["commTimers"][node] = timer + timer = TimerReset(self.nodes[node]['commTimeout'], self._event_comm_timeout, args=(session, node)); + self.sessions[session]["commTimers"][node] = timer; #print("Starting timer " + str(timer) + ", timeout: " + str(self.nodes[node]['commTimeout'])) - timer.start() - self.nodes[node]['device'].request_fields(process_fields, flags=flags, session=session, callback=self._device_field_request_callback) + timer.start(); + self.nodes[node]['device'].request_fields(process_fields, flags=flags, session=session, callback=self._device_field_request_callback); def _event_comm_timeout(self, session, nodeId): - """ + """ Triggered if any of the readout operations timeout. Sends a failure message back to the client, stops communicating with the failing device. - + Arguments: session -- The request session id nodeId -- The id of the device which timed out """ - msg = self.xmpp.Message() - msg['from'] = self.sessions[session]['to'] - msg['to'] = self.sessions[session]['from'] - msg['failure']['seqnr'] = self.sessions[session]['seqnr'] - msg['failure']['error']['text'] = "Timeout" - msg['failure']['error']['nodeId'] = nodeId - msg['failure']['error']['timestamp'] = datetime.datetime.now().replace(microsecond=0).isoformat() + msg = self.xmpp.Message(); + msg['from'] = self.sessions[session]['to']; + msg['to'] = self.sessions[session]['from']; + msg['failure']['seqnr'] = self.sessions[session]['seqnr']; + msg['failure']['error']['text'] = "Timeout"; + msg['failure']['error']['nodeId'] = nodeId; + msg['failure']['error']['timestamp'] = datetime.datetime.now().replace(microsecond=0).isoformat(); # Drop communication with this device and check if we are done - self.sessions[session]["nodeDone"][nodeId] = True + self.sessions[session]["nodeDone"][nodeId] = True; if (self._all_nodes_done(session)): - msg['failure']['done'] = 'true' - msg.send() + msg['failure']['done'] = 'true'; + msg.send(); # The session is complete, delete it #print("del session " + session + " due to timeout") - del self.sessions[session] + del self.sessions[session]; def _event_delayed_req(self, session, process_fields, req_flags): """ @@ -390,47 +390,47 @@ class XEP_0323(BasePlugin): flags -- [optional] flags to pass to the devices, e.g. momentary Formatted as a dictionary like { "flag name": "flag value" ... } """ - msg = self.xmpp.Message() - msg['from'] = self.sessions[session]['to'] - msg['to'] = self.sessions[session]['from'] - msg['started']['seqnr'] = self.sessions[session]['seqnr'] - msg.send() + msg = self.xmpp.Message(); + msg['from'] = self.sessions[session]['to']; + msg['to'] = self.sessions[session]['from']; + msg['started']['seqnr'] = self.sessions[session]['seqnr']; + msg.send(); if self.threaded: tr_req = Thread(target=self._threaded_node_request, args=(session, process_fields, req_flags)) tr_req.start() else: - self._threaded_node_request(session, process_fields, req_flags) + self._threaded_node_request(session, process_fields, req_flags); def _all_nodes_done(self, session): - """ + """ Checks wheter all devices are done replying to the readout. - + Arguments: session -- The request session id """ for n in self.sessions[session]["nodeDone"]: if not self.sessions[session]["nodeDone"][n]: - return False - return True + return False; + return True; def _device_field_request_callback(self, session, nodeId, result, timestamp_block, error_msg=None): - """ + """ Callback function called by the devices when they have any additional data. - Composes a message with the data and sends it back to the client, and resets + Composes a message with the data and sends it back to the client, and resets the timeout timer for the device. - + Arguments: session -- The request session id nodeId -- The device id which initiated the callback result -- The current result status of the readout. Valid values are: "error" - Readout failed. "fields" - Contains readout data. - "done" - Indicates that the readout is complete. May contain + "done" - Indicates that the readout is complete. May contain readout data. - timestamp_block -- [optional] Only applies when result != "error" + timestamp_block -- [optional] Only applies when result != "error" The readout data. Structured as a dictionary: - { + { timestamp: timestamp for this datablock, fields: list of field dictionary (one per readout field). readout field dictionary format: @@ -442,7 +442,7 @@ class XEP_0323(BasePlugin): dataType: The datatype of the field. Only applies to type enum. flags: [optional] data classifier flags for the field, e.g. momentary Formatted as a dictionary like { "flag name": "flag value" ... } - } + } } error_msg -- [optional] Only applies when result == "error". Error details when a request failed. @@ -452,99 +452,99 @@ class XEP_0323(BasePlugin): return if result == "error": - self.sessions[session]["commTimers"][nodeId].cancel() + self.sessions[session]["commTimers"][nodeId].cancel(); - msg = self.xmpp.Message() - msg['from'] = self.sessions[session]['to'] - msg['to'] = self.sessions[session]['from'] - msg['failure']['seqnr'] = self.sessions[session]['seqnr'] - msg['failure']['error']['text'] = error_msg - msg['failure']['error']['nodeId'] = nodeId - msg['failure']['error']['timestamp'] = datetime.datetime.now().replace(microsecond=0).isoformat() + msg = self.xmpp.Message(); + msg['from'] = self.sessions[session]['to']; + msg['to'] = self.sessions[session]['from']; + msg['failure']['seqnr'] = self.sessions[session]['seqnr']; + msg['failure']['error']['text'] = error_msg; + msg['failure']['error']['nodeId'] = nodeId; + msg['failure']['error']['timestamp'] = datetime.datetime.now().replace(microsecond=0).isoformat(); # Drop communication with this device and check if we are done - self.sessions[session]["nodeDone"][nodeId] = True + self.sessions[session]["nodeDone"][nodeId] = True; if (self._all_nodes_done(session)): - msg['failure']['done'] = 'true' + msg['failure']['done'] = 'true'; # The session is complete, delete it # print("del session " + session + " due to error") - del self.sessions[session] - msg.send() + del self.sessions[session]; + msg.send(); else: - msg = self.xmpp.Message() - msg['from'] = self.sessions[session]['to'] - msg['to'] = self.sessions[session]['from'] - msg['fields']['seqnr'] = self.sessions[session]['seqnr'] + msg = self.xmpp.Message(); + msg['from'] = self.sessions[session]['to']; + msg['to'] = self.sessions[session]['from']; + msg['fields']['seqnr'] = self.sessions[session]['seqnr']; if timestamp_block is not None and len(timestamp_block) > 0: - node = msg['fields'].add_node(nodeId) - ts = node.add_timestamp(timestamp_block["timestamp"]) + node = msg['fields'].add_node(nodeId); + ts = node.add_timestamp(timestamp_block["timestamp"]); for f in timestamp_block["fields"]: - data = ts.add_data( typename=f['type'], - name=f['name'], - value=f['value'], - unit=f['unit'], - dataType=f['dataType'], - flags=f['flags']) + data = ts.add_data( typename=f['type'], + name=f['name'], + value=f['value'], + unit=f['unit'], + dataType=f['dataType'], + flags=f['flags']); if result == "done": - self.sessions[session]["commTimers"][nodeId].cancel() - self.sessions[session]["nodeDone"][nodeId] = True - msg['fields']['done'] = 'true' + self.sessions[session]["commTimers"][nodeId].cancel(); + self.sessions[session]["nodeDone"][nodeId] = True; + msg['fields']['done'] = 'true'; if (self._all_nodes_done(session)): # The session is complete, delete it # print("del session " + session + " due to complete") - del self.sessions[session] + del self.sessions[session]; else: # Restart comm timer - self.sessions[session]["commTimers"][nodeId].reset() + self.sessions[session]["commTimers"][nodeId].reset(); - msg.send() + msg.send(); def _handle_event_cancel(self, iq): - """ Received Iq with cancel - this is a cancel request. + """ Received Iq with cancel - this is a cancel request. Delete the session and confirm. """ - seqnr = iq['cancel']['seqnr'] + seqnr = iq['cancel']['seqnr']; # Find the session for s in self.sessions: if self.sessions[s]['from'] == iq['from'] and self.sessions[s]['to'] == iq['to'] and self.sessions[s]['seqnr'] == seqnr: # found it. Cancel all timers for n in self.sessions[s]["commTimers"]: - self.sessions[s]["commTimers"][n].cancel() + self.sessions[s]["commTimers"][n].cancel(); # Confirm - iq.reply() - iq['type'] = 'result' - iq['cancelled']['seqnr'] = seqnr - iq.send(block=False) - + iq.reply(); + iq['type'] = 'result'; + iq['cancelled']['seqnr'] = seqnr; + iq.send(block=False); + # Delete session del self.sessions[s] return # Could not find session, send reject - iq.reply() - iq['type'] = 'error' - iq['rejected']['seqnr'] = seqnr - iq['rejected']['error'] = "Cancel request received, no matching request is active." - iq.send(block=False) + iq.reply(); + iq['type'] = 'error'; + iq['rejected']['seqnr'] = seqnr; + iq['rejected']['error'] = "Cancel request received, no matching request is active."; + iq.send(block=False); - # ================================================================= + # ================================================================= # Client side (data retriever) API def request_data(self, from_jid, to_jid, callback, nodeIds=None, fields=None, flags=None): - """ + """ Called on the client side to initiade a data readout. Composes a message with the request and sends it to the device(s). Does not block, the callback will be called when data is available. - + Arguments: from_jid -- The jid of the requester to_jid -- The jid of the device(s) - callback -- The callback function to call when data is availble. - + callback -- The callback function to call when data is availble. + The callback function must support the following arguments: from_jid -- The jid of the responding device(s) @@ -565,7 +565,7 @@ class XEP_0323(BasePlugin): The timestamp of data in this callback. One callback will only contain data from one timestamp. fields -- [optional] Mandatory when result == "fields". - List of field dictionaries representing the readout data. + List of field dictionaries representing the readout data. Dictionary format: { typename: The field type (numeric, boolean, dateTime, timeSpan, string, enum) @@ -575,11 +575,11 @@ class XEP_0323(BasePlugin): dataType: The datatype of the field. Only applies to type enum. flags: [optional] data classifier flags for the field, e.g. momentary. Formatted as a dictionary like { "flag name": "flag value" ... } - } + } error_msg -- [optional] Mandatory when result == "rejected" or "failure". - Details about why the request is rejected or failed. - "rejected" means that the request is stopped, but note that the + Details about why the request is rejected or failed. + "rejected" means that the request is stopped, but note that the request will continue even after a "failure". "failure" only means that communication was stopped to that specific device, other device(s) (if any) will continue their readout. @@ -593,131 +593,131 @@ class XEP_0323(BasePlugin): session -- Session identifier. Client can use this as a reference to cancel the request. """ - iq = self.xmpp.Iq() - iq['from'] = from_jid - iq['to'] = to_jid - iq['type'] = "get" - seqnr = self._get_new_seqnr() - iq['id'] = seqnr - iq['req']['seqnr'] = seqnr + iq = self.xmpp.Iq(); + iq['from'] = from_jid; + iq['to'] = to_jid; + iq['type'] = "get"; + seqnr = self._get_new_seqnr(); + iq['id'] = seqnr; + iq['req']['seqnr'] = seqnr; if nodeIds is not None: for nodeId in nodeIds: - iq['req'].add_node(nodeId) + iq['req'].add_node(nodeId); if fields is not None: for field in fields: - iq['req'].add_field(field) + iq['req'].add_field(field); - iq['req']._set_flags(flags) + iq['req']._set_flags(flags); - self.sessions[seqnr] = {"from": iq['from'], "to": iq['to'], "seqnr": seqnr, "callback": callback} - iq.send(block=False) + self.sessions[seqnr] = {"from": iq['from'], "to": iq['to'], "seqnr": seqnr, "callback": callback}; + iq.send(block=False); - return seqnr + return seqnr; def cancel_request(self, session): - """ + """ Called on the client side to cancel a request for data readout. Composes a message with the cancellation and sends it to the device(s). - Does not block, the callback will be called when cancellation is + Does not block, the callback will be called when cancellation is confirmed. - + Arguments: session -- The session id of the request to cancel """ seqnr = session - iq = self.xmpp.Iq() + iq = self.xmpp.Iq(); iq['from'] = self.sessions[seqnr]['from'] - iq['to'] = self.sessions[seqnr]['to'] - iq['type'] = "get" - iq['id'] = seqnr - iq['cancel']['seqnr'] = seqnr - iq.send(block=False) + iq['to'] = self.sessions[seqnr]['to']; + iq['type'] = "get"; + iq['id'] = seqnr; + iq['cancel']['seqnr'] = seqnr; + iq.send(block=False); def _get_new_seqnr(self): """ Returns a unique sequence number (unique across threads) """ - self.seqnr_lock.acquire() - self.last_seqnr += 1 - self.seqnr_lock.release() - return str(self.last_seqnr) + self.seqnr_lock.acquire(); + self.last_seqnr = self.last_seqnr + 1; + self.seqnr_lock.release(); + return str(self.last_seqnr); def _handle_event_accepted(self, iq): """ Received Iq with accepted - request was accepted """ - seqnr = iq['accepted']['seqnr'] + seqnr = iq['accepted']['seqnr']; result = "accepted" if iq['accepted']['queued'] == 'true': result = "queued" - callback = self.sessions[seqnr]["callback"] - callback(from_jid=iq['from'], result=result) + callback = self.sessions[seqnr]["callback"]; + callback(from_jid=iq['from'], result=result); def _handle_event_rejected(self, iq): - """ Received Iq with rejected - this is a reject. + """ Received Iq with rejected - this is a reject. Delete the session. """ - seqnr = iq['rejected']['seqnr'] - callback = self.sessions[seqnr]["callback"] - callback(from_jid=iq['from'], result="rejected", error_msg=iq['rejected']['error']) + seqnr = iq['rejected']['seqnr']; + callback = self.sessions[seqnr]["callback"]; + callback(from_jid=iq['from'], result="rejected", error_msg=iq['rejected']['error']); # Session terminated - del self.sessions[seqnr] + del self.sessions[seqnr]; def _handle_event_cancelled(self, iq): - """ - Received Iq with cancelled - this is a cancel confirm. - Delete the session. + """ + Received Iq with cancelled - this is a cancel confirm. + Delete the session. """ #print("Got cancelled") - seqnr = iq['cancelled']['seqnr'] - callback = self.sessions[seqnr]["callback"] - callback(from_jid=iq['from'], result="cancelled") + seqnr = iq['cancelled']['seqnr']; + callback = self.sessions[seqnr]["callback"]; + callback(from_jid=iq['from'], result="cancelled"); # Session cancelled - del self.sessions[seqnr] + del self.sessions[seqnr]; def _handle_event_fields(self, msg): - """ + """ Received Msg with fields - this is a data reponse to a request. If this is the last data block, issue a "done" callback. """ - seqnr = msg['fields']['seqnr'] - callback = self.sessions[seqnr]["callback"] + seqnr = msg['fields']['seqnr']; + callback = self.sessions[seqnr]["callback"]; for node in msg['fields']['nodes']: for ts in node['timestamps']: - fields = [] + fields = []; for d in ts['datas']: - field_block = {} - field_block["name"] = d['name'] - field_block["typename"] = d._get_typename() - field_block["value"] = d['value'] + field_block = {}; + field_block["name"] = d['name']; + field_block["typename"] = d._get_typename(); + field_block["value"] = d['value']; if not d['unit'] == "": field_block["unit"] = d['unit']; if not d['dataType'] == "": field_block["dataType"] = d['dataType']; - flags = d._get_flags() + flags = d._get_flags(); if not len(flags) == 0: - field_block["flags"] = flags - fields.append(field_block) - - callback(from_jid=msg['from'], result="fields", nodeId=node['nodeId'], timestamp=ts['value'], fields=fields) + field_block["flags"] = flags; + fields.append(field_block); + callback(from_jid=msg['from'], result="fields", nodeId=node['nodeId'], timestamp=ts['value'], fields=fields); + if msg['fields']['done'] == "true": - callback(from_jid=msg['from'], result="done") + callback(from_jid=msg['from'], result="done"); # Session done - del self.sessions[seqnr] + del self.sessions[seqnr]; def _handle_event_failure(self, msg): - """ + """ Received Msg with failure - our request failed - Delete the session. + Delete the session. """ - seqnr = msg['failure']['seqnr'] - callback = self.sessions[seqnr]["callback"] - callback(from_jid=msg['from'], result="failure", nodeId=msg['failure']['error']['nodeId'], timestamp=msg['failure']['error']['timestamp'], error_msg=msg['failure']['error']['text']) + seqnr = msg['failure']['seqnr']; + callback = self.sessions[seqnr]["callback"]; + callback(from_jid=msg['from'], result="failure", nodeId=msg['failure']['error']['nodeId'], timestamp=msg['failure']['error']['timestamp'], error_msg=msg['failure']['error']['text']); # Session failed - del self.sessions[seqnr] + del self.sessions[seqnr]; def _handle_event_started(self, msg): + """ + Received Msg with started - our request was queued and is now started. """ - Received Msg with started - our request was queued and is now started. - """ - seqnr = msg['started']['seqnr'] - callback = self.sessions[seqnr]["callback"] - callback(from_jid=msg['from'], result="started") - + seqnr = msg['started']['seqnr']; + callback = self.sessions[seqnr]["callback"]; + callback(from_jid=msg['from'], result="started"); + |