class ImageManager():
    def __init__(self):
        # Image data
        self._frame = None
        # Flow control
        self._condition = Condition()

    def timestamp(self, img):
        now = datetime.datetime.now()
        stamp.stamp(img, (10, 10), str(now), size=20)

        return img

    def update_frame(self, frame):
        self._frame = BytesIO(frame)
        self.ready = True

    @property
    def frame(self):
        return self._frame

    @property
    def ready(self):
        return self._condition

    @ready.setter
    def ready(self, cond):
        if cond is True:
            self._condition.notify_all()

    def get_data(cls, account, source_filter, limit=100, skip=0):
        """
        Gathers card information from Google Sheets
        GET https://spreadsheets.google.com/feeds/list/[spreadsheet]/[worksheet]/private/full
        """
        if not account or not account.enabled:
            raise ValueError('cannot gather information without an account')
        client = AsyncHTTPClient()

        if source_filter.spreadsheet is None:
            raise ValueError('required parameter spreadsheet missing')
        if source_filter.worksheet is None:
            raise ValueError('required parameter worksheet missing')
        uri = "https://docs.google.com/spreadsheets/d/{}/export?format=csv&gid={}".format(
            source_filter.spreadsheet, source_filter.worksheet
        )

        app_log.info(
            "Start retrieval of worksheet {}/{} for {}".format(source_filter.spreadsheet, source_filter.worksheet,
                                                               account._id))

        lock = Condition()
        oauth_client = account.get_client()
        uri, headers, body = oauth_client.add_token(uri)
        req = HTTPRequest(uri, headers=headers, body=body, streaming_callback=lambda c: cls.write(c))

        client.fetch(req, callback=lambda r: lock.notify())
        yield lock.wait(timeout=timedelta(seconds=MAXIMUM_REQ_TIME))

        app_log.info(
            "Finished retrieving worksheet for {}".format(account._id))

class Window(object):
    def __init__(self, parent, stream_id, initial_window_size):
        self.parent = parent
        self.stream_id = stream_id
        self.cond = Condition()
        self.closed = False
        self.size = initial_window_size

    def close(self):
        self.closed = True
        self.cond.notify_all()

    def _raise_error(self, code, message):
        if self.parent is None:
            raise ConnectionError(code, message)
        else:
            raise StreamError(self.stream_id, code)

    def adjust(self, amount):
        self.size += amount
        if self.size > constants.MAX_WINDOW_SIZE:
            self._raise_error(constants.ErrorCode.FLOW_CONTROL_ERROR,
                              "flow control window too large")
        self.cond.notify_all()

    def apply_window_update(self, frame):
        try:
            window_update, = struct.unpack('>I', frame.data)
        except struct.error:
            raise ConnectionError(constants.ErrorCode.FRAME_SIZE_ERROR,
                                  "WINDOW_UPDATE incorrect size")
        # strip reserved bit
        window_update = window_update & 0x7fffffff
        if window_update == 0:
            self._raise_error(constants.ErrorCode.PROTOCOL_ERROR,
                              "window update must not be zero")
        self.adjust(window_update)

    @gen.coroutine
    def consume(self, amount):
        while not self.closed and self.size <= 0:
            yield self.cond.wait()
        if self.closed:
            raise StreamClosedError()
        if self.size < amount:
            amount = self.size
        if self.parent is not None:
            amount = yield self.parent.consume(amount)
        self.size -= amount
        raise gen.Return(amount)

    def clear(self):
        """Reset this PeerGroup.

        This closes all connections to all known peers and forgets about
        these peers.

        :returns:
            A Future that resolves with a value of None when the operation
            has finished
        """
        if self._resetting:
            # If someone else is already resetting the PeerGroup, just block
            # on them to be finished.
            yield self._reset_condition.wait()
            raise gen.Return(None)

        self._resetting = True
        if self._reset_condition is None:
            self._reset_condition = Condition()

        try:
            for peer in self._peers.values():
                peer.close()
        finally:
            self._peers = {}
            self._resetting = False
            self._reset_condition.notify_all()

    def __init__ (self, device_server, stream, address):
        self.fw_version = 0.0
        self.recv_msg_cond = Condition()
        self.recv_msg = {}
        self.send_msg_sem = Semaphore(1)
        self.pending_request_cnt = 0
        self.device_server = device_server
        self.stream = stream
        self.address = address
        self.stream.set_nodelay(True)
        self.timeout_handler_onlinecheck = None
        self.timeout_handler_offline = None
        self.killed = False
        self.sn = ""
        self.private_key = ""
        self.node_id = 0
        self.iv = None
        self.cipher_down = None
        self.cipher_up = None

        #self.state_waiters = []
        #self.state_happened = []

        self.event_waiters = []
        self.event_happened = []

        self.ota_ing = False
        self.ota_notify_done_future = None
        self.post_ota = False
        self.online_status = True

 def __init__(self, session, home):
     self.scrapper = Scrapper(session, home)
     self.session = session
     self.home = home
     self.condition = Condition()
     tornado.ioloop.IOLoop.current()\
         .spawn_callback(self.main)

 def __init__(self, application, request, **kwargs):
     super(PingHandler, self).__init__(application, request, **kwargs)
     self.callback_queue = None
     self.condition = Condition()
     self.response = None
     self.corr_id = str(uuid.uuid4())
     self.in_channel = self.application.get_app_component().rabbitmq[
         'client'].channels['in']

