def insert_aggregate(self, agg_topic_id, agg_type, period, ts,
                         data, topic_ids):
        """
        Insert aggregate data collected for a specific  time period into
        database. Data is inserted into <agg_type>_<period> table

        :param agg_topic_id: topic id
        :param agg_type: type of aggregation
        :param period: time period of aggregation
        :param ts: end time of aggregation period (not inclusive)
        :param data: computed aggregate
        :param topic_ids: topic ids or topic ids for which aggregate was
                          computed
        :return: True if execution was successful, False otherwise
        """

        if not self.__connect():
            print("connect to database failed.......")
            return False
        table_name = agg_type + '_' + period
        _log.debug("Inserting aggregate: {} {} {} {} into table {}".format(
            ts, agg_topic_id, jsonapi.dumps(data), str(topic_ids), table_name))
        self.__cursor.execute(
            self.insert_aggregate_stmt(table_name),
            (ts, agg_topic_id, jsonapi.dumps(data), str(topic_ids)))
        self.commit()
        return True

    def capture_analysis_data(self, peer, sender, bus, topic, headers, message):
        '''Capture device data and submit it to be published by a historian.
        
        Filter out only the */all topics for publishing to the historian.
        '''
        
        if topic.endswith("/all") or '/all/' in topic:
#             _log.debug("Unmatched topic: {}".format(topic))
            return
        
        # Because of the above if we know that all is in the topic so
        # we strip it off to get the base device
#         parts = topic.split('/')
        parts = topic.split('/')
        device = '/'.join(parts[1:-1]) #'/'.join(reversed(parts[2:]))
        
        _log.debug("found topic {}".format(topic))
        
        if topic.endswith('Timestamp'):
            pass
        
        try:
            value = float(message[0])
        except:
            value = message[0]
        # Because message is a single point we need to have the point in the
        # message
        real_message = {parts[-1]: value}
        real_message = [jsonapi.dumps(real_message), jsonapi.dumps({})]
        
        self.capture_data(peer, sender, bus, topic, headers, real_message, device)

    def update_status(self, status, context=None):
        """
        Updates the internal state of the `Status` object.

        This method will throw errors if the context is not serializable or
        if the status parameter is not within the ACCEPTABLE_STATUS tuple.

        :param status:
        :param context:
        :return:
        """
        if status not in ACCEPTABLE_STATUS:
            raise ValueError('Invalid status value {}'.format(status))
        try:
            jsonapi.dumps(context)
        except TypeError:
            raise ValueError('Context must be JSON serializable.')

        status_changed = status != self._status
        self._status = status
        self._context = context
        self._last_updated = format_timestamp(get_aware_utc_now())

        if status_changed and self._status_changed_callback:
            print(self._status_changed_callback())

	def send(self, envelope, msg, error=False):
		if error:
			msg = jsonapi.dumps({'error': msg})
		else:
			# FIXME: exception handling should be better done
			# but there are too many json libraries out there
			try: msg = jsonapi.dumps({'result': msg})
			except Exception:
				msg = jsonapi.dumps({'proxy': repr(msg)})

		envelope.append(msg)
		return self.sock.send_multipart(envelope)

 def process_multiframe_response(self, env, multipart_message):
     timestamp, command, args, kwargs, result, error_str, error = map(
             self.deserialize_frame, multipart_message
     )
     data = {'timestamp': timestamp, 'command': command, 'args': args,
             'kwargs': kwargs, 'result': result, 'error_str': error_str,
             'error': error}
     data = self._filter_multiframe_response_fields(data)
     try:
         json_data = jsonapi.dumps(data)
     except Exception, e:
         import traceback
         data = {'result': None, 'error_str': traceback.format_exc()}
         json_data = jsonapi.dumps(data)

 def post(self):
     km = self.application.kernel_manager
     notebook_id = self.get_argument('notebook', default=None)
     kernel_id = km.start_kernel(notebook_id)
     data = {'ws_url':self.ws_url,'kernel_id':kernel_id}
     self.set_header('Location', '/'+kernel_id)
     self.finish(jsonapi.dumps(data))

