Python genericpath 模块，exists() 实例源码

def declaration_path(name):
    """Return the path to an included declaration"""
    from os.path import dirname, join, exists
    import  metatabdecl
    from metatab.exc import IncludeError

    d = dirname(metatabdecl.__file__)

    path = join(d, name)

    if not exists(path):
        path = join(d, name + '.csv')

    if not exists(path):
        raise IncludeError("No local declaration file for name '{}' ".format(name))

    return path


# From http://stackoverflow.com/a/295466

def make_dir_structure(base_dir):
    """Make the build directory structure. """

    def maybe_makedir(*args):

        p = join(base_dir, *args)

        if exists(p) and not isdir(p):
            raise IOError("File '{}' exists but is not a directory ".format(p))

        if not exists(p):
            makedirs(p)

    maybe_makedir(DOWNLOAD_DIR)
    maybe_makedir(PACKAGE_DIR)
    maybe_makedir(OLD_DIR)

def check_atlas(atlas):
    """Validation of atlas input."""

    # when its a name for pre-defined atlas
    if isinstance(atlas, str):
        if not pexists(atlas): # just a name
            atlas = atlas.lower()
            if atlas not in parcellate.atlas_list:
                raise ValueError('Invalid choice of atlas. Accepted : {}'.format(parcellate.atlas_list))
        elif os.path.isdir(atlas): # cortical atlas in Freesurfer org
            if not parcellate.check_atlas_annot_exist(atlas):
                raise ValueError('Given atlas folder does not contain Freesurfer label annot files. '
                                 'Needed : given_atlas_dir/label/?h.aparc.annot')
        elif pexists(atlas): # may be a volumetric atlas?
            try:
                atlas = nibabel.load(atlas)
            except:
                traceback.print_exc()
                raise ValueError('Unable to read the provided image volume. Must be a nifti 2d volume, readable by nibabel.')
        else:
            raise ValueError('Unable to decipher or use the given atlas.')
    else:
        raise NotImplementedError('Atlas must be a string, providing a name or path to Freesurfer folder or a 3D nifti volume.')

    return atlas

def make_output_path_graph(out_dir, subject, str_prefixes):
    "Constructs path to save a multigraph to disk."

    if out_dir is not None:
        # get outpath returned from hiwenet, based on dist name and all other parameters
        # choose out_dir name  based on dist name and all other parameters
        out_subject_dir = pjoin(out_dir, subject)
        if not pexists(out_subject_dir):
            os.mkdir(out_subject_dir)

        if isinstance(str_prefixes, str):
            str_prefixes = [str_prefixes, ]

        out_file_name = '{}_graynet.graphml'.format('_'.join(str_prefixes))
        out_weights_path = pjoin(out_subject_dir, out_file_name)
    else:
        out_weights_path = None

    return out_weights_path

def check_subjects(subjects_info):
    "Ensure subjects are provided and their data exist."

    if isinstance(subjects_info, str):
        if not pexists(subjects_info):
            raise IOError('path to subject list does not exist: {}'.format(subjects_info))
        subjects_list = np.genfromtxt(subjects_info, dtype=str)
    elif isinstance(subjects_info, collections.Iterable):
        if len(subjects_info) < 1:
            raise ValueError('Empty subject list.')
        subjects_list = subjects_info
    else:
        raise ValueError('Invalid value provided for subject list. \n '
                         'Must be a list of paths, or path to file containing list of paths, one for each subject.')

    subject_id_list = np.atleast_1d(subjects_list)
    num_subjects = subject_id_list.size
    if num_subjects < 1:
        raise ValueError('Input subject list is empty.')

    num_digits_id_size = len(str(num_subjects))
    max_id_width = max(map(len, subject_id_list))

    return subject_id_list, num_subjects, max_id_width, num_digits_id_size

def check_params_single_edge(base_features, in_dir, atlas, smoothing_param, node_size, out_dir, return_results):
    """"""

    check_features(base_features)

    if atlas.lower() not in parcellate.atlas_list:
        raise ValueError('Invalid atlas choice. Use one of {}'.format(parcellate.atlas_list))

    if not pexists(in_dir):
        raise IOError('Input directory at {} does not exist.'.format(in_dir))

    if out_dir is None and return_results is False:
        raise ValueError('Results are neither saved to disk or being received when returned.\n'
                         'Specify out_dir (not None) or make return_results=True')

    if out_dir is not None and not pexists(out_dir):
        os.mkdir(out_dir)

    # no checks on subdivison size yet, as its not implemented

    return

