Source code for idmtools.assets.file_list

idmtools FileList classes.

Copyright 2021, Bill & Melinda Gates Foundation. All rights reserved.
from typing import List
import os
import re
from idmtools.assets import Asset, AssetCollection
from idmtools.utils.local_os import LocalOS

[docs]class FileList: """ Special utility class to help handling user files. """
[docs] def __init__(self, root=None, files_in_root=None, recursive=False, ignore_missing=False, relative_path=None, max_depth=3): """ Represents a set of files that are specified RELATIVE to root. e.g. /a/b/c.json could be : root: '/a' files_in_root: ['b/c.json'] :param root: The dir all files_in_root are relative to. :param files_in_root: The listed files """ self.files: List[Asset] = [] self.ignore_missing = ignore_missing self.max_depth = max_depth # Make sure we have correct separator # os.path.normpath(f) would be best but is not working the same way on UNIX systems if files_in_root is not None: if LocalOS.is_window(): files_in_root = [os.path.normpath(f) for f in files_in_root] else: files_in_root = [re.sub(r"[\\/]", os.sep, os.path.normpath(f)) for f in files_in_root] if root: self.add_path(path=root, files_in_dir=files_in_root, recursive=recursive, relative_path=relative_path)
[docs] def add_asset_file(self, af): """ Method used to add asset file. Args: af: asset file to add Returns: None """ self.files.append(af)
[docs] def add_file(self, path, relative_path=''): """ Method used to add a file. Args: path: file oath relative_path: file relative path Returns: None """ # If already present -> bypass for f in self.files: if f.absolute_path and os.path.abspath(f.absolute_path) == os.path.abspath(path): return if os.path.isdir(path): raise ValueError("%s is a directory. add_file is expecting a file!" % path) absolute_path = os.path.abspath(path) file_name = os.path.basename(path) try: af = Asset(filename=file_name, relative_path=relative_path, absolute_path=absolute_path) self.add_asset_file(af) except Exception as e: if not self.ignore_missing: raise e
[docs] def add_path(self, path, files_in_dir=None, relative_path=None, recursive=False): """ Add a path to the file list. Args: path: The path to add (needs to be a dictionary) files_in_dir: If we want to only retrieve certain files in this path relative_path: relative_path: The relative path prefixed to each added files recursive: Do we want to browse recursively Returns: None """ # Little safety if not os.path.isdir(path) and not path.startswith('\\\\'): raise RuntimeError("add_path() requires a directory. '%s' is not." % path) if not recursive: if files_in_dir is None: files_in_dir = [f for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))] for file_name in files_in_dir: file_path = os.path.join(path, file_name) f_relative_path = os.path.normpath( file_path.replace(path, '').strip(os.sep).replace(os.path.basename(file_path), '')) if relative_path is not None: f_relative_path = os.path.join(relative_path, f_relative_path) self.add_file(file_path, relative_path=f_relative_path) else: # Walk through the path for root, _subdirs, files in os.walk(path): # Little safety to not go too deep depth = root[len(path) + len(os.path.sep):].count(os.path.sep) if depth > self.max_depth: continue # Add the files in the current dir for f in files: # Find the file relative path compared to the root folder # If the relative_path is . -> change it into '' f_relative_path = os.path.normpath(os.path.relpath(root, path)) if f_relative_path == '.': f_relative_path = '' # if files_in_dir specified -> skip the ones not included if files_in_dir is not None and f not in files_in_dir and os.path.join(f_relative_path, f) not in files_in_dir: continue # if we want to force a relative path -> force it if relative_path is not None: f_relative_path = os.path.join(relative_path, f_relative_path) # add the file self.add_file(os.path.join(root, f), relative_path=f_relative_path)
[docs] def to_asset_collection(self) -> AssetCollection: """ Convert a file list to an asset collection. Returns: AssetCollection version of filelist """ ac = AssetCollection() ac.assets = self.files return ac
[docs] @staticmethod def from_asset_collection(asset_collection: AssetCollection) -> 'FileList': """ Create a FileList from a AssetCollection. Args: asset_collection: AssetCollection to convert. Returns: FileList version of AssetCollection """ fl = FileList() fl.files = [asset_collection.assets] return fl
def __len__(self): """ Length of FileList. Returns: Length """ return len(self.files) def __iter__(self): """ Iterator. Basically we proxy self.files. Returns: Iterator """ return self.files.__iter__() def __getitem__(self, item): """ Get file. Args: item: Item Returns: Item """ return self.files.__getitem__(item)