Python threading 模块，Timer() 实例源码

def blinking():
    if keepalive:
        threading.Timer(10.0, blinking).start()

        # Only blink when we are actually building
        if building or error:
            # If error, blink red.
            if error:
                color = "red"
            else:
                color = "yellow"

            alloff()
            pin = getcode(color)
            GPIO.output(pin, True)
            time.sleep(3)
            GPIO.output(pin, False)


# Check every 10s if we are building, if not or done get latest status

def testCaptureSequence_StopTime(self):
    start_time = 1.0
    stop_time = time.time() + 1.0

    self.capture_messages[-1].time += time.time()
    threading.Timer(0.1, self.send_capture_messages).start()

    captured_seq = self.midi_hub.capture_sequence(
        120, start_time, stop_time=stop_time)

    expected_seq = music_pb2.NoteSequence()
    expected_seq.tempos.add(qpm=120)
    expected_seq.total_time = stop_time
    testing_lib.add_track_to_sequence(
        expected_seq, 0,
        [Note(1, 64, 2, 5), Note(2, 64, 3, 4), Note(3, 64, 4, stop_time)])
    self.assertProtoEquals(captured_seq, expected_seq)

def execWait(cmd, outfile=None, timeout=0):
        result = ""
        env = os.environ
        proc = subprocess.Popen(cmd, executable='/bin/bash', env=env, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, shell=True)

        if timeout:
            timer = threading.Timer(timeout, proc.kill)
            timer.start()
        result = proc.communicate()[0]
        if timeout:
            if timer.is_alive():
                timer.cancel()
        if outfile:
            if Utils.fileExists(outfile):
                print("FILE ALREADY EXISTS!!!!")
            else:
                tmp_result = "\033[0;33m(" + time.strftime(
                    "%Y.%m.%d-%H.%M.%S") + ") <pentest> #\033[0m " + cmd + Utils.newLine() + Utils.newLine() + result
                Utils.writeFile(tmp_result, outfile, 'ab')
        return result

def start(self):
        self.deviceHandler.start()
        if self.protocol == "udp":
            self.loadState()        
            self.logger.debug("udpHeartbeatSeconds = {0}".format(self.udpHeartbeatSeconds))
            self.logger.debug("udpDataPacketInterval = {0}".format(self.udpDataPacketInterval))
            self.udpServer = SocketServer.UDPServer(('0.0.0.0', 0), IotUDPHandler)
            self.udpServer.service = self
            self.udpServer.role = IotUDPHandler.CLIENT
            self.logger.info("starting UDP client at {0}:{1} connecting to {2}, state at {3}".format(self.udpServer.server_address[0], self.udpServer.server_address[1], self.serverAddr, self.stateFile))            
            timer = threading.Timer(0.5, self.repeat)
            timer.daemon = True
            timer.start()
            self.udpServer.serve_forever()      
        elif self.protocol == "ssl":
            while True:
                self.logger.info("Connecting by SSL to server at {0}".format(self.serverAddr))
                try:
                    sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
                    self.logger.debug("using caCertFile={0}, deviceCertFile={1}, deviceKeyFile={2}".format(self.caCertFile, self.deviceCertFile, self.deviceKeyFile))
                    sslSocket = ssl.wrap_socket(sock, ca_certs=self.caCertFile, cert_reqs=ssl.CERT_REQUIRED, certfile=self.deviceCertFile, keyfile=self.deviceKeyFile, ssl_version=ssl.PROTOCOL_TLSv1)     
                    sslSocket.connect((self.serverAddr.split(':')[0], int(self.serverAddr.split(':')[1])))   
                    servercert = sslSocket.getpeercert()
                    subject = dict(x[0] for x in servercert['subject'])
                    self.logger.info("Connected to server with valid certificate, CN={0}".format(subject['commonName']))  
                    self.sslSocket = sslSocket
                    sslThread = threading.Thread(target = self.sslListen, args = (self.sslSocket,))
                    sslThread.daemon = True
                    sslThread.start()
                    while True:
                        payload = self.deviceHandler.getMessagePayload()
                        self.logger.debug("Sending payload to {0} by SSL: {1}".format(self.serverAddr, payload))
                        iotcommon.sendMessage(self.sslSocket, payload)
                        time.sleep(self.sslIntervalSeconds)
                except Exception as e: 
                    self.logger.exception(e)
                time.sleep(10)