class PingHandler(firenado.tornadoweb.TornadoHandler):

    def __init__(self, application, request, **kwargs):
        super(PingHandler, self).__init__(application, request, **kwargs)
        self.callback_queue = None
        self.condition = Condition()
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.in_channel = self.application.get_app_component().rabbitmq[
            'client'].channels['in']

    @gen.coroutine
    def post(self):
        self.in_channel.queue_declare(exclusive=True,
                                      callback=self.on_request_queue_declared)
        yield self.condition.wait()

        self.write(self.response)

    def on_request_queue_declared(self, response):
        logger.info('Request temporary queue declared.')
        self.callback_queue = response.method.queue
        self.in_channel.basic_consume(self.on_response, no_ack=True,
                                      queue=self.callback_queue)
        self.in_channel.basic_publish(
            exchange='',
            routing_key='ping_rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,
            ),
            body=self.request.body)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = {
                'data': body.decode("utf-8"),
                'date': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            }
            self.in_channel.queue_delete(queue=self.callback_queue)
            self.condition.notify()

    def test_future_close_callback(self):
        # Regression test for interaction between the Future read interfaces
        # and IOStream._maybe_add_error_listener.
        rs, ws = yield self.make_iostream_pair()
        closed = [False]
        cond = Condition()

        def close_callback():
            closed[0] = True
            cond.notify()
        rs.set_close_callback(close_callback)
        try:
            ws.write(b'a')
            res = yield rs.read_bytes(1)
            self.assertEqual(res, b'a')
            self.assertFalse(closed[0])
            ws.close()
            yield cond.wait()
            self.assertTrue(closed[0])
        finally:
            rs.close()
            ws.close()

    def __init__(self, buf=None, auto_close=True):
        """In-Memory based stream

        :param buf: the buffer for the in memory stream
        """
        self._stream = deque()
        if buf:
            self._stream.append(buf)
        self.state = StreamState.init
        self._condition = Condition()
        self.auto_close = auto_close

        self.exception = None

class GameScrapperWorker(object):

    def __init__(self, session, home):
        self.scrapper = Scrapper(session, home)
        self.session = session
        self.home = home
        self.condition = Condition()
        tornado.ioloop.IOLoop.current()\
            .spawn_callback(self.main)

    @tornado.gen.coroutine
    def main(self):
        _log().info('scrapper sleeping')
        yield self.condition.wait()
        _log().info('scrapper woke up')
        self.scrapper.scrap_missing()
        tornado.ioloop.IOLoop.current()\
            .spawn_callback(self.main)
        raise tornado.gen.Return(False)

    def __init__(self, tchannel, score_threshold=None):
        """Initializes a new PeerGroup.

        :param tchannel:
            TChannel used for communication by this PeerGroup
        :param score_threshold:
            A value in the ``[0, 1]`` range. If specifiede, this requires that
            chosen peers havea score higher than this value when performing
            requests.
        """
        self.tchannel = tchannel

        self._score_threshold = score_threshold

        # Dictionary from hostport to Peer.
        self._peers = {}

        # Notified when a reset is performed. This allows multiple coroutines
        # to block on the same reset.
        self._resetting = False
        self._reset_condition = Condition()