def check_params_multiedge(base_feature_list, input_dir, atlas, smoothing_param,
                           node_size, out_dir, return_results):
    """Validation of parameters and appropriate type casting if necessary."""

    check_features(base_feature_list)

    if atlas.lower() not in parcellate.atlas_list:
        raise ValueError('Invalid atlas choice. Use one of {}'.format(parcellate.atlas_list))

    if not pexists(input_dir):
        raise IOError('Input directory at {} does not exist.'.format(input_dir))

    if out_dir is None and return_results is False:
        raise ValueError('Results are neither saved to disk or being received when returned.\n'
                         'Specify out_dir (not None) or make return_results=True')

    if out_dir is not None and not pexists(out_dir):
        os.mkdir(out_dir)

    # no checks on subdivison size yet, as its not implemented

    return

def add_image(self, base_path, file):
        if not os.path.exists(file):
            file = os.path.join(base_path, file)
        texture = Image(source=file).texture
        print('loading', file)
        texture.mag_filter = 'nearest'
        if texture is None:
            sys.exit('failed to locate image file %r' % file)

        id = self.firstgid
        th = self.tile_height + self.spacing
        tw = self.tile_width + self.spacing
        for j in range(texture.height // th):
            for i in range(texture.width // tw):
                x = (i * tw) + self.margin
                # convert the y coordinate to OpenGL (0 at bottom of texture)
                y = texture.height - ((j + 1) * th)
                tile = texture.get_region(x, y, self.tile_width, self.tile_height)
                self.tiles.append(Tile(id, tile, self))
                id += 1

def save(config, is_system=False):
        """
        Save configuration ``config``.
        """
        ConfigurationManager.update(config)
        location = SYSTEM_CONFIG if is_system else RUNTIME_CONFIG
        try:
            LOG.info("Saving config: " + location)
            dir = location.replace("/jarbas_runtime.conf", "").replace(
                "/jarbas.conf", "")
            if not exists(dir):
                mkdir(dir)
            try:
                loc_config = load_commented_json(location)
            except:
                loc_config = {}

            with open(location, 'w') as f:
                config = loc_config.update(config)
                json.dump(config, f)
        except Exception as e:
            LOG.error(e)

def new_metatab_file(mt_file, template):
    template = template if template else 'metatab'

    if not exists(mt_file):
        doc = make_metatab_file(template)

        doc.write_csv(mt_file)

        return True

    else:

        return False

def ensure_dir(path):
    if path and not exists(path):
        makedirs(path)

def get_eval_dir(self):
        answer_dir = join(self.dir, "answers")
        if not exists(answer_dir):
            os.mkdir(answer_dir)
        return answer_dir

def get_best_weights(self):
        if exists(self.best_weight_dir):
            return tf.train.latest_checkpoint(self.best_weight_dir)
        return None

def restore_checkpoint(self, sess, var_list=None, load_ema=True):
        """
        Restores either the best weights or the most recent checkpoint, assuming the correct
        variables have already been added to the tf default graph e.g., .get_prediction()
        has been called the model stored in `self`.
        Automatically detects if EMA weights exists, and if they do loads them instead
        """
        checkpoint = self.get_best_weights()
        if checkpoint is None:
            print("Loading most recent checkpoint")
            checkpoint = self.get_latest_checkpoint()
        else:
            print("Loading best weights")

        if load_ema:
            if var_list is None:
                # Same default used by `Saver`
                var_list = tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES) + \
                           tf.get_collection(tf.GraphKeys.SAVEABLE_OBJECTS)

            # Automatically check if there are EMA variables, if so use those
            reader = tf.train.NewCheckpointReader(checkpoint)
            ema = tf.train.ExponentialMovingAverage(0)
            ema_names = {ema.average_name(x): x for x in var_list
                         if reader.has_tensor(ema.average_name(x))}
            if len(ema_names) > 0:
                print("Found EMA weights, loading them")
                ema_vars = set(x for x in ema_names.values())
                var_list = {v.op.name: v for v in var_list if v not in ema_vars}
                var_list.update(ema_names)

        saver = tf.train.Saver(var_list)
        saver.restore(sess, checkpoint)

def reseed(self):
        # this will only work if app has permission to modify seed.db, for example NOT IOS.
        # PS) hopefully it's obvious that this will only effect new games based off of new seed.db.
        # all_tmx_files_to_db will just update seed.db if it exists, but for now, just wipe the slate clean to be sure
        if exists('seed.db'):
            print('deleting seed.db, existing...')
            os.remove('seed.db')
        all_tmx_files_to_db()
        Notification(message='Reseed complete!').open()

def fromxml(cls, tag, tilemap, firstgid=None):
        if 'source' in tag.attrib:
            firstgid = int(tag.attrib['firstgid'])
            path = tag.attrib['source']
            if not os.path.exists(path):
                path = os.path.join(tilemap.file_path, path)
            with open(path) as f:
                tileset = ElementTree.fromstring(f.read())
            return cls.fromxml(tileset, firstgid)

        name = tag.attrib['name']
        if firstgid is None:
            firstgid = int(tag.attrib['firstgid'])
        tile_width = int(tag.attrib['tilewidth'])
        tile_height = int(tag.attrib['tileheight'])
        spacing = int(tag.get('spacing', 0))
        margin = int(tag.get('margin', 0))

        tileset = cls(name, tile_width, tile_height, firstgid,
                      spacing, margin)

        for c in tag.getchildren():
            if c.tag == "image":
                # create a tileset
                tileset.add_image(tilemap.file_path, c.attrib['source'])
            elif c.tag == 'tile':
                gid = tileset.firstgid + int(c.attrib['id'])
                tileset.get_tile(gid).loadxml(c)
        return tileset

def _create_if_not_exists(self):
        if exists(self._file_path):
            return
        dist_file = join(dirname(__file__), '..', 'resources', 'config-dist.json')
        copy(dist_file, self._file_path)

def __load(config, location):
        if exists(location) and isfile(location):
            try:
                ConfigurationLoader.merge_conf(
                    config, load_commented_json(location))
                LOG.debug("Configuration '%s' loaded" % location)
            except Exception, e:
                LOG.error("Error loading configuration '%s'" % location)
                LOG.error(repr(e))
        else:
            LOG.debug("Configuration '%s' not found" % location)
        return config

def download(self, url):

        working_dir = self.working_dir if self.working_dir else ''

        r = Resource()

        # For local files, don't download, just reference in place.
        if url.scheme == 'file':
            r.cache_path = Url(url.resource_url).path
            r.download_time = None

            # Many places the file may exist
            locations = {  # What a mess ...
                abspath(r.cache_path),
                abspath(r.cache_path.lstrip('/')),
                abspath(join(working_dir, r.cache_path)),
                abspath(r.cache_path.lstrip('/'))
            }

            for l in locations:
                if exists(l):
                    r.sys_path = l
                    break
            else:
                raise DownloadError(("File resource does not exist. Found none of:"
                                     "\n{}\n\nWorking dir = {}\ncache_path={}\nspec_path={}")
                                    .format('\n'.join(locations), working_dir, r.cache_path, url.path))

        else:
            # Not a local file, so actually need to download it.
            try:
                r.cache_path, r.download_time = self._download_with_lock(url.resource_url)
            except AccessError as e:
                # Try again, using a URL that we may have configured an account for. This is
                # primarily S3 urls, with Boto or AWS credential
                try:
                    r.cache_path, r.download_time = self._download_with_lock(url.auth_resource_url)
                except AttributeError:
                    raise e

            r.sys_path = self.cache.getsyspath(r.cache_path)

        return r

def _download(self, url, cache_path):
        import requests

        def copy_callback(read, total):
            if self.callback:
                self.callback('copy_file', read, total)

        if self.callback:
            self.callback('download', url, 0)

        if url.startswith('s3:'):

            from appurl.url import Url

            s3url = Url(url)

            try:
                with self.cache.open(cache_path, 'wb') as f:
                    s3url.object.download_fileobj(f)
            except Exception as e:
                raise DownloadError("Failed to fetch S3 url '{}': {}".format(url, e))

        elif url.startswith('ftp:'):
            from contextlib import closing

            with closing(urlopen(url)) as fin:

                with self.cache.open(cache_path, 'wb') as fout:

                    read_len = 16 * 1024
                    total_len = 0
                    while 1:
                        buf = fin.read(read_len)
                        if not buf:
                            break
                        fout.write(buf)
                        total_len += len(buf)

                        if self.callback:
                            copy_callback(len(buf), total_len)

        else:

            try:
                r = requests.get(url, stream=True)
                r.raise_for_status()
            except SSLError as e:
                raise DownloadError("Failed to GET {}: {} ".format(url, e))

            # Requests will auto decode gzip responses, but not when streaming. This following
            # monkey patch is recommended by a core developer at
            # https://github.com/kennethreitz/requests/issues/2155
            if r.headers.get('content-encoding') == 'gzip':
                r.raw.read = functools.partial(r.raw.read, decode_content=True)

            with self.cache.open(cache_path, 'wb') as f:
                copy_file_or_flo(r.raw, f, cb=copy_callback)

            assert self.cache.exists(cache_path)