def delay(self, timeout, func):
        def func_wrapper():
            try:
                func()
            except Exception:
                self.exception()

        t = threading.Timer(timeout, func_wrapper, ())
        t.start()

        return t

def schedule(self, timeout, interval, func):
        tw = TimerWraper()

        def func_wrapper():
            start = time.time()

            try:
                func()
            except Exception:
                self.exception()

            with tw.cancel_lock:
                if not tw.canceled:
                    tw.timer = threading.Timer(abs(interval - (time.time() - start)), func_wrapper, ())
                    tw.timer.start()

        tw.timer = threading.Timer(timeout, func_wrapper, ())
        tw.timer.start()

        return tw

def execute(self):
        try:
            self.system.run()
        except (ReadTimeout, ConnectionError, JSONDecodeError):
            pass
        except exceptions.TradingSystemException as e:
            curr = datetime.now()
            print('{time} - {text}'.format(time=curr.strftime('%Y-%m-%d %H:%M:%S'), text=str(e)))
        except Exception as e:
            curr = datetime.now()
            print('{time} - {text} - {args}'.format(time=curr.strftime('%Y-%m-%d %H:%M:%S'), text=str(e), args=e.args))
            traceback.print_exc()

        if self.interval:
            threading.Timer(self.interval, self.execute).start()

def _schedule_auto_flush(self):
        """
        Schedule an automatic flush of the current batch.
        """

        if not self.auto_flush_timeout:
            return  # Auto-flush is disabled.

        self.state_lock.acquire()
        try:
            if self.flush_timer:
                return

            self.flush_timer = Timer(self.auto_flush_timeout, self.flush)
            self.flush_timer.daemon = True
            self.flush_timer.start()
        finally:
            self.state_lock.release()

def test_bp_exe_by_name(proc32_64_debug):
    class TSTBP(windows.debug.Breakpoint):
        COUNTER = 0
        def trigger(self, dbg, exc):
            TSTBP.COUNTER += 1
            assert TSTBP.COUNTER == 1
            # Kill the target in 0.5s
            # It's not too long
            # It's long enought to get trigger being recalled if implem is broken
            threading.Timer(0.5, proc32_64_debug.exit).start()

    exepe = proc32_64_debug.peb.exe
    entrypoint = exepe.get_OptionalHeader().AddressOfEntryPoint
    exename = os.path.basename(proc32_64_debug.peb.imagepath.str)
    d = windows.debug.Debugger(proc32_64_debug)
    # The goal is to test bp of format 'exename!offset' so we craft a string based on the entrypoint
    d.add_bp(TSTBP("{name}!{offset}".format(name=exename, offset=entrypoint)))
    d.loop()
    assert TSTBP.COUNTER == 1

def exit_after(s):
    def outer(fn):
        def inner(*args, **kwargs):
            try:
                timer = Timer(s, quit_function)
                timer.start()
                result = fn(*args, **kwargs)
            except KeyboardInterrupt:
                logger = init_logger("exit_after_decorator")
                logger.info("Timeout reached!")
                print("Timeout reached!")
                timer.cancel()
                return -1
            finally:
                timer.cancel()
            return result
        return inner
    return outer

def commit(self, restart=True):
        if self.period:
            self.timer.cancel()

        with self.lock:
            ramreader = self._get_ram_reader()
            self._make_ram_index()

        if self.bufferedcount:
            self.writer.add_reader(ramreader)
        self.writer.commit(**self.commitargs)
        self.bufferedcount = 0

        if restart:
            self.writer = self.index.writer(**self.writerargs)
            if self.period:
                self.timer = threading.Timer(self.period, self.commit)
                self.timer.start()