class ImporterWorker(object):

    def __init__(self, session, home, scrapper):
        self.session = session
        self.home = home
        self.condition = Condition()
        self.scrapper = scrapper
        tornado.ioloop.IOLoop.current()\
            .spawn_callback(self.main)

    @tornado.gen.coroutine
    def main(self):
        _log().info('importer sleeping')
        yield self.condition.wait()
        _log().info('importer woke up')
        changed = True
        while changed:
            changed = yield self.work()
        tornado.ioloop.IOLoop.current()\
            .spawn_callback(self.main)

    @tornado.gen.coroutine
    def work(self):
        changed = False
        fi = slickbird.FileImporter(self.session, self.home)
        for f in self.session.query(orm.Importerfile)\
                .filter(orm.Importerfile.status == 'scanning'):
            changed = True
            r, status = fi.file_import(f.filename)
            f.status = status
            if status == 'moved':
                self.scrapper.condition.notify()
            yield tornado.gen.moment
        self.session.commit()
        self.scrapper.condition.notify()
        raise tornado.gen.Return(changed)

 def __init__(self):
     # Image data
     self._frame = None
     # Flow control
     self._condition = Condition()

class DeviceConnection(object):

    state_waiters = {}
    state_happened = {}

    def __init__ (self, device_server, stream, address):
        self.fw_version = 0.0
        self.recv_msg_cond = Condition()
        self.recv_msg = {}
        self.send_msg_sem = Semaphore(1)
        self.pending_request_cnt = 0
        self.device_server = device_server
        self.stream = stream
        self.address = address
        self.stream.set_nodelay(True)
        self.timeout_handler_onlinecheck = None
        self.timeout_handler_offline = None
        self.killed = False
        self.sn = ""
        self.private_key = ""
        self.node_id = 0
        self.iv = None
        self.cipher_down = None
        self.cipher_up = None

        #self.state_waiters = []
        #self.state_happened = []

        self.event_waiters = []
        self.event_happened = []

        self.ota_ing = False
        self.ota_notify_done_future = None
        self.post_ota = False
        self.online_status = True

    @gen.coroutine
    def secure_write (self, data):
        if self.cipher_down:
            cipher_text = self.cipher_down.encrypt(pad(data))
            yield self.stream.write(cipher_text)

    @gen.coroutine
    def wait_hello (self):
        try:
            self._wait_hello_future = self.stream.read_bytes(64) #read 64bytes: 32bytes SN + 32bytes signature signed with private key
            str1 = yield gen.with_timeout(timedelta(seconds=10), self._wait_hello_future,
                                         io_loop=ioloop.IOLoop.current())
            self.idle_time = 0  #reset the idle time counter

            if len(str1) != 64:
                self.stream.write("sorry\r\n")
                yield gen.sleep(0.1)
                self.kill_myself()
                gen_log.debug("receive length != 64")
                raise gen.Return(100) # length not match 64

            if re.match(r'@\d\.\d', str1[0:4]):
                #new version firmware
                self._wait_hello_future = self.stream.read_bytes(4) #read another 4bytes
                str2 = yield gen.with_timeout(timedelta(seconds=10), self._wait_hello_future, io_loop=ioloop.IOLoop.current())

                self.idle_time = 0  #reset the idle time counter

                if len(str2) != 4:
                    self.stream.write("sorry\r\n")
                    yield gen.sleep(0.1)
                    self.kill_myself()
                    gen_log.debug("receive length != 68")
                    raise gen.Return(100) # length not match 64

                str1 += str2
                self.fw_version = float(str1[1:4])
                sn = str1[4:36]
                sig = str1[36:68]
            else:
                #for version < 1.1
                sn = str1[0:32]
                sig = str1[32:64]

            gen_log.info("accepted sn: %s @fw_version %.1f" % (sn, self.fw_version))

            #query the sn from database
            node = None
            cur = self.device_server.cur
            cur.execute('select * from nodes where node_sn="%s"'%sn)
            rows = cur.fetchall()
            if len(rows) > 0:
                node = rows[0]

            if not node:
                self.stream.write("sorry\r\n")
                yield gen.sleep(0.1)
                self.kill_myself()
                gen_log.info("node sn not found")
                raise gen.Return(101) #node not found

            key = node['private_key']
            key = key.encode("ascii")

#.........这里部分代码省略.........

