def init_driver_process_client(self):
        """
        @brief Launch the driver process and driver client
        @retval return driver process and driver client object
        """
        log.info("Startup Driver Process")

        this_pid = os.getpid()
        (dvr_proc, cmd_port, evt_port) = ZmqDriverProcess.launch_process(
            self._test_config.driver_module, self._test_config.driver_class, self._test_config.working_dir, this_pid
        )
        self.driver_process = dvr_proc
        log.info(
            "Started driver process for %d %d %s %s"
            % (cmd_port, evt_port, self._test_config.driver_module, self._test_config.driver_class)
        )
        log.info("Driver process pid %d" % self.driver_process.pid)

        # Create driver client.
        self.driver_client = ZmqDriverClient("localhost", cmd_port, evt_port)
        log.info(
            "Created driver client for %d %d %s %s"
            % (cmd_port, evt_port, self._test_config.driver_module, self._test_config.driver_class)
        )

        # Start client messaging.
        self.driver_client.start_messaging(self.event_received)
        log.info("Driver messaging started.")
        gevent.sleep(0.5)

    def check_authentication_token(self, token_string=''):
        """Checks given token and returns a dict with actor id if valid.

        @param token_string    str
        @retval token_info    dict
        @throws BadRequest    Illegal parameter type of value
        @throws NotFound    Token string not found
        @throws Unauthorized    Token not valid anymore or otherwise
        """
        token_id = "token_%s" % token_string
        token = self.container.object_store.read(token_id)
        if not isinstance(token, SecurityToken):
            raise Inconsistent("Token illegal type")
        if token.token_type != TokenTypeEnum.ACTOR_AUTH:
            raise BadRequest("Illegal token type")
        if token.token_string != token_string:
            raise Inconsistent("Found token's token_string does not match")
        cur_time = get_ion_ts_millis()
        if token.status != "OPEN":
            raise Unauthorized("Token status invalid")
        if cur_time >= int(token.expires):
            raise Unauthorized("Token expired")

        token_info = dict(actor_id=token.actor_id,
                    expiry=token.expires,
                    token=token,
                    token_id=token_id)

        log.info("Authentication token %s resolved to actor %s, expiry %s", token_string, token.actor_id, token.expires)

        return token_info

    def on_restart(self, process, config, **kwargs):
        self.process = process
        inst_objs, _ = process.container.resource_registry.find_resources(restype=RT.Instrument, id_only=False)
        active_agents = []
        for inst in inst_objs:
            if len(inst.agent_state) >= 1:
                active_agents.append(inst._id)

        if not active_agents:
            return

        log.info("Restarting %s agents: %s", len(active_agents), active_agents)

        svc_client = ScionManagementProcessClient(process=process)
        for inst_id in active_agents:
            try:
                svc_client.start_agent(inst_id)
            except Exception as ex:
                log.exception("Cannot restart agent for %s" % inst_id)
                if "Agent already active" in ex.message:
                    try:
                        svc_client.stop_agent(inst_id)
                    except Exception:
                        pass
                    try:
                        svc_client.start_agent(inst_id)
                    except Exception:
                        log.warn("Agent stop/start for %s unsuccessful" % inst_id)

    def _trigger_func(self, stream_id):

        sine_ampl = 2.0 # Amplitude in both directions
        samples = 60
        sine_curr_deg = 0 # varies from 0 - 360

        startTime = time.time()
        count = samples #something other than zero

        while True:
            count = time.time() - startTime
            sine_curr_deg = (count % samples) * 360 / samples

            c = [sine_ampl * math.sin(math.radians(sine_curr_deg))]
            t = [sine_ampl * 2 * math.sin(math.radians(sine_curr_deg + 45))]
            p = [sine_ampl * 4 * math.sin(math.radians(sine_curr_deg + 60))]

            lat = lon = [0.0]
            tvar = [time.time()]

            ctd_packet = ctd_stream_packet(stream_id=stream_id,
                c=c, t=t, p = p, lat = lat, lon = lon, time=tvar)

            log.info('SinusoidalCtdPublisher sending 1 record!')
            self.publisher.publish(ctd_packet)

            time.sleep(1.0)

 def _stop_event_subscribers(self):
     """
     Stop event subscribers on cleanup.
     """
     log.info('cleanup: _stop_event_subscribers called.')
     for sub in self._event_subscribers:
         sub.stop()

    def _set_calibration_for_data_product(self, dp_obj, dev_cfg):
        from ion.util.direct_coverage_utils import DirectCoverageAccess
        from coverage_model import SparseConstantType

        log.debug("Setting calibration for data product '%s'", dp_obj.name)
        dataset_ids, _ = self.rr.find_objects(dp_obj, PRED.hasDataset, id_only=True)
        publisher = EventPublisher(OT.InformationContentModifiedEvent)
        if not dataset_ids:
            data_product_management = DataProductManagementServiceProcessClient(process=self)
            log.debug(" Creating dataset for data product %s", dp_obj.name)
            data_product_management.create_dataset_for_data_product(dp_obj._id)
            dataset_ids, _ = self.rr.find_objects(dp_obj, PRED.hasDataset, id_only=True)
            if not dataset_ids:
                raise NotFound('No datasets were found for this data product, ensure that it was created')
        for dataset_id in dataset_ids:
            # Synchronize with ingestion
            with DirectCoverageAccess() as dca:
                cov = dca.get_editable_coverage(dataset_id)
                # Iterate over the calibrations
                for cal_name, contents in dev_cfg.iteritems():
                    if cal_name in cov.list_parameters() and isinstance(cov.get_parameter_context(cal_name).param_type, SparseConstantType):
                        value = float(contents['value'])
                        log.info(' Updating Calibrations for %s in %s', cal_name, dataset_id)
                        cov.set_parameter_values(cal_name, value)
                    else:
                        log.warn(" Calibration %s not found in dataset", cal_name)
                publisher.publish_event(origin=dataset_id, description="Calibrations Updated")
        publisher.close()
        log.info("Calibration set for data product '%s' in %s coverages", dp_obj.name, len(dataset_ids))

    def delete_dataset(self, agent_instance_id, resource_id):

        res_obj = self.rr.read(resource_id)
        dpms = DataProductManagementServiceProcessClient(process=self)

        # Find data products from device id
        count_ds = 0
        dp_objs, _ = self.rr.find_objects(resource_id, PRED.hasOutputProduct, RT.DataProduct, id_only=False)
        for dp_obj in dp_objs:
            if dpms.is_persisted(dp_obj._id):
                raise BadRequest("DataProduct %s '%s' is currently persisted", dp_obj._id, dp_obj.name)

            ds_objs, _ = self.rr.find_objects(dp_obj._id, PRED.hasDataset, RT.Dataset, id_only=False)
            for ds_obj in ds_objs:
                # Delete coverage
                cov_path = DatasetManagementService._get_coverage_path(ds_obj._id)
                if os.path.exists(cov_path):
                    log.info("Removing coverage tree at %s", cov_path)
                    shutil.rmtree(cov_path)
                else:
                    raise OSError("Coverage path does not exist %s" % cov_path)

                # Delete Dataset and associations
                self.rr.delete(ds_obj._id)
                count_ds += 1

        log.info("Datasets and coverages deleted for device %s '%s': %s", resource_id, res_obj.name, count_ds)

    def test_direct_access_vsp_client_disconnect(self):
        """
        Test agent direct_access mode for virtual serial port when shutdown by tcp client disconnect. 
        """

        self.assertInstrumentAgentState(ResourceAgentState.UNINITIALIZED)
    
        self.assertSetInstrumentState(ResourceAgentEvent.INITIALIZE, ResourceAgentState.INACTIVE)

        self.assertSetInstrumentState(ResourceAgentEvent.GO_ACTIVE, ResourceAgentState.IDLE)

        self.assertSetInstrumentState(ResourceAgentEvent.RUN, ResourceAgentState.COMMAND)

        cmd = AgentCommand(command=ResourceAgentEvent.GO_DIRECT_ACCESS,
                           kwargs={'session_type':DirectAccessTypes.vsp,
                           'session_timeout':600,
                           'inactivity_timeout':600})
        
        retval = self.assertSetInstrumentState(cmd, ResourceAgentState.DIRECT_ACCESS)

        log.info("GO_DIRECT_ACCESS retval=" + str(retval.result))

        tcp_client = self._start_tcp_client(retval)
        
        self.assertTrue(tcp_client.send_data('ts\r\n'))
        
        tcp_client.disconnect()

        self.assertInstrumentAgentState(ResourceAgentState.COMMAND, timeout=20)        
                
        self.assertSetInstrumentState(ResourceAgentEvent.RESET, ResourceAgentState.UNINITIALIZED)

    def test_direct_access_vsp_inactivity_timeout(self):
        """
        Test agent direct_access mode for virtual serial port when shutdown by inactivity timeout. 
        """

        self.assertInstrumentAgentState(ResourceAgentState.UNINITIALIZED)
    
        self.assertSetInstrumentState(ResourceAgentEvent.INITIALIZE, ResourceAgentState.INACTIVE)

        self.assertSetInstrumentState(ResourceAgentEvent.GO_ACTIVE, ResourceAgentState.IDLE)

        self.assertSetInstrumentState(ResourceAgentEvent.RUN, ResourceAgentState.COMMAND)

        cmd = AgentCommand(command=ResourceAgentEvent.GO_DIRECT_ACCESS,
                           kwargs={'session_type':DirectAccessTypes.vsp,
                           'session_timeout':600,
                           'inactivity_timeout':10})
        
        retval = self.assertSetInstrumentState(cmd, ResourceAgentState.DIRECT_ACCESS)

        log.info("GO_DIRECT_ACCESS retval=" + str(retval.result))

        self._start_tcp_client(retval)

        self.assertInstrumentAgentState(ResourceAgentState.COMMAND, timeout=30)        
        
        # test for starting and inactivity timeout w/o ever connecting TCP client 
        retval = self.assertSetInstrumentState(cmd, ResourceAgentState.DIRECT_ACCESS)
        self.assertInstrumentAgentState(ResourceAgentState.COMMAND, timeout=30)        
        
        self.assertSetInstrumentState(ResourceAgentEvent.RESET, ResourceAgentState.UNINITIALIZED)

 def consume_event(*args, **kwargs):
     log.info('Test recieved ION event: args=%s, kwargs=%s, event=%s.', 
              str(args), str(kwargs), str(args[0]))
     self._events_received.append(args[0])
     if self._event_count > 0 and \
         self._event_count == len(self._events_received):
         self._async_event_result.set()

    def test_direct_access_telnet_session_setup_failure(self):
        """
        Test agent direct_access mode for telnet when session setup fails. 
        """

        self.assertInstrumentAgentState(ResourceAgentState.UNINITIALIZED)
    
        self.assertSetInstrumentState(ResourceAgentEvent.INITIALIZE, ResourceAgentState.INACTIVE)

        self.assertSetInstrumentState(ResourceAgentEvent.GO_ACTIVE, ResourceAgentState.IDLE)

        self.assertSetInstrumentState(ResourceAgentEvent.RUN, ResourceAgentState.COMMAND)

        cmd = AgentCommand(command=ResourceAgentEvent.GO_DIRECT_ACCESS,
                           kwargs={'session_type': DirectAccessTypes.telnet,
                           'session_timeout':600,
                           'inactivity_timeout':600})

        # test that DA session quits when bad token is sent
        retval = self.assertSetInstrumentState(cmd, ResourceAgentState.DIRECT_ACCESS)

        log.info("GO_DIRECT_ACCESS retval=" + str(retval.result))

        tcp_client = self._start_tcp_client(retval)
        
        self.assertTrue(tcp_client.start_telnet(token=retval.result['token'], handshake=False))

        self.assertInstrumentAgentState(ResourceAgentState.COMMAND, timeout=20)        

        def start_DeviceStatusAlertEvent_subscriber(value_id, sub_type):
            """
            @return async_event_result  Use it to wait for the expected event
            """
            event_type = "DeviceStatusAlertEvent"

            async_event_result = AsyncResult()

            def consume_event(evt, *args, **kwargs):
                log.info('DeviceStatusAlertEvent_subscriber received evt: %s', str(evt))
                if evt.type_ != event_type or \
                   evt.value_id != value_id or \
                   evt.sub_type != sub_type:
                    return

                async_event_result.set(evt)

            kwargs = dict(event_type=event_type,
                          callback=consume_event,
                          origin=self.p_root.platform_device_id,
                          sub_type=sub_type)

            sub = EventSubscriber(**kwargs)
            sub.start()
            log.info("registered DeviceStatusAlertEvent subscriber: %s", kwargs)

            self._event_subscribers.append(sub)
            sub._ready_event.wait(timeout=self._receive_timeout)

            return async_event_result

    def _call_plugins(self, method, process, config, **kwargs):
        bootstrap_plugins = config.get_safe("bootstrap_plugins", None)
        if bootstrap_plugins is None:
            log.warn("Bootstrapper called without bootstrap_plugins config")

        # Finding the system actor ID. If found, construct call context headers.
        # This may be called very early in bootstrap with no system actor yet existing
        system_actor, _ = process.container.resource_registry.find_resources(
            RT.ActorIdentity, name=self.CFG.system.system_actor, id_only=True
        )
        system_actor_id = system_actor[0] if system_actor else "anonymous"

        actor_headers = {
            "ion-actor-id": system_actor_id,
            "ion-actor-roles": {"ION": ["ION_MANAGER", "ORG_MANAGER"]} if system_actor else {},
        }

        # Set the call context of the current process
        with process.push_context(actor_headers):

            for plugin_info in bootstrap_plugins:
                plugin_mod, plugin_cls = plugin_info.get("plugin", [None, None])
                plugin_cfg = plugin_info.get("config", None)
                plugin_cfg = dict_merge(config, plugin_cfg) if plugin_cfg is not None else config

                try:
                    log.info("Bootstrapping plugin %s.%s ...", plugin_mod, plugin_cls)
                    plugin = for_name(plugin_mod, plugin_cls)
                    plugin_func = getattr(plugin, method)
                    plugin_func(process, plugin_cfg, **kwargs)
                except AbortBootstrap as abort:
                    raise
                except Exception as ex:
                    log.exception("Error bootstrapping plugin %s.%s", plugin_mod, plugin_cls)

def start_fake_instrument_agent(container, stream_config={}, message_headers=None):

    # Create agent config.
    agent_config = {
        'driver_config' : DVR_CONFIG,
        'stream_config' : stream_config,
        'agent'         : {'resource_id': IA_RESOURCE_ID},
        'test_mode' : True
    }

    # Start instrument agent.

    log.debug("TestInstrumentAgent.setup(): starting IA.")
    container_client = ContainerAgentClient(node=container.node,
        name=container.name)

    ia_pid = container_client.spawn_process(name=IA_NAME,
        module=IA_MOD,
        cls=IA_CLS,
        config=agent_config, headers=message_headers)

    log.info('Agent pid=%s.', str(ia_pid))

    # Start a resource agent client to talk with the instrument agent.

    ia_client = ResourceAgentClient(IA_RESOURCE_ID, process=FakeProcess())
    log.info('Got ia client %s.', str(ia_client))

    return ia_client

    def _set_resource(self):
        attrNames = self.ATTR_NAMES
        writ_attrNames = self.WRITABLE_ATTR_NAMES

        # do valid settings:

        # TODO more realistic value depending on attribute's type
        attrs = [(attrName, self.VALID_ATTR_VALUE) for attrName in attrNames]
        log.info("%r: setting attributes=%s", self.PLATFORM_ID, attrs)
        kwargs = dict(attrs=attrs)
        cmd = AgentCommand(command=PlatformAgentEvent.SET_RESOURCE, kwargs=kwargs)
        retval = self._execute_agent(cmd)
        attr_values = retval.result
        self.assertIsInstance(attr_values, dict)
        for attrName in attrNames:
            if attrName in writ_attrNames:
                self._verify_valid_attribute_id(attrName, attr_values)
            else:
                self._verify_not_writable_attribute_id(attrName, attr_values)

        # try invalid settings:

        # set invalid values to writable attributes:
        attrs = [(attrName, self.INVALID_ATTR_VALUE) for attrName in writ_attrNames]
        log.info("%r: setting attributes=%s", self.PLATFORM_ID, attrs)
        kwargs = dict(attrs=attrs)
        cmd = AgentCommand(command=PlatformAgentEvent.SET_RESOURCE, kwargs=kwargs)
        retval = self._execute_agent(cmd)
        attr_values = retval.result
        self.assertIsInstance(attr_values, dict)
        for attrName in writ_attrNames:
            self._verify_attribute_value_out_of_range(attrName, attr_values)

        def got_event(evt, *args, **kwargs):
            if not self._active:
                log.warn("%r: got_event called but manager has been destroyed",
                         self._platform_id)
                return

            if evt.type_ != event_type:
                log.trace("%r: ignoring event type %r. Only handle %r directly",
                          self._platform_id, evt.type_, event_type)
                return

            if evt.sub_type != sub_type:
                log.trace("%r: ignoring event sub_type %r. Only handle %r",
                          self._platform_id, evt.sub_type, sub_type)
                return

            state = self._agent.get_agent_state()

            statuses = formatted_statuses(self.aparam_aggstatus,
                                          self.aparam_child_agg_status,
                                          self.aparam_rollup_status)

            invalidated_children = self._agent._get_invalidated_children()

            log.info("%r/%s: (%s) status report triggered by diagnostic event:\n"
                     "%s\n"
                     "%40s : %s\n",
                     self._platform_id, state, self.resource_id, statuses,
                     "invalidated_children", invalidated_children)

    def _get_agent_instance_id(self, resource_id):
        dsaids, _ = self.rr.find_objects(subject=resource_id,
                                         predicate=PRED.hasAgentInstance,
                                         object_type=RT.ExternalDatasetAgentInstance,
                                         id_only=True)

        iaids, _ = self.rr.find_objects(subject=resource_id,
                                        predicate=PRED.hasAgentInstance,
                                        object_type=RT.InstrumentAgentInstance,
                                        id_only=True)

        paids, _ = self.rr.find_objects(subject=resource_id,
                                        predicate=PRED.hasAgentInstance,
                                        object_type=RT.PlatformAgentInstance,
                                        id_only=True)

        aids = dsaids + iaids + paids
        if len(aids) > 1:
            log.error("Multiple agent instances found")
            raise BadRequest("Failed to identify agent instance")

        if len(aids) == 0:
            log.error("Agent instance not found")
            raise BadRequest("Failed to identify agent instance")

        log.info("Found agent instance ID: %s", aids[0])
        return aids[0]

 def process_packet_cb(packet, route, stream):
     if not isinstance(packet, DataPacket):
         log.warn("Received a non DataPacket message")
     self.recv_packets.append(packet)
     self.recv_rows += len(packet.data["data"])
     log.info("Received data packet #%s: rows=%s, cols=%s", len(self.recv_packets), len(packet.data["data"]),
              packet.data["cols"])

    def recover_data(self, agent_instance_id, resource_id):
        res_obj = self.rr.read(resource_id)

        if res_obj.type_ != RT.InstrumentDevice:
            log.warn("Ignoring resource because it is not an instrument: %s - %s", res_obj.name, res_obj.type_)
            self._recover_data_status['ignored'].append("%s (%s)" % (res_obj.name, res_obj.type_))
            return

        self.recover_start = self.CFG.get("recover_start", None)
        self.recover_end = self.CFG.get("recover_end", None)

        if self.recover_end is None:
            raise BadRequest("Missing recover_end parameter")
        if self.recover_start is None:
            raise BadRequest("Missing recover_start parameter")

        try:
            ia_client = ResourceAgentClient(resource_id, process=self)
            log.info('Got ia client %s.', str(ia_client))
            ia_client.execute_resource(command=AgentCommand(command=DriverEvent.GAP_RECOVERY, args=[self.recover_start, self.recover_end]))

            self._recover_data_status['success'].append(res_obj.name)
        except Exception as e:
            log.warn("Failed to start recovery process for %s", res_obj.name)
            log.warn("Exception: %s", e)
            self._recover_data_status['fail'].append("%s (%s)" % (res_obj.name, e))

    def _traverse(self, platform_id, parent_platform_objs=None):
        """
        Recursive routine that repeatedly calls _prepare_platform to build
        the object dictionary for each platform.

        @param platform_id ID of the platform to be visited
        @param parent_platform_objs dict of objects associated to parent
                        platform, if any.

        @retval the dict returned by _prepare_platform at this level.
        """

        log.info("Starting _traverse for %r", platform_id)

        plat_objs = self._prepare_platform(platform_id, parent_platform_objs)

        self.all_platforms[platform_id] = plat_objs

        # now, traverse the children:
        retval = self.rsn_oms.getSubplatformIDs(platform_id)
        subplatform_ids = retval[platform_id]
        for subplatform_id in subplatform_ids:
            self._traverse(subplatform_id, plat_objs)

        # note, topology indexed by platform_id
        self.topology[platform_id] = plat_objs['children']

        return plat_objs