def handle_request(self, topic, headers, message, matched):
            msg = jsonapi.loads(message[0])
            request_url = self.baseUrl

            # Identify if a zipcode or region/city was sent
            # Build request URL
            if 'zipcode' in msg:
                request_url += msg['zipcode'] + '.json'
            elif ('region' in msg) and ('city' in msg):
                self.requestUrl += msg['region'] + "/" + msg['city'] + ".json"
            else:
                _log.error('Invalid request, no zipcode or region/city in request')
                # TODO: notify requester of error

            # Request data
            (valid_data, observation) = self.request_data(request_url)

            # If data is valid, publish
            if valid_data:
                resp_headers = {}
                resp_headers[headers_mod.TO] = headers[headers_mod.REQUESTER_ID]
                resp_headers['agentID'] = agent_id
                resp_headers[headers_mod.FROM] = agent_id
                self.publish_all(observation, 'weather' + topic_delim + 'response', resp_headers)
            else:
                if observation == REQUESTS_EXHAUSTED:
                    _log.error('No requests avaliable')
                    # TODO: report error to client
                # Else, log that the data was invalid and report an error to the requester
                else:
                    _log.error('Weather API response was invalid')

 def handle_new(self, headers, message):
     '''Send schedule request response.'''
     self._log.info('handle new schedule request')
     requester = headers.get('requesterID')
     self.task_id = headers.get('taskID')
     # priority = headers.get('priority')
     requests = []
     try:
         requests = jsonapi.loads(message[0])
         requests = requests[0]
     except (ValueError, IndexError) as ex:
         self._log.info('error, message not in expected format (json)')
         self._log.error('bad request: {request}, {error}'
                         .format(request=requests, error=str(ex)))
         requests = []
     _, start, end = requests
     self.start_time = parser.parse(start, fuzzy=True)
     self.end_time = parser.parse(end, fuzzy=True)
     event = sched.Event(self.announce, args=[requests, requester])
     self.scheduled_event = event
     self.schedule(self.start_time, event)
     topic = topics.ACTUATOR_SCHEDULE_RESULT()
     headers = self.get_headers(requester, task_id=self.task_id)
     headers['type'] = SCHEDULE_ACTION_NEW
     self.publish_json(topic, headers,
                       {
                           'result': 'SUCCESS',
                           'data': 'NONE',
                           'info': 'NONE'
                       })

        def on_received_message(self, topic, headers, message, matched):
            _log.debug("Message received")
            _log.debug("MESSAGE: "+ jsonapi.dumps(message[0]))
            _log.debug("TOPIC: "+ topic)
            data = jsonapi.loads(message[0])

            results = app_instance.run(datetime.now(),data)
            self._process_results(results)

 def handle_set(self, topic, headers, message, match):
     '''Respond to ACTUATOR_SET topic.'''
     self._log.info('set actuator')
     point = match.group(1)
     _, _, _, point_name = point.rsplit('/', 4)
     requester = headers.get('requesterID')
     headers = self.get_headers(requester)
     value = jsonapi.loads(message[0])
     value_path = topic.replace('actuator/set', '')
     self.push_result_topic_pair(point_name, headers, value_path, value)

 def schedule_result(self, topic, headers, message, match):
     '''Actuator response (FAILURE, SUCESS).'''
     _log.debug('Actuator Response')
     msg = jsonapi.loads(message[0])
     msg = msg['result']
     _log.debug('Schedule Device ACCESS')
     if self.keys:
         if msg == "SUCCESS":
             self.command_equip()
         elif msg == "FAILURE":
             _log.debug('Auto-correction of device failed.')

 def on_received_message(self, topic, headers, message, matched):
     '''Subscribe to device data and convert data to correct type for
     the driven application.
     '''
     _log.debug("Message received")
     _log.debug("MESSAGE: " + jsonapi.dumps(message[0]))
     _log.debug("TOPIC: " + topic)
     data = jsonapi.loads(message[0])
     
     #TODO: grab the time from the header if it's there or use now if not
     self.received_input_datetime = datetime.utcnow()
     results = app_instance.run(self.received_input_datetime, data)
     self._process_results(results)

        def on_received_message(self, topic, headers, message, matched):
            '''Subscribe to device data and convert data to correct type for
            the driven application.
            '''
            _log.debug("Message received")
            _log.debug("MESSAGE: " + jsonapi.dumps(message[0]))
            _log.debug("TOPIC: " + topic)
            data = jsonapi.loads(message[0])
            if not converter.initialized and \
                config.get('conversion_map') is not None:
                converter.setup_conversion_map(config.get('conversion_map'),
                                               data.keys())
            data = converter.process_row(data)

            results = app_instance.run(datetime.now(), data)
            self._process_results(results)

        def handle_new(self, headers, message, now):
            requester = headers.get("requesterID")
            taskID = headers.get("taskID")
            priority = headers.get("priority")

            _log.debug(
                "Got new schedule request: {headers}, {message}".format(headers=str(headers), message=str(message))
            )

            try:
                requests = jsonapi.loads(message[0])
            except (ValueError, IndexError) as ex:
                # Could be ValueError of JSONDecodeError depending
                # on if simplesjson was used.  JSONDecodeError
                # inherits from ValueError

                # We let the schedule manager tell us this is a bad request.
                _log.error("bad request: {request}, {error}".format(request=requests, error=str(ex)))
                requests = []

            result = self._schedule_manager.request_slots(requester, taskID, requests, priority, now)
            success = SCHEDULE_RESPONSE_SUCCESS if result.success else SCHEDULE_RESPONSE_FAILURE

            # If we are successful we do something else with the real result data
            data = result.data if not result.success else {}

            topic = topics.ACTUATOR_SCHEDULE_RESULT()
            headers = self.get_headers(requester, task_id=taskID)
            headers["type"] = SCHEDULE_ACTION_NEW
            self.publish_json(topic, headers, {"result": success, "data": data, "info": result.info_string})

            # Dealing with success and other first world problems.
            if result.success:
                self.update_device_state_and_schedule(now)
                for preempted_task in result.data:
                    topic = topics.ACTUATOR_SCHEDULE_RESULT()
                    headers = self.get_headers(preempted_task[0], task_id=preempted_task[1])
                    headers["type"] = SCHEDULE_ACTION_CANCEL
                    self.publish_json(
                        topic,
                        headers,
                        {
                            "result": SCHEDULE_CANCEL_PREEMPTED,
                            "info": "",
                            "data": {"agentID": requester, "taskID": taskID},
                        },
                    )

        def handle_set(self, topic, headers, message, match):
            _log.debug('handle_set: {topic},{headers}, {message}'.
                       format(topic=topic, headers=headers, message=message))
            point = match.group(1)
            collection_tokens, point_name = point.rsplit('/', 1)
            requester = headers.get('requesterID')
            headers = self.get_headers(requester)
            if not message:
                error = {'type': 'ValueError', 'value': 'missing argument'}
                _log.debug('ValueError: '+str(error))
                self.push_result_topic_pair(ERROR_RESPONSE_PREFIX,
                                            point, headers, error)
                return
            else:
                try:
                    message = jsonapi.loads(message[0])
                    if isinstance(message, bool):
                        message = int(message)
                except ValueError as ex:
                    # Could be ValueError of JSONDecodeError depending
                    # on if simplesjson was used.  JSONDecodeError
                    # inherits from ValueError
                    _log.debug('ValueError: '+message)
                    error = {'type': 'ValueError', 'value': str(ex)}
                    self.push_result_topic_pair(ERROR_RESPONSE_PREFIX,
                                                point, headers, error)
                    return

            if self.check_lock(collection_tokens, requester):
                request_url = '/'.join([url, collection_tokens,
                                        ACTUATOR_COLLECTION, point_name])
                payload = {'state': str(message)}
                try:
                    r = requests.put(request_url, params=payload, timeout=connection_timeout)
                    self.process_smap_request_result(r, point, requester)
                except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as ex:
                    
                    error = {'type': ex.__class__.__name__, 'value': str(ex)}
                    self.push_result_topic_pair(ERROR_RESPONSE_PREFIX,
                                                point, headers, error)
                    _log.debug('ConnectionError: '+str(error))
            else:
                error = {'type': 'LockError',
                         'value': 'does not have this lock'}
                _log.debug('LockError: '+str(error))
                self.push_result_topic_pair(ERROR_RESPONSE_PREFIX,
                                            point, headers, error)

 def on_set_error(self, topic, headers, message, match):
     '''Setting of point on device failed, log failure message.'''
     _log.debug('Set ERROR')
     msg = jsonapi.loads(message[0])
     msg = msg['type']
     _log.debug('Actuator Error: ({}, {}, {})'.
                format(msg,
                       self.current_key,
                       self.commands[self.current_key]))
     self.keys.remove(self.current_key)
     if self.keys:
         self.command_equip()
     else:
         headers = {
             'type':  'CANCEL_SCHEDULE',
             'requesterID': agent_id,
             'taskID': actuator_id
             }
         self.publish_json(topics.ACTUATOR_SCHEDULE_REQUEST(),
                           headers, {})
         self.keys = None

        def handle_request(self, peer, sender, bus, topic, headers, message):
            
            if sender == 'pubsub.compat':
                msg = jsonapi.loads(message[0])
            else:
                msg = message
                
            request_url = self.baseUrl

            # Identify if a zipcode or region/city was sent
            # Build request URL
            if 'zipcode' in msg:
                request_url = self.build_url_with_zipcode(msg['zipcode'])
            elif ('region' in msg) and ('city' in msg):
                request_url = self.build_url_with_city(msg['region'], msg['city'])
            else:
                _log.error('Invalid request, no zipcode '
                           'or region/city in request')
                # TODO: notify requester of error

            # Request data
            (valid_data, observation) = self.request_data(request_url)

            # If data is valid, publish
            if valid_data:
                resp_headers = {}
                resp_headers[headers_mod.TO] = headers[headers_mod.REQUESTER_ID]
                resp_headers['agentID'] = agent_id
                resp_headers[headers_mod.FROM] = agent_id
                _topic = 'weather' + TOPIC_DELIM + 'response'
                self.publish_all(observation, _topic, resp_headers)
            else:
                if observation == REQUESTS_EXHAUSTED:
                    _log.error('No requests avaliable')
                    # TODO: report error to client
                # Else, log that the data was invalid and report an error
                # to the requester
                else:
                    _log.error('Weather API response was invalid')

        def handle_request(self, peer, sender, bus, topic, headers, message):

            if sender == "pubsub.compat":
                msg = jsonapi.loads(message[0])
            else:
                msg = message

            request_url = self.baseUrl

            # Identify if a zipcode or region/city was sent
            # Build request URL
            if "zipcode" in msg:
                request_url += msg["zipcode"] + ".json"
            elif ("region" in msg) and ("city" in msg):
                self.requestUrl += msg["region"] + "/" + msg["city"] + ".json"
            else:
                _log.error("Invalid request, no zipcode " "or region/city in request")
                # TODO: notify requester of error

            # Request data
            (valid_data, observation) = self.request_data(request_url)

            # If data is valid, publish
            if valid_data:
                resp_headers = {}
                resp_headers[headers_mod.TO] = headers[headers_mod.REQUESTER_ID]
                resp_headers["agentID"] = agent_id
                resp_headers[headers_mod.FROM] = agent_id
                _topic = "weather" + TOPIC_DELIM + "response"
                self.publish_all(observation, _topic, resp_headers)
            else:
                if observation == REQUESTS_EXHAUSTED:
                    _log.error("No requests avaliable")
                    # TODO: report error to client
                # Else, log that the data was invalid and report an error
                # to the requester
                else:
                    _log.error("Weather API response was invalid")

        def on_rec_analysis_message(self, topic, headers, message, matched):
            # Do the analysis based upon the data passed (the old code).
            # print self._subdevice_values, self._device_values
            obj = jsonapi.loads(message[0])
            dev_list = topic.split('/')
            device_or_subdevice = dev_list[-2]
            device_id = [dev for dev in self._master_devices
                         if dev == device_or_subdevice]
            subdevice_id = [dev for dev in self._master_subdevices
                            if dev == device_or_subdevice]
            if not device_id and not subdevice_id:
                return
            if isinstance(device_or_subdevice, unicode):
                device_or_subdevice = (
                    device_or_subdevice.decode('utf-8').encode('ascii')
                )

            def agg_subdevice(obj):
                sub_obj = {}
                for key, value in obj.items():
                    sub_key = ''.join([key, '_', device_or_subdevice])
                    sub_obj[sub_key] = value
                if len(dev_list) > 5:
                    self._subdevice_values.update(sub_obj)
                    self._needed_subdevices.remove(device_or_subdevice)
                else:
                    self._device_values.update(sub_obj)
                    self._needed_devices.remove(device_or_subdevice)
                return
            # The below if statement is used to distinguish between unit/all
            # and unit/sub-device/all
            if (device_or_subdevice not in self._needed_devices and
                    device_or_subdevice not in self._needed_subdevices):
                _log.error("Warning device values already present, "
                           "reinitializing")
                self._initialize_devices()
            agg_subdevice(obj)
            if self._should_run_now():
                field_names = {}
                self._device_values.update(self._subdevice_values)
                for k, v in self._device_values.items():
                    field_names[k.lower() if isinstance(k, str) else k] = v
                if not converter.initialized and \
                        conv_map is not None:
                    converter.setup_conversion_map(
                        map_names,
                        field_names
                    )
                obj = converter.process_row(field_names)
                results = app_instance.run(datetime.now(),
                                           obj)
                self.received_input_datetime = datetime.utcnow()
                # results = app_instance.run(
                # dateutil.parser.parse(self._subdevice_values['Timestamp'],
                #                       fuzzy=True), self._subdevice_values)
                self._process_results(results)
                self._initialize_devices()
            else:
                needed = deepcopy(self._needed_devices)
                needed.extend(self._needed_subdevices)
                _log.info("Still need {} before running."
                          .format(needed))

        def handle_full_request(self, topic, headers, message, matched):
            # Path is part of topic.
            path = topic[len(topics.BASE_ARCHIVER_FULL_REQUEST):]
            result_layout = headers.get('ResultLayout', 'HIERARCHICAL')

            # Range is message.  It will either be "start"-"end" or 1h, 1d,
            # etc... from now
            range_str = message[0]
            source = headers.get('SourceName', source_name)

            # Find UUID for path
            payload = ('select * where Metadata/SourceName="{}" '
                       'and Path~"{}"'.format(source, path))

            done = False
            retries = 0
            while not done and retries <= 5:
                # TODO: Need to do some error handling here!
                try:
                    r = requests.post(archiver_url, data=payload)
                    if r.status_code == 200:
                        # Data should be a list of dictionaries at this point in time
                        uuid_list = jsonapi.loads(r.text)
                        done = True
                    else:
                        print str(retries) + ": " + str(r.status_code) + ": " + payload
                        retries += 1
                except ValueError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                except ConnectionError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                    #Can get a 503 network busy
                    #TODO: Respond with error

            # TODO: Need to do some error handling here!

            # Data should be a list of dictionaries at this point in time
            if not uuid_list:
                # TODO: log this error
                return

            timeseries = {}
            hierarchichal = {}

            payload_template = "select data in {} where {{}}".format(range_str)
            if len(uuid_list) == 0:
                return

            uuid_clause = "uuid='{}'".format(uuid_list[0]['uuid'])
            for stream in uuid_list[1:]:
                uuid_clause += (" or uuid='{}'".format(stream['uuid']))
            payload = payload_template.format(uuid_clause)