def mm_heartbeat(self):
        # Check if stop or set next timer
        if self.shutdown:
            return
        threading.Timer(self.hb_timer, self.mm_heartbeat).start()

        address = ("http://" + self.mm_host + ":" + self.mm_port + "/alexapi?action=AVSHB")

        logger.debug("Sending MM Heatbeat")

        try:
            response = urlopen(address).read()
        except URLError as err:
            logger.error("URLError: %s", err.reason)
            return

        logger.debug("Response: " + response)

def process_response(self, response):
        if response is not None:
            answer = response.get('text')
            if not self.ignore_indicator:
                self.process_indicator(answer)
            response['text'] = norm(answer)
            self.last_response = response
            if self.response_listener is None:
                self.stdout.write('{}[by {}]: {}\n'.format(
                    self.botname, response.get('botid'),
                    response.get('text')))
            else:
                try:
                    threading.Timer(0, self.response_listener.on_response, (self.session, response)).start()
                except Exception as ex:
                    logger.error(ex)

def process_indicator(self, reply):
        cmd, timeout = None, None
        for match in re.findall(r'\[.*\]', reply):
            match = match.strip()
            match = match.replace(' ', '')
            if match == '[loopback=0]':
                self.cancel_timer()
                return
            match = match.replace(']', '')
            match = match.replace('[', '')
            if '=' in match:
                cmd, timeout = match.split('=')
                self.timeout = float(timeout)/1000
            else:
                cmd = match
            cmd = '[{}]'.format(cmd)
        if self.timeout is not None and cmd is not None:
            self.cancel_timer()
            self.timer = threading.Timer(self.timeout, self.ask, (cmd, ))
            self.timer.start()
            logger.info("Start {} timer with timeout {}".format(
                cmd, self.timeout))

def _process_move_from_event(self, drive, local_parent_path, ent_name, is_folder):
        # First try finding the item in database.
        rel_parent_path = _get_rel_parent_path(drive, local_parent_path)
        item_store = self._items_store_man.get_item_storage(drive)
        q = item_store.get_items_by_id(local_parent_path=local_parent_path, item_name=ent_name)
        try:
            item_id, item = q.popitem()
            if item.is_folder != is_folder:
                raise KeyError()
        except KeyError:
            # If the record does not match, sync the parent after some time.
            threading.Timer(self.SYNC_PARENT_DELAY_SEC, self._sync_parent_dir_of, (drive, rel_parent_path))
            return
        task = delete_task.DeleteItemTask(parent_task=self._task_bases[drive], rel_parent_path=rel_parent_path,
                                          item_name=ent_name, is_folder=is_folder)
        task.item_obj = item
        self._delayed_tasks.add(task)
        threading.Timer(self.MOVE_DETECTION_DELAY_SEC, self._enqueue_delayed_task, task)

def execute(cmd, shell=False, root=False, timeout=10):
    try:
        if root:
            cmd.insert(0, 'sudo')
        LOG.info(cmd)
        subproc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE, shell=shell)
        timer = Timer(timeout, lambda proc: proc.kill(), [subproc])
        timer.start()
        subproc.wait()
        stdcode = subproc.returncode
        stdout = subproc.stdout.readlines()
        stderr = subproc.stderr.readlines()
        timer.cancel()

        def list_strip(lines):
            return [line.strip() for line in lines]
        return stdcode, list_strip(stderr) if stdcode else list_strip(stdout)
    except Exception as e:
        LOG.error(e)
        raise

def on_selection_modified_async(self, view):
        """
        ST API method: gets called when selection changed
        """
        if len(view.sel()) > 1 or not INFOS.addon:
            return None
        try:
            region = view.sel()[0]
        except Exception:
            return None
        if region == self.prev_selection:
            return None
        self.prev_selection = region
        delay = self.settings.get("tooltip_delay", 200)
        if self.timer:
            self.timer.cancel()
        self.timer = Timer(delay / 1000, self.show_tooltip, (view,))
        self.timer.start()

