Source code for openmdao.main.datatypes.file

"""
Support for file variables.
"""

import copy
import os.path

# pylint: disable-msg=E0611,F0401
from openmdao.main.file_supp import FileMetadata, RemoteFile
from openmdao.main.rbac import rbac
from openmdao.main.variable import Variable

#Public Symbols
__all__ = ['File', 'FileRef']


[docs]class File(Variable): """ A trait wrapper for a :class:`FileRef` object. If `default_value` is a string, then a :class:`FileRef` will be created using that for the path. If `default_value` is a file object, then a :class:`FileRef` will be created for the file's name, if the named file exists. For input files :attr:`legal_types` may be set to a list of expected 'content_type' strings. Then upon assignment the actual 'content_type' must match one of the :attr:`legal_types` strings. Also, if :attr:`local_path` is set, then upon assignent the associated file will be copied to that path. """ def __init__(self, default_value=None, iotype=None, **metadata): if default_value is not None: if isinstance(default_value, FileRef): pass elif isinstance(default_value, basestring): default_value = FileRef(default_value) elif isinstance(default_value, file) and \ hasattr(default_value, 'name') and \ os.path.exists(default_value.name): default_value = FileRef(default_value.name) else: raise TypeError('File default value must be a FileRef.') if iotype is not None: metadata['iotype'] = iotype if iotype == 'out': if 'legal_types' in metadata: raise ValueError("'legal_types' invalid for output File.") if 'local_path' in metadata: raise ValueError("'local_path' invalid for output File.") # iotype of None => we can't check anything. super(File, self).__init__(default_value, **metadata) # It appears this scheme won't pickle, requiring a hack in Container... # def get_default_value(self): # """ Return (default_value_type, default_value). """ # return (8, self.make_default) # # def make_default(self, obj): # """ Make a default value for obj. """ # iotype = self._metadata['iotype'] # if iotype == 'out': # default = self.default_value.copy(obj) # else: # default = None # return default
[docs] def validate(self, obj, name, value): """ Verify that `value` is a FileRef of a legal type. """ if value is None: return value if isinstance(value, basestring): value = FileRef(value) elif isinstance(value, file) and hasattr(value, 'name') and \ os.path.exists(value.name): value = FileRef(value.name) if isinstance(value, FileRef): legal_types = self._metadata.get('legal_types', None) if legal_types: if value.content_type not in legal_types: raise ValueError("Content type '%s' not one of %s" % (value.content_type, legal_types)) return value else: self.error(obj, name, value)
[docs] def post_setattr(self, obj, name, value): """ If 'local_path' is set on an input, then copy the source FileRef's file to that path. """ if value is None: return iotype = self._metadata.get('iotype') if iotype == 'out': if value.owner is None: value.owner = obj return path = self._metadata.get('local_path', None) if not path: return owner = _get_valid_owner(obj) if os.path.isabs(path): if owner is None: raise ValueError('local_path %s is absolute and no path checker' ' is available.' % path) owner.check_path(path) else: if owner is None: raise ValueError('local_path %s is relative and no absolute' ' directory is available.' % path) directory = owner.get_abs_directory() path = os.path.join(directory, path) # If accessing same path on same host (i.e. passthrough), skip. if isinstance(value, FileRef): try: src_path = value.abspath() except Exception as exc: raise RuntimeError("Can't get source path for local copy: %s" % (str(exc) or repr(exc))) else: if path == src_path: return mode = 'wb' if value.binary else 'w' try: src = value.open() except Exception as exc: raise RuntimeError("Can't open source for local copy: %s" % (str(exc) or repr(exc))) try: dst = open(path, mode) except Exception as exc: src.close() raise RuntimeError("Can't open destination for local copy: %s" % (str(exc) or repr(exc))) chunk = 1 << 20 # 1MB data = src.read(chunk) while data: dst.write(data) data = src.read(chunk) src.close() dst.close()
[docs]class FileRef(FileMetadata): """ A reference to a file on disk. As well as containing metadata information, it supports :meth:`open` to read the file's contents. Before :meth:`open` is called, 'owner' must be set to an object supporting :meth:`check_path` and :meth:`get_abs_directory` (typically a :class:`Component` or one of its child :class:`Container` objects). """ def __init__(self, path, owner=None, **metadata): super(FileRef, self).__init__(path, **metadata) self.owner = owner
[docs] def copy(self, owner): """ Return a copy of ourselves, owned by `owner`. owner: Component The component used to determine the root for relative paths and checking the legality of absolute paths. """ ref = copy.copy(self) ref.owner = owner return ref
[docs] def abspath(self): """ Return absolute path to file. """ if self.owner is None: raise ValueError("abspath() failed: no owner specified for FileRef") path = self.path if os.path.isabs(path): try: self.owner.check_path(path) except AttributeError: owner = _get_valid_owner(self.owner) if owner is None: raise ValueError("Path '%s' is absolute and no path checker" " is available." % path) self.owner = owner self.owner.check_path(path) else: try: directory = self.owner.get_abs_directory() except AttributeError: owner = _get_valid_owner(self.owner) if owner is None: raise ValueError("Path '%s' is relative and no absolute" " directory is available." % path) self.owner = owner directory = self.owner.get_abs_directory() path = os.path.join(directory, path) return path
@rbac('owner', proxy_types=[RemoteFile])
[docs] def open(self): """ Open file for reading. """ mode = 'rb' if self.binary else 'rU' return RemoteFile(open(self.abspath(), mode))
def _get_valid_owner(owner): """ Try to find an owner that supports the required functionality. """ while owner is not None: if hasattr(owner, 'check_path') and \ hasattr(owner, 'get_abs_directory'): return owner if hasattr(owner, 'parent'): owner = owner.parent else: return None return None
OpenMDAO Home