class DeviceConnection(object):

    state_waiters = {}
    state_happened = {}

    def __init__ (self, device_server, stream, address):
        self.recv_msg_cond = Condition()
        self.recv_msg = {}
        self.send_msg_sem = Semaphore(1)
        self.pending_request_cnt = 0
        self.device_server = device_server
        self.stream = stream
        self.address = address
        self.stream.set_nodelay(True)
        self.idle_time = 0;
        self.killed = False
        self.sn = ""
        self.private_key = ""
        self.node_id = 0
        self.name = ""
        self.iv = None
        self.cipher = None

        #self.state_waiters = []
        #self.state_happened = []

        self.event_waiters = []
        self.event_happened = []

        self.ota_ing = False
        self.ota_notify_done_future = None
        self.post_ota = False
        self.online_status = True

    @gen.coroutine
    def secure_write (self, data):
        if self.cipher:
            cipher_text = self.cipher.encrypt(pad(data))
            yield self.stream.write(cipher_text)

    @gen.coroutine
    def wait_hello (self):
        try:
            self._wait_hello_future = self.stream.read_bytes(64) #read 64bytes: 32bytes SN + 32bytes signature signed with private key
            str = yield gen.with_timeout(timedelta(seconds=10), self._wait_hello_future, io_loop=self.stream.io_loop)
            self.idle_time = 0  #reset the idle time counter

            if len(str) != 64:
                self.stream.write("sorry\r\n")
                yield gen.sleep(0.1)
                self.kill_myself()
                gen_log.debug("receive length != 64")
                raise gen.Return(100) # length not match 64

            sn = str[0:32]
            sig = str[32:64]

            gen_log.info("accepted sn: "+ sn)

            #query the sn from database
            node = None
            for n in NODES_DATABASE:
                if n['node_sn'] == sn:
                    node = n
                    break

            if not node:
                self.stream.write("sorry\r\n")
                yield gen.sleep(0.1)
                self.kill_myself()
                gen_log.info("node sn not found")
                raise gen.Return(101) #node not found

            key = node['node_key']
            key = key.encode("ascii")

            sig0 = hmac.new(key, msg=sn, digestmod=hashlib.sha256).digest()
            gen_log.debug("sig:     "+ binascii.hexlify(sig))
            gen_log.debug("sig calc:"+ binascii.hexlify(sig0))

            if sig0 == sig:
                #send IV + AES Key
                self.sn = sn
                self.private_key = key
                self.node_id = node['node_sn']
                self.name = node['name']
                gen_log.info("valid hello packet from node %s" % self.name)
                #remove the junk connection of the same sn
                self.stream.io_loop.add_callback(self.device_server.remove_junk_connection, self)
                #init aes
                self.iv = Random.new().read(AES.block_size)
                self.cipher = AES.new(key, AES.MODE_CFB, self.iv, segment_size=128)
                cipher_text = self.iv + self.cipher.encrypt(pad("hello"))
                gen_log.debug("cipher text: "+ cipher_text.encode('hex'))
                self.stream.write(cipher_text)
                raise gen.Return(0)
            else:
                self.stream.write("sorry\r\n")
                yield gen.sleep(0.1)
                self.kill_myself()
#.........这里部分代码省略.........

 def __init__(self, parent, stream_id, initial_window_size):
     self.parent = parent
     self.stream_id = stream_id
     self.cond = Condition()
     self.closed = False
     self.size = initial_window_size

class PeerGroup(object):
    """A PeerGroup represents a collection of Peers.

    Requests routed through a PeerGroup can be sent to either a specific peer
    or a peer chosen at random.
    """

    def __init__(self, tchannel, score_threshold=None):
        """Initializes a new PeerGroup.

        :param tchannel:
            TChannel used for communication by this PeerGroup
        :param score_threshold:
            A value in the ``[0, 1]`` range. If specifiede, this requires that
            chosen peers havea score higher than this value when performing
            requests.
        """
        self.tchannel = tchannel

        self._score_threshold = score_threshold

        # Dictionary from hostport to Peer.
        self._peers = {}

        # Notified when a reset is performed. This allows multiple coroutines
        # to block on the same reset.
        self._resetting = False

        # We'll create a Condition here later. We want to avoid it right now
        # because it has a side-effect of scheduling some dummy work on the
        # ioloop, which prevents us from forking (if you're into that).
        self._reset_condition = None

    def __str__(self):
        return "<PeerGroup peers=%s>" % str(self._peers)

    @gen.coroutine
    def clear(self):
        """Reset this PeerGroup.

        This closes all connections to all known peers and forgets about
        these peers.

        :returns:
            A Future that resolves with a value of None when the operation
            has finished
        """
        if self._resetting:
            # If someone else is already resetting the PeerGroup, just block
            # on them to be finished.
            yield self._reset_condition.wait()
            raise gen.Return(None)

        self._resetting = True
        if self._reset_condition is None:
            self._reset_condition = Condition()

        try:
            for peer in self._peers.values():
                peer.close()
        finally:
            self._peers = {}
            self._resetting = False
            self._reset_condition.notify_all()

    def get(self, hostport):
        """Get a Peer for the given destination.

        A new Peer is added and returned if one does not already exist for the
        given host-port. Otherwise, the existing Peer is returned.
        """
        assert hostport, "hostport is required"
        if hostport not in self._peers:
            self._peers[hostport] = Peer(self.tchannel, hostport)
        return self._peers[hostport]

    def lookup(self, hostport):
        """Look up a Peer for the given host and port.

        Returns None if a Peer for the given host-port does not exist.
        """
        assert hostport, "hostport is required"
        return self._peers.get(hostport, None)

    def remove(self, hostport):
        """Delete the Peer for the given host port.

        Does nothing if a matching Peer does not exist.

        :returns: The removed Peer
        """
        assert hostport, "hostport is required"
        return self._peers.pop(hostport, None)

    def add(self, peer):
        """Add an existing Peer to this group.

        A peer for the given host-port must not already exist in the group.
        """
        assert peer, "peer is required"