def Update(self):
    # Get our wake-up thread ready...
    self.ticker = threading.Timer(self.INTERVAL, self.Update)
    try:
      # Check if any of our topics have new info to pass on
      if len(self.topics):     
        refresh = False
        for topic in self.topics.itervalues():
          topic.Update(self)
          if topic.HasChanged():
            refresh = True
          topic.Reset()

        if refresh:
          self.SignalExcel()
    finally:
      self.ticker.start() # Make sure we get to run again

def debounce(func):
    def debounced(self, *args, **kwargs):
        flow_settings = find_flow_settings(
            self.view.window().project_data()
        )
        debounce_ms = flow_settings.get('debounce_ms')

        def call_func():
            func(self, *args, **kwargs)
        try:
            debounced.timer.cancel()
        except(AttributeError):
            pass

        debounced.timer = Timer(debounce_ms / 1000, call_func)
        debounced.timer.start()
    return debounced

def repeat(start_time, repeat_time):
    if repeat_time < 1:
        logging.error('Repeating function must have a repeat time greater than 1 second')

        def repeat_inner(func):
            return func
        return repeat_inner

    def repeat_inner(func):
        @wraps(func)
        def func_inner():
            t = threading.Timer(repeat_time, func_inner)
            t.daemon = True
            t.start()
            return func()
        t = threading.Timer(start_time, func_inner)
        t.daemon = True
        t.start()
        return func_inner
    return repeat_inner

def doRpcServer(port, stopTimeSec):
    class EchoHandler(object):
        def Echo123(self, msg1, msg2, msg3):
            return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3))
        def EchoStruct(self, msg):
            return ("%s" % msg)

    addr = msgpackrpc.Address('localhost', port)
    server = msgpackrpc.Server(EchoHandler())
    server.listen(addr)
    # run thread to stop it after stopTimeSec seconds if > 0
    if stopTimeSec > 0:
        def myStopRpcServer():
            server.stop()
        t = threading.Timer(stopTimeSec, myStopRpcServer)
        t.start()
    server.start()

def testCaptureSequence_StopSignal(self):
    start_time = 1.0

    threading.Timer(0.1, self.send_capture_messages).start()

    captured_seq = self.midi_hub.capture_sequence(
        120, start_time,
        stop_signal=midi_hub.MidiSignal(type='control_change', control=1))

    expected_seq = music_pb2.NoteSequence()
    expected_seq.tempos.add(qpm=120)
    expected_seq.total_time = 6.0
    testing_lib.add_track_to_sequence(
        expected_seq, 0,
        [Note(1, 64, 2, 5), Note(2, 64, 3, 4), Note(3, 64, 4, 6)])
    self.assertProtoEquals(captured_seq, expected_seq)

def testCaptureSequence_Mono(self):
    start_time = 1.0

    threading.Timer(0.1, self.send_capture_messages).start()
    self.midi_hub = midi_hub.MidiHub(self.port, self.port,
                                     midi_hub.TextureType.MONOPHONIC)
    captured_seq = self.midi_hub.capture_sequence(
        120, start_time,
        stop_signal=midi_hub.MidiSignal(type='control_change', control=1))

    expected_seq = music_pb2.NoteSequence()
    expected_seq.tempos.add(qpm=120)
    expected_seq.total_time = 6
    testing_lib.add_track_to_sequence(
        expected_seq, 0,
        [Note(1, 64, 2, 3), Note(2, 64, 3, 4), Note(3, 64, 4, 6)])
    self.assertProtoEquals(captured_seq, expected_seq)