#
            # Request data and store Readings in timeseries[path]
            full_data = None
            tries = 0
            done = False
            while not done and retries <= 5:
            # TODO: Need to do some error handling here!
                try:
                    r = requests.post(archiver_url, data=payload)
                    if 'Syntax error' in r.text:
                        # TODO Log this error
                        self.publish(topics.BASE_ARCHIVER_RESPONSE + path, None, 'Syntax error in date range')
                        return

            # Request data for UUID in range
            #[{"uuid": "5b94d5ed-1e1d-51cf-a6d8-afae5c055292", "Readings": [[1368750281000.0, 75.5], [1368750341000.0, 75.5], ...
                    done = True
                except ValueError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                except ConnectionError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1

            if 'Syntax error' in r.text:
                # TODO Log this error
                self.publish(topics.BASE_ARCHIVER_RESPONSE + path,
                             None, 'Syntax error in date range')
                return

            # Request data for UUID in range
            #[{"uuid": "5b94d5ed-1e1d-51cf-a6d8-afae5c055292",
            # "Readings": [[1368750281000.0, 75.5], [1368750341000.0, 75.5], ...

            full_data = jsonapi.loads(r.text)
            
            reading_dict = {}
            
            for readings in full_data:
                reading_dict[readings['uuid']] = readings.get('Readings', [])
            
            for stream in uuid_list:
#                 uuid_clause += (" or uuid='{}'".format(stream['uuid']))
                path = stream['Path']
                print stream
                (hierarchichal, timeseries) = build_paths(path, hierarchichal, timeseries)
#.........这里部分代码省略.........

        def handle_request(self, topic, headers, message, matched):
            # Path is part of topic.
            path = topic[len(topics.BASE_ARCHIVER_REQUEST):]

            # Range is message.  It will either be "start"-"end" or 1h, 1d,
            # etc... from now
            range_str = message[0]
            source = headers.get('SourceName', source_name)

            # Find UUID for path
            payload = ('select uuid where Metadata/SourceName="{}" '
                       'and Path="{}"'.format(source, path))

            done = False
            retries = 0
            uuid_list = []
            while not done and retries <= 5:
                # TODO: Need to do some error handling here!
                try:
                    r = requests.post(archiver_url, data=payload)
                    if r.status_code == 200:
                        # Data should be a list of dictionaries at this point in time
                        uuid_list = jsonapi.loads(r.text)
                        done = True
                    else:
                        print str(retries) + ": " + str(r.status_code) + ": " + payload
                        retries += 1
                except ValueError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                except ConnectionError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                    #Can get a 503 network busy
                    #TODO: Respond with error

            # TODO: Need to do some error handling here!

            # Data should be a list of dictionaries at this point in time
            if not uuid_list:
                # TODO: log this error
                return
            payload_template = "select data in {} where {{}}".format(range_str)

            uuid_clause = "uuid='{}'".format(uuid_list[0]['uuid'])
            for stream in uuid_list[1:]:
                uuid_clause.append(" or uuid='{}'".format(stream['uuid']))
            payload = payload_template.format(uuid_clause)

            full_data = None
            tries = 0
            done = False
            while not done and retries <= 5:
            # TODO: Need to do some error handling here!
                try:
                    r = requests.post(archiver_url, data=payload)
                    if 'Syntax error' in r.text:
                        # TODO Log this error
                        self.publish(topics.BASE_ARCHIVER_RESPONSE + path, None, 'Syntax error in date range')
                        return

            # Request data for UUID in range
            #[{"uuid": "5b94d5ed-1e1d-51cf-a6d8-afae5c055292", "Readings": [[1368750281000.0, 75.5], [1368750341000.0, 75.5], ...
                    done = True
                except ValueError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1
                except ConnectionError as e:
                    print str(retries) + ": " + str(e) + ": " + payload
                    retries += 1

            if 'Syntax error' in r.text:
                # TODO Log this error
                self.publish(topics.BASE_ARCHIVER_RESPONSE + path,
                             None, 'Syntax error in date range')
                return

            # Request data for UUID in range
            #[{"uuid": "5b94d5ed-1e1d-51cf-a6d8-afae5c055292",
            # "Readings": [[1368750281000.0, 75.5], [1368750341000.0, 75.5], ...
            full_data = jsonapi.loads(r.text)
            data = full_data[0].get('Readings', [])
            pub_headers = {headers_mod.FROM: 'ArchiverAgent',
                           headers_mod.TO: headers[headers_mod.FROM] if headers_mod.FROM in headers else 'Unknown'}
            if data > 0:
                # There was data for this stream in the specified range.
                # Convert data to json and publish
                self.publish_json(topics.BASE_ARCHIVER_RESPONSE + path, pub_headers, data)

 def on_new_data(self, topic, headers, message, match):
     data = jsonapi.loads(message[0])
     mixed_air_temperature=data
     print(mixed_air_temperature)