def groupofatoms(groupofatoms):
    '''
    Serialises a MMTK protein.
    '''
    # Prepare the data and configuration.
    config = None
    universe = groupofatoms.universe()
    if universe is not None:
        config = universe.contiguousObjectConfiguration([groupofatoms])

    # Serialise the data.
    buffer = StringIO()
    file = PDBOutputFile(buffer)
    file.write(groupofatoms, config)

    # Retrieve the content.
    pdb = buffer.getvalue()
    file.close()

    # Store it in the json object that is sent to javascript.
    result = {'pdb': pdb}

    # Specify the javascript handler.
    result['handler'] = 'GroupOfAtoms'

    return jsonapi.dumps(result)

    def __init__(self, init_state, host):
        self.state = init_state
        
        context = zmq.Context()
        
        self.publisher = context.socket(zmq.PUB)
        self.publisher.bind('tcp://*:{}'.format(IO.STATE))
        
        self.event = context.socket(zmq.PUB)
        self.event.bind('tcp://*:{}'.format(IO.EVENT))

        snapshot = context.socket(zmq.ROUTER)
        snapshot.bind('tcp://*:{}'.format(IO.SNAPSHOT))

        self.association = context.socket(zmq.REQ)
        self.association.connect('tcp://{}:{}'.format(host, IO.ASSOCIATION))

        incoming = context.socket(zmq.PULL)
        incoming.bind('tcp://*:{}'.format(IO.EXTERNAL))

        poller = zmq.Poller()
        poller.register(incoming, zmq.POLLIN)
        poller.register(snapshot, zmq.POLLIN)

        while True:
            events = dict(poller.poll())
            
            if incoming in events:
                self.parse(incoming.recv_json())
                 
            if snapshot in events:
                address, _, message = snapshot.recv_multipart()
                snapshot.send_multipart([ address, 
                                          b'',
                                          dumps(self.state) ])

 def on_recv(self, msg):
     msg = self.session.feed_identities(msg)[1]
     msg = self.session.unserialize(msg)
     msg_id = msg["parent_header"]["msg_id"]
     kc = self.waiting.pop(msg_id)
     del msg["header"]["date"]
     kc.send("complete/shell," + jsonapi.dumps(msg))

def test_store_list_get_configuration(vc_vcp_platforms):
    vc, vcp = vc_vcp_platforms

    data = dict(
        bim=50,
        baz="foo",
        bar="lambda"
    )
    str_data = jsonapi.dumps(data)
    identity = "foo.bar"
    config_name = "fuzzywidgets"
    api = APITester(vc.jsonrpc_endpoint)

    platforms = api.list_platforms()
    platform_uuid = platforms[0]["uuid"]

    resp = api.store_agent_config(platform_uuid, identity, config_name,
                                  str_data)
    assert resp is None

    resp = api.list_agent_configs(platform_uuid, identity)
    assert config_name == resp[0]

    resp = api.get_agent_config(platform_uuid, identity, config_name)
    assert str_data == resp

def jsonify(item):
	"""
	Serializes an object into *optimized* JSON (meaning no whitespace is used).

	"""

	return jsonapi.dumps(item, separators = (",", ":"))