def addReminder(name, time, uuid, body='', urgency=0, hidden=True):
    """
    Queue reminder.

    Show notification at the specified time. With the given name as title and an optional body
    for further information.
    The mandatory is used to identify the reminder and remove it with removeReminder().
    If the reminder should show up in the list printed by 'remind print' hidden (default: True)
    should be set to false. In this case the reminder is requeued at startup. If reminders are
    used e.g. with a todo list for due dates, hidden should probably be set to true so that the
    list is not cluttered with automatically created data.
    If the reminder needs a different priority, it can be set with urgency to critical (=2),
    high (=1) or normal (=0, default).
    """
    waitTime = time - dt.now()
    n = notify2.Notification(name, body)
    n.set_urgency(urgency)
    timerList[uuid] = Timer(waitTime.total_seconds(), showAlarm, [n, name])
    timerList[uuid].start()
    newItem = {'name': name, 'time': time, 'hidden': hidden, 'uuid': uuid}
    reminderList['items'].append(newItem)
    reminderList['items'] = sort(reminderList['items'])
    write_file("reminderlist.txt", reminderList)

def test_relay():
   """Test relay on and off cycle"""

   # check if the output is high
   print 'current control output is: ', is_output_high(), ' (should be off)'

   # start the relay
   start_relay()

   print 'current control output is: ', is_output_high(), ' (should be on)'

   # setup a timer to stop the relay after 5 seconds
   t = Timer(5, stop_relay)
   t.start()

   # wait for the timer to finish
   t.join()

def test_recv(self, tco):
        pdu = nfc.llcp.pdu.UnnumberedInformation(1, 1, HEX('1122'))
        assert tco.enqueue(pdu) is True
        assert tco.recv() == pdu
        threading.Timer(0.01, tco.close).start()
        with pytest.raises(nfc.llcp.Error) as excinfo:
            tco.recv()
        assert excinfo.value.errno == errno.EPIPE
        with pytest.raises(nfc.llcp.Error) as excinfo:
            tco.recv()
        assert excinfo.value.errno == errno.ESHUTDOWN


# =============================================================================
# Logical Data Link
# =============================================================================

def test_recvfrom(self, tco):
        pdu = nfc.llcp.pdu.Symmetry()
        assert tco.enqueue(pdu) is False
        pdu = nfc.llcp.pdu.UnnumberedInformation(1, 1, (tco.recv_miu+1) * b'1')
        assert tco.enqueue(pdu) is False
        pdu = nfc.llcp.pdu.UnnumberedInformation(1, 1, HEX('1122'))
        assert tco.enqueue(pdu) is True
        assert tco.recvfrom() == (pdu.data, pdu.ssap)
        threading.Timer(0.01, tco.close).start()
        with pytest.raises(nfc.llcp.Error) as excinfo:
            tco.recvfrom()
        assert excinfo.value.errno == errno.EPIPE
        with pytest.raises(nfc.llcp.Error) as excinfo:
            tco.recvfrom()
        assert excinfo.value.errno == errno.ESHUTDOWN


# =============================================================================
# Data Link Connection
# =============================================================================

def test_accept_connect(self, llc, ldl, dlc, peer_miu, send_miu):
            with pytest.raises(nfc.llcp.Error) as excinfo:
                llc.accept(object())
            assert excinfo.value.errno == errno.ENOTSOCK
            with pytest.raises(nfc.llcp.Error) as excinfo:
                llc.accept(ldl)
            assert excinfo.value.errno == errno.EOPNOTSUPP
            with pytest.raises(nfc.llcp.Error) as excinfo:
                llc.accept(dlc)
            assert excinfo.value.errno == errno.EINVAL
            connect_pdu = nfc.llcp.pdu.Connect(4, 32, peer_miu)
            threading.Timer(0.01, llc.dispatch, (connect_pdu,)).start()
            llc.bind(dlc, b'urn:nfc:sn:snep')
            llc.listen(dlc, 0)
            sock = llc.accept(dlc)
            assert isinstance(sock, nfc.llcp.tco.DataLinkConnection)
            assert llc.getsockopt(sock, nfc.llcp.SO_SNDMIU) == send_miu
            assert llc.getpeername(sock) == 32
            assert llc.getsockname(sock) == 4