#.........这里部分代码省略.........

class InMemStream(Stream):

    def __init__(self, buf=None, auto_close=True):
        """In-Memory based stream

        :param buf: the buffer for the in memory stream
        """
        self._stream = deque()
        if buf:
            self._stream.append(buf)
        self.state = StreamState.init
        self._condition = Condition()
        self.auto_close = auto_close

        self.exception = None

    def clone(self):
        new_stream = InMemStream()
        new_stream.state = self.state
        new_stream.auto_close = self.auto_close
        new_stream._stream = deque(self._stream)
        return new_stream

    def read(self):

        def read_chunk(future):
            if self.exception:
                future.set_exception(self.exception)
                return future

            chunk = ""

            while len(self._stream) and len(chunk) < common.MAX_PAYLOAD_SIZE:
                chunk += self._stream.popleft()

            future.set_result(chunk)
            return future

        read_future = tornado.concurrent.Future()

        # We're not ready yet
        if self.state != StreamState.completed and not len(self._stream):
            wait_future = self._condition.wait()
            wait_future.add_done_callback(
                lambda f: f.exception() or read_chunk(read_future)
            )
            return read_future

        return read_chunk(read_future)

    def write(self, chunk):
        if self.exception:
            raise self.exception

        if self.state == StreamState.completed:
            raise UnexpectedError("Stream has been closed.")

        if chunk:
            self._stream.append(chunk)
            self._condition.notify()

        # This needs to return a future to match the async interface.
        r = tornado.concurrent.Future()
        r.set_result(None)
        return r

    def set_exception(self, exception):
        self.exception = exception
        self.close()

    def close(self):
        self.state = StreamState.completed
        self._condition.notify()

    def get_data(cls, account, source_filter, limit=100, skip=0):
        source_filter = OneDriveFileFilter(source_filter)

        if source_filter.file is None:
            raise ValueError("required parameter file missing")

        app_log.info("Starting to retrieve file for {}".format(account._id))

        client = AsyncHTTPClient()
        uri = "https://api.onedrive.com/v1.0/drive/items/{}/content".format(source_filter.file)
        lock = Condition()

        def crawl_url(url):
            # some yummy regex
            location_header_regex = re.compile(r"^Location:\s?(?P<uri>http:/{2}\S+)")
            http_status_regex = re.compile(r"^HTTP/[\d\.]+\s(?P<status>\d+)")
            receiving_file = False

            # define our callbacks
            def header_callback(header):
                m = http_status_regex.match(header)
                if m is not None:
                    # process our HTTP status header
                    status = m.group("status")
                    if int(status) == 200:
                        # if we're 200, we're receiving the file, not just a redirect
                        app_log.info("Receiving file {} for account {}".format(source_filter.file, account._id))
                        global receiving_file
                        receiving_file = True
                m = location_header_regex.match(header)
                if m is not None:
                    # process our location header
                    uri = m.group("uri")
                    # and grab _that_ url
                    app_log.info("Following redirect for file {}".format(source_filter.file))
                    crawl_url(uri)

            def stream_callback(chunk):
                # only dump out chunks that are of the file we're looking for
                global receiving_file
                if receiving_file:
                    app_log.info("Writing chunk of {}B".format(chunk.__len__()))
                    cls.write(chunk)

            def on_completed(resp):
                if 200 <= resp.code <= 299:
                    lock.notify()

            oauth_client = account.get_client()
            uri, headers, body = oauth_client.add_token(url)
            req = HTTPRequest(
                uri, headers=headers, body=body, header_callback=header_callback, streaming_callback=stream_callback
            )
            client.fetch(req, callback=on_completed)

        crawl_url(uri)
        # wait for us to complete
        try:
            yield lock.wait(timeout=timedelta(seconds=MAXIMUM_REQ_TIME))
            app_log.info("File {} retrieved successfully".format(source_filter.file))
        except gen.TimeoutError:
            app_log.error("Request for file {} => {} timed out!".format(source_filter.file, account._id))