Python os 模块，sync() 实例源码

def flush(self):
        """Writes changes to flash"""
        if self.dirty:
            with open(self.filename, "wt") as file:
                file.write(json.dumps(self.data))
                file.flush()
            os.sync()
            self.dirty = False

def connect(wait=True, timeout=10, show_wait_message=False, prompt_on_fail=True, dialog_title='TiLDA'):
    retry_connect = True

    while retry_connect:
        if nic().is_connected():
            return

        details = connection_details()
        if not details:
            if prompt_on_fail:
                choose_wifi(dialog_title=dialog_title)
            else:
                raise OSError("No valid wifi configuration")

        if not wait:
            connect_wifi(details, timeout=None, wait=False)
            return
        else:
            try:
                if show_wait_message:
                    with dialogs.WaitingMessage(text="Connecting to '%s'...\n(%ss timeout)" % (details['ssid'], timeout), title=dialog_title):
                        connect_wifi(details, timeout=timeout, wait=True)
                else:
                    connect_wifi(details, timeout=timeout, wait=True)
            except OSError:
                if prompt_on_fail:
                    retry_connect = dialogs.prompt_boolean(
                        text="Failed to connect to '%s'" % details['ssid'],
                        title=dialog_title,
                        true_text="Try again",
                        false_text="Forget it",
                    )
                    if not retry_connect:
                        os.remove('wifi.json')
                        os.sync()
                        # We would rather let you choose a new network here, but
                        # scanning doesn't work after a connect at the moment
                        pyb.hard_reset()
                else:
                    raise

def download_to(self, target, timeout=90):
        start_time = time.time()
        if not self.socket:
            raise OSError("Invalid response socket state. Has the content already been consumed?")
        try:
            if "Content-Length" in self.headers:
                remaining = int(self.headers["Content-Length"])
            elif "content-length" in self.headers:
                remaining = int(self.headers["content-length"])
            else:
                raise Exception("No Content-Length")

            with open(target, 'wb') as f:
                f.write(self.content_so_far)
                remaining -= len(self.content_so_far)
                del self.content_so_far
                while remaining > 0:
                    buf = self.socket.recv(BUFFER_SIZE)
                    f.write(buf)
                    remaining -= len(buf)

                    if (time.time() - start_time) > timeout:
                        raise Exception("HTTP request timeout")

                f.flush()
            os.sync()

        finally:
            self.close()

def shutdown(self):
        '''Call the dbus proxy to start the shutdown.'''
        if self._proxy:
            os.sync()
            self._proxy(*self._args)

def setup_block_run(self, block: RunProgramBlock, runs: int = 1):
        os.sync()

def _execute(self, db: Database):
        os.sync()

def clear_cache():
    # flush file system buffers
    os.sync()
    # force kernel to drop clean caches
    try:
        subprocess.check_call("sudo sh -c 'echo 3 > /proc/sys/vm/drop_caches'",
                              stderr=subprocess.STDOUT,
                              universal_newlines=True,
                              shell=True)
    except Exception as err:
        print('Fail to drop clean caches: %s' % (err))

def choose_wifi(dialog_title='TiLDA'):
    filtered_aps = []
    with dialogs.WaitingMessage(text='Scanning for networks...', title=dialog_title):
        visible_aps = nic().list_aps()
        visible_aps.sort(key=lambda x:x['rssi'], reverse=True)
        # We'll get one result for each AP, so filter dupes
        for ap in visible_aps:
            title = ap['ssid']
            security = get_security_level(ap)
            if security:
                title = title + ' (%s)' % security
            ap = {
                'title': title,
                'ssid': ap['ssid'],
                'security': security,
            }
            if ap['ssid'] not in [ a['ssid'] for a in filtered_aps ]:
                filtered_aps.append(ap)
        del visible_aps

    ap = dialogs.prompt_option(
        filtered_aps,
        text='Choose wifi network',
        title=dialog_title
    )
    if ap:
        key = None
        if ap['security'] != 0:
            # Backward compat
            if ap['security'] == None:
                ap['security'] = 'wifi'

            key = dialogs.prompt_text(
                "Enter %s key" % ap['security'],
                width = 310,
                height = 220
            )
        with open("wifi.json", "wt") as file:
            if key:
                conn_details = {"ssid": ap['ssid'], "pw": key}
            else:
                conn_details = {"ssid": ap['ssid']}

            file.write(json.dumps(conn_details))
        os.sync()
        # We can't connect after scanning for some bizarre reason, so we reset instead
        pyb.hard_reset()

def setup_block_run(self, block: RunProgramBlock):
        num = self.misc_settings["free_pagecache"] + 2 * self.misc_settings["free_dentries_inodes"]
        self._exec_command("sync; echo {} > /proc/sys/vm/drop_caches".format(num))