def __init__(self, network, defaults, app):
        self.network = network
        self.defaults = defaults
        self.pushing = False
        self.app = app
        super().__init__()

        # endless loop for updating information displayed
        def update_statistics():
            self.ids.listening_interface_label.text = 'Listening interface: {}:{}'.format(bptc.ip, bptc.port)
            self.ids.event_count_label.text = '{} events, {} confirmed'.format(len(self.hashgraph.lookup_table.keys()),
                                                                               len(self.hashgraph.ordered_events))
            self.ids.last_push_sent_label.text = 'Last push sent: {}'.format(self.network.last_push_sent)
            self.ids.last_push_received_label.text = 'Last push received: {}'.format(self.network.last_push_received)

            t = threading.Timer(1, update_statistics)
            t.daemon = True
            t.start()

        update_statistics()

def convert(a):
    ctState.currentState = a
    if ctState.currentState != st.nomal:
        print("CurrentState %s" % a)
    if ctState.currentState == st.knockOr:
        global knockNum
        knockNum += 1
        if knockNum == 2:
            ctState.currentState = st.knock
    if ctState.currentState == st.knock:
        # url = "192.168.12.101"
        # port = 9001
        # MessageHandle.CallAlarm(url, port, 1)
        print "Tong zhi Yong hu :you ren qiao men"




# Timer???

def __watch_and_store_partitions(self):
        """
        Flush all the dirty partitions to the backend store. This methods gets called periodically on a timer
        :return:
        """
        logging.debug("checking for dirty partitions in cache: {} found".format(len(self.__dirty_partitions)))
        while self.__dirty_partitions:
            pid = self.__dirty_partitions.pop()
            logging.warning("flushing modified partition {} from cache to mgmt container {}".format(pid,
                                                                                                    self.partitionStore.containerNameSdosMgmt))
            try:
                self.partitionStore.writePartition(pid, io.BytesIO(self.partitionCache[pid]))
            except Exception:
                self.__dirty_partitions.add(pid)
                logging.exception(
                    "storing changed partition {} failed! {} dirty partitions left to store. Leaving this execution.".format(
                        pid, len(self.__dirty_partitions)))
                break
        threading.Timer(10, self.__watch_and_store_partitions).start()

def Update(self):
    # Get our wake-up thread ready...
    self.ticker = threading.Timer(self.INTERVAL, self.Update)
    try:
      # Check if any of our topics have new info to pass on
      if len(self.topics):     
        refresh = False
        for topic in self.topics.itervalues():
          topic.Update(self)
          if topic.HasChanged():
            refresh = True
          topic.Reset()

        if refresh:
          self.SignalExcel()
    finally:
      self.ticker.start() # Make sure we get to run again

def doRpcServer(port, stopTimeSec):
    class EchoHandler(object):
        def Echo123(self, msg1, msg2, msg3):
            return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3))
        def EchoStruct(self, msg):
            return ("%s" % msg)

    addr = msgpackrpc.Address('localhost', port)
    server = msgpackrpc.Server(EchoHandler())
    server.listen(addr)
    # run thread to stop it after stopTimeSec seconds if > 0
    if stopTimeSec > 0:
        def myStopRpcServer():
            server.stop()
        t = threading.Timer(stopTimeSec, myStopRpcServer)
        t.start()
    server.start()

def doRpcServer(port, stopTimeSec):
    class EchoHandler(object):
        def Echo123(self, msg1, msg2, msg3):
            return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3))
        def EchoStruct(self, msg):
            return ("%s" % msg)

    addr = msgpackrpc.Address('localhost', port)
    server = msgpackrpc.Server(EchoHandler())
    server.listen(addr)
    # run thread to stop it after stopTimeSec seconds if > 0
    if stopTimeSec > 0:
        def myStopRpcServer():
            server.stop()
        t = threading.Timer(stopTimeSec, myStopRpcServer)
        t.start()
    server.start()