def do_rpc(method, params=None, auth_token=None, rpc_root=None):
    """ A utility method for calling json rpc based funnctions.

    :param method: The method to call
    :param params: the parameters to the method
    :param auth_token: A token if the user has one.
    :param rpc_root: Root of jsonrpc api.
    :return: The result of the rpc method.
    """

    assert rpc_root, "Must pass a jsonrpc url in to the function."

    json_package = {
        'jsonrpc': '2.0',
        'id': '2503402',
        'method': method,
    }

    if auth_token:
        json_package['authorization'] = auth_token

    if params:
        json_package['params'] = params

    data = jsonapi.dumps(json_package)

    return requests.post(rpc_root, data=data)

    def routerRecv(self, message):
        """
        message = [ ... , request, image/blank]

        request = {'timestamp': timestamp, 
                   'task': 'detection'/'recognition'/'tracking',
                   'parameters': (...)}
        """
        request = loads(message[-2])

        if request["task"] == "detection":
            self.logger.debug("start detection")
            with open("image.jpg", "wb") as f:
                f.write(message[-1])
            sleep = random.randint(1, 2)  # detection
            time.sleep(sleep)
            message[-2] = dumps("detection")
            message[-1] = ""
            self.rtr.send_multipart(message)

        elif request["task"] == "tracking":
            self.logger.debug("prepare to tracking")
            message[-1] = "finish"
            tracker.Tracker(self.rtr, message)
        else:
            self.logger.debug("requested task is not supported")

 def configure_plain(self, domain='*', passwords=None):
     '''
     Configure PLAIN authentication for a given domain. PLAIN authentication
     uses a plain-text password file. To cover all domains, use "*".
     You can modify the password file at any time; it is reloaded automatically.
     '''
     self.pipe.send_multipart([b'PLAIN', b(domain, self.encoding), jsonapi.dumps(passwords or {})])

 def publish_to_smap(self, smap_identifier, afdd_msg, smap_energyid, energy_impact):
     '''
     Push diagnostic results and energy
     impact to sMAP historian
     '''
     self._log.debug(''.join(['Push to sMAP - ', smap_identifier, str(afdd_msg),
                              ' Energy Impact: ', str(energy_impact)]))
     mytime = int(time.time())
     if smap_energyid is not None:
         content = {
             smap_identifier: {
                  "Readings": [[mytime, afdd_msg]],
                  "Units": "TU",
                  "data_type": "double"
              },
               smap_energyid: {
                  "Readings": [[mytime, energy_impact]],
                  "Units": "kWh",
                  "data_type": "double"}
          }
     else:
         content = {
             smap_identifier: {
                  "Readings": [[mytime, afdd_msg]],
                  "Units": "TU",
                  "data_type": "double"
              }
         }
     self._agent.publish(self.smap_path, self.headers, jsonapi.dumps(content))

 def send(self, socket):
     """Send key-value message to socket; any empty frames are sent as such."""
     key = "" if self.key is None else self.key
     seq_s = struct.pack("!q", self.sequence)
     body = "" if self.body is None else self.body
     prop_s = json.dumps(self.properties)
     socket.send_multipart([key, seq_s, self.uuid, prop_s, body])

    def _set_override_off(self, pattern):
        """Turn off override condition on all devices matching the pattern. It removes the pattern from the override
        patterns set, clears the list of overriden devices  and reevaluates the state of devices. It then cancels the
        pending override event and removes pattern from the config store.
        :param pattern: Override pattern to be removed.
        :type pattern: str
        """

        pattern = pattern.lower()

        # If pattern exactly matches
        if pattern in self._override_patterns:
            self._override_patterns.discard(pattern)
            # Cancel any pending override events
            self._cancel_override_events(pattern)
            self._override_devices.clear()
            patterns = dict()
            # Build override devices list again
            for pat in self._override_patterns:
                for device in self.instances:
                    device = device.lower()
                    if fnmatch.fnmatch(device, pat):
                        self._override_devices.add(device)

                if self._override_interval_events[pat] is None:
                    patterns[pat] = str(0.0)
                else:
                    evt, end_time = self._override_interval_events[pat]
                    patterns[pat] = utils.format_timestamp(end_time)

            self.vip.config.set("override_patterns", jsonapi.dumps(patterns))
        else:
            _log.error("Override Pattern did not match!")
            raise OverrideError(
                "Pattern {} does not exist in list of override patterns".format(pattern))

    def _on_platform_message(self,peer, sender, bus, topic, headers, message):
        """
        Callback function for vcp agent to publish to.

        Platforms that are being managed should publish to this topic with
        the agent_list and other interesting things that the volttron
        central shsould want to know.
        """
        self._log.debug('ON PLATFORM MESSAGE!')
        expected_prefix = "platforms/{}/".format(self.vip_identity)

        if not topic.startswith(expected_prefix):
            self._log.warn(
                "Unexpected topic published to stats function: {}".format(
                    topic
                ))
            return

        self._log.debug("TOPIC WAS: {}".format(topic))
        self._log.debug("MESSAGE WAS: {}".format(message))
        self._log.debug("Expected topic: {}".format(expected_prefix))
        self._log.debug(
            "Are Equal: {}".format(topic.startswith(expected_prefix)))
        self._log.debug("topic type: {} prefix_type: {}".format(type(topic),
                                                                type(
                                                                    expected_prefix)))

        # Pull off the "real" topic from the prefix
        # topic = topic[len(expected_prefix):]

        topicsplit = topic.split('/')
        if len(topicsplit) < 2:
            self._log.error('Invalid topic length published to volttron central')
            return

        # Topic is platforms/<platform_uuid>/otherdata
        topicsplit = topic.split('/')

        if len(topicsplit) < 3:
            self._log.warn("Invalid topic length no operation or datatype.")
            self._log.warn("Topic was {}".format(topic))
            return

        _, platform_uuid, op_or_datatype, other = topicsplit[0], \
                                                  topicsplit[1], \
                                                  topicsplit[2], topicsplit[
                                                                 3:]

        if op_or_datatype in ('iam', 'configure'):
            if not other:
                self._log.error("Invalid response to iam or configure endpoint")
                self._log.error(
                    "the sesson token was not included in response from vcp.")
                return

            ws_endpoint = "/vc/ws/{}/{}".format(other[0], op_or_datatype)
            self._log.debug('SENDING MESSAGE TO {}'.format(ws_endpoint))
            self._vc.vip.web.send(ws_endpoint, jsonapi.dumps(message))
        else:
            self._log.debug("OP WAS: {}".format(op_or_datatype))

 def create_message_and_publish(self, topic, message):
     headers = {
         headers_mod.FROM: AGENT_ID,
         headers_mod.CONTENT_TYPE: headers_mod.CONTENT_TYPE.JSON,
         headers_mod.DATE: datetime.datetime.today().isoformat(),
     }
     self.vip.pubsub.publish("pubsub", topic, headers, jsonapi.dumps(message))

 def get(self):
     nbm = self.application.notebook_manager
     km = self.application.kernel_manager
     files = nbm.list_notebooks()
     for f in files :
         f['kernel_id'] = km.kernel_for_notebook(f['notebook_id'])
     self.finish(jsonapi.dumps(files))