def __init__(self, host='192.168.1.1'):
        self.host = host

        self.sequence = 1
        self.timer = 0.2
        self.com_watchdog_timer = threading.Timer(self.timer, self.commwdg)
        self.lock = threading.Lock()
        self.speed = 0.2
        self.at(ardrone.at.config, 'general:navdata_demo', 'TRUE')
        self.at(ardrone.at.config, 'control:altitude_max', '20000')
        self.video_pipe, video_pipe_other = multiprocessing.Pipe()
        self.nav_pipe, nav_pipe_other = multiprocessing.Pipe()
        self.com_pipe, com_pipe_other = multiprocessing.Pipe()
        self.network_process = ardrone.network.ARDroneNetworkProcess(self.host, nav_pipe_other, video_pipe_other, com_pipe_other)
        self.network_process.start()
        self.ipc_thread = ardrone.network.IPCThread(self)
        self.ipc_thread.start()
        self.image = PIL.Image.new('RGB', (640, 360))
        self.navdata = dict()
        self.time = 0

def keep_status_bar_cover_alive(dev):
    """
    keep sending 'turn on status bar cover' broadcast, for the cover effect is not very perfect
    :param dev: device id to process
    :return: None
    """
    StatusBarCover.switch_cover(dev, True)
    threading.Timer(5, keep_status_bar_cover_alive, [dev]).start()

def adjust_limits(self):
        self.logger.info("adjust_limits: checking midpoint limits")
        current_time = datetime.datetime.today()
        for orderid in self.openorders.keys():
            if self.openorders[orderid].adjust_periodical == True and orderid in self.__MapToExecuteOrderID:
                if (current_time - self.openorders[orderid].last_adjust).total_seconds() > self.limit_adjust_interval*0.9:

                    contract = self.openorders[orderid].contract
                    if (current_time - self.openorders[orderid].placed_date) >= self.max_adjust_time:
                        self.openorders[orderid].ba_offset = 0
                        self.logger.info("adjust_limits: %s setting ba_offset to zero because max_adjust_time as passed",self.openorders[orderid].contract.m_symbol)

                    midpoint = self._calc_midpoint(self.openorders[orderid].bid,self.openorders[orderid].ask,self.openorders[orderid].ba_offset,self.openorders[orderid].action,oderid=orderid)
                    self.logger.debug("midpoint %s", midpoint)
                    if midpoint is not None:

                        order = makeStkOrder(self.openorders[orderid].vollume, self.openorders[orderid].action, self._account, ordertype=self.openorders[orderid].ordertype)
                        order.m_lmtPrice = midpoint

                        exec_orderid = self.__MapToExecuteOrderID[orderid]
                        self.con.placeOrder(exec_orderid,contract,order)
                        self.openorders[orderid].last_adjust = datetime.datetime.today()
                        self.openorders[orderid].limitprice = order.m_lmtPrice
                        self.logger.info("adjust_limits: Updated %s order with id %s to %s (bid: %s / ask: %s)",self.openorders[orderid].contract.m_symbol, orderid, order.m_lmtPrice, self.openorders[orderid].bid, self.openorders[orderid].ask)

        #reschedule the timer
        if len(self.openorders.keys()) > 0:
            self.adjist_limits_thread = Timer(self.limit_adjust_interval, self.adjust_limits, ()).start()
        else:
            self.adjist_limits_thread = None

def buildrunning():
    if keepalive:
        threading.Timer(10.0, buildrunning).start()
        if not error:
            checkJobsBuilding()


# Initiate the threads

def repeat(self):
        try:
            self.udpHeartbeat()
        except Exception as e: 
            self.logger.exception(e)
        except:   
            self.logger.error("error on executing heartbeat: {0} ".format(sys.exc_info()[0]))           
        timer = threading.Timer(self.udpHeartbeatSeconds, self.repeat)
        timer.daemon = True
        timer.start()

def start(self):
        self.loadState()   
        self.serverHandler.start()
        sslThread = threading.Thread(target = self.startSsl)
        sslThread.daemon = True
        sslThread.start()
        timer = threading.Timer(self.taskIntervalSecond, self.repeat)
        timer.daemon = True
        timer.start()       
        self.udpServer = SocketServer.UDPServer((self.udpHost, self.udpPort), IotUDPHandler)
        self.logger.info("starting UDP server listening at {0}:{1}".format(self.udpServer.server_address[0], self.udpServer.server_address[1]))
        self.udpServer.service = self
        self.udpServer.role = IotUDPHandler.SERVER        
        self.udpServer.serve_forever()

def repeat(self):
        try:
            self.task()
        except Exception as e: 
            self.logger.exception(e)
        except:   
            self.logger.error("error on executing task: {0} ".format(sys.exc_info()[0]))           
        timer = threading.Timer(self.taskIntervalSecond, self.repeat)
        timer.daemon = True
        timer.start()

def start(self):
        self.logger.info("starting device handler")
        threading.Timer(7, self.test).start()

def on_deactivated(self, view):

        # TODO Review clearing the cmdline history, does it need to be an event?
        # Because views load asynchronously, do not restore history index
        # .on_activated(), but here instead. Otherwise, the .score_selector()
        # call won't yield the desired results.
        if view.score_selector(0, 'text.excmdline') > 0:
            view.run_command('clear_cmdline_history_index')

        self._on_deactivate_callback_timer = Timer(0.25, self._on_deactivate_callback)
        self._on_deactivate_callback_timer.start()

def start(self):
        """Start periodically sending registration messages after threshold amount of time"""
        if not self.send_messages:
            self.send_messages = True
            self.current_thread = threading.Timer(self.threshold, self._wake)
            self.current_thread.start()

def _wake(self):
        """This class periodically sends registration messages using Clog"""
        if self.send_messages:
            self.publish_registration_messages()
            # The purpose of the Timer is for _wake to ensure it is called
            self.current_thread = threading.Timer(self.threshold, self._wake)
            self.current_thread.start()

def _update_database(self):
        """
        Determines if a database update is needed and if so downloads a new one from MaxMind
        """
        if not self._download_database:
            logger.warning("_update_database called while in pre-specified database"
                           " mode... returning without updates...")
            return

        logger.info("Checking for MaxMind database updates...")
        try:
            if not self._is_update_needed():
                logger.info("No database updates to retrieve.")
                return

            logger.info("Retrieving MaxMind Database...")
            response = requests.get(self.MAXMIND_FREE_DB_URL)
            response.raise_for_status()
            logger.info('Retrieved MaxMind database.')

            data = gzip.GzipFile(fileobj=StringIO.StringIO(response.content))

            # Write the database to a temporary file
            fd, file_path = tempfile.mkstemp()
            with os.fdopen(fd, 'wb') as temp_file:
                temp_file.write(data.read())

            self._swap_database(file_path)
            logger.info("MaxMind database updated.")
        except:
            logger.exception("Failed to update MaxMind database.")
        finally:
            # Schedule this function to run again in the configured update interval
            self._update_thread = Timer(self._update_interval * 60 * 60, self._update_database)
            self._update_thread.daemon = True
            self._update_thread.start()

def run(self):
        while self.__running.isSet():
            self.__flag.wait()
            while not self.thread_stop:
                '?????????,?????????'

                try:
                    if self.queue.empty() is False:
                        __task = self.queue.get()  # ????
                        assert isinstance(__task, dict)
                        if __task['func'] != None:

                            eval(__task['func'])
                            self.queue.__task_done()  # ??????
                        else:
                            pass
                    else:
                        QA_util_log_info("From Engine %s  Engine will waiting for new task ..." % str(
                            threading.current_thread()))
                        time.sleep(1)
                except:
                    time.sleep(1)
                    self.run()
                __res = self.__QA_queue_status()  # ????????
                if __res > 0:
                    #QA_util_log_info("From Engine %s: There are still %d tasks to do" % (str(threading.current_thread()), __res))
                    pass
                # input()
                threading.Timer(0.005, self.run)

def __init__(self, *args, **kwds):
        self.logger = logging.getLogger(str(self.__class__.__name__))
        self.__class__.Timers.append(self)
        self.timer = threading.Timer(*args, **kwds)

def __getattr__(self, name):
        try:
            return super(Timer, self).__getattr(name)
        except AttributeError as e:
            if self.timer:
                return getattr(self.timer, name)
            else:
                raise