<s> """<STR_LIT>""" <EOL> from __future__ import generators <EOL> import sys , os , stat , re , atexit <EOL> import py <EOL> from py . __ . path import common <EOL> iswin32 = sys . platform == "<STR_LIT:win32>" <EOL> if iswin32 : <EOL> from py . __ . path . local . win import WinMixin as PlatformMixin <EOL> else : <EOL> from py . __ . path . local . posix import PosixMixin as PlatformMixin <EOL> class LocalPath ( common . FSPathBase , PlatformMixin ) : <EOL> """<STR_LIT>""" <EOL> _path_cache = { } <EOL> sep = os . sep <EOL> class Checkers ( common . FSCheckers ) : <EOL> def _stat ( self ) : <EOL> try : <EOL> return self . _statcache <EOL> except AttributeError : <EOL> try : <EOL> self . _statcache = self . path . stat ( ) <EOL> except py . error . ELOOP : <EOL> self . _statcache = self . path . lstat ( ) <EOL> return self . _statcache <EOL> def dir ( self ) : <EOL> return stat . S_ISDIR ( self . _stat ( ) . mode ) <EOL> def file ( self ) : <EOL> return stat . S_ISREG ( self . _stat ( ) . mode ) <EOL> def exists ( self ) : <EOL> return self . _stat ( ) <EOL> def link ( self ) : <EOL> st = self . path . lstat ( ) <EOL> return stat . S_ISLNK ( st . mode ) <EOL> def __new__ ( cls , path = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( path , common . FSPathBase ) : <EOL> if path . __class__ == cls : <EOL> return path <EOL> path = path . strpath <EOL> self = object . __new__ ( cls ) <EOL> if not path : <EOL> self . strpath = os . getcwd ( ) <EOL> elif isinstance ( path , str ) : <EOL> self . strpath = os . path . abspath ( os . path . normpath ( str ( path ) ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> assert isinstance ( self . strpath , str ) <EOL> return self <EOL> def __hash__ ( self ) : <EOL> return hash ( self . strpath ) <EOL> def computehash ( self , hashtype = "<STR_LIT>" , chunksize = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> hash = self . _gethashinstance ( hashtype ) <EOL> f = self . open ( '<STR_LIT:rb>' ) <EOL> try : <EOL> while <NUM_LIT:1> : <EOL> buf = f . read ( chunksize ) <EOL> if not buf : <EOL> return hash . hexdigest ( ) <EOL> hash . update ( buf ) <EOL> finally : <EOL> f . close ( ) <EOL> def new ( self , ** kw ) : <EOL> """<STR_LIT>""" <EOL> obj = object . __new__ ( self . __class__ ) <EOL> drive , dirname , basename , purebasename , ext = self . _getbyspec ( <EOL> "<STR_LIT>" ) <EOL> if '<STR_LIT>' in kw : <EOL> if '<STR_LIT>' in kw or '<STR_LIT>' in kw : <EOL> raise ValueError ( "<STR_LIT>" % kw ) <EOL> else : <EOL> pb = kw . setdefault ( '<STR_LIT>' , purebasename ) <EOL> try : <EOL> ext = kw [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> if ext and not ext . startswith ( '<STR_LIT:.>' ) : <EOL> ext = '<STR_LIT:.>' + ext <EOL> kw [ '<STR_LIT>' ] = pb + ext <EOL> kw . setdefault ( '<STR_LIT>' , drive ) <EOL> kw . setdefault ( '<STR_LIT>' , dirname ) <EOL> kw . setdefault ( '<STR_LIT>' , self . sep ) <EOL> obj . strpath = os . path . normpath ( <EOL> "<STR_LIT>" % kw ) <EOL> return obj <EOL> def _getbyspec ( self , spec ) : <EOL> """<STR_LIT>""" <EOL> res = [ ] <EOL> parts = self . strpath . split ( self . sep ) <EOL> args = filter ( None , spec . split ( '<STR_LIT:U+002C>' ) ) <EOL> append = res . append <EOL> for name in args : <EOL> if name == '<STR_LIT>' : <EOL> append ( parts [ <NUM_LIT:0> ] ) <EOL> elif name == '<STR_LIT>' : <EOL> append ( self . sep . join ( [ '<STR_LIT>' ] + parts [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) ) <EOL> else : <EOL> basename = parts [ - <NUM_LIT:1> ] <EOL> if name == '<STR_LIT>' : <EOL> append ( basename ) <EOL> else : <EOL> i = basename . rfind ( '<STR_LIT:.>' ) <EOL> if i == - <NUM_LIT:1> : <EOL> purebasename , ext = basename , '<STR_LIT>' <EOL> else : <EOL> purebasename , ext = basename [ : i ] , basename [ i : ] <EOL> if name == '<STR_LIT>' : <EOL> append ( purebasename ) <EOL> elif name == '<STR_LIT>' : <EOL> append ( ext ) <EOL> else : <EOL> raise ValueError , "<STR_LIT>" % name <EOL> return res <EOL> def join ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not args : <EOL> return self <EOL> strpath = self . strpath <EOL> sep = self . sep <EOL> strargs = [ str ( x ) for x in args ] <EOL> if kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) : <EOL> for i in range ( len ( strargs ) - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> ) : <EOL> if os . path . isabs ( strargs [ i ] ) : <EOL> strpath = strargs [ i ] <EOL> strargs = strargs [ i + <NUM_LIT:1> : ] <EOL> break <EOL> for arg in strargs : <EOL> arg = arg . strip ( sep ) <EOL> if iswin32 : <EOL> arg = arg . strip ( '<STR_LIT:/>' ) <EOL> arg = arg . replace ( '<STR_LIT:/>' , sep ) <EOL> if arg : <EOL> if not strpath . endswith ( sep ) : <EOL> strpath += sep <EOL> strpath += arg <EOL> obj = self . new ( ) <EOL> obj . strpath = os . path . normpath ( strpath ) <EOL> return obj <EOL> def __eq__ ( self , other ) : <EOL> s1 = str ( self ) <EOL> s2 = str ( other ) <EOL> if iswin32 : <EOL> s1 = s1 . lower ( ) <EOL> s2 = s2 . lower ( ) <EOL> return s1 == s2 <EOL> def open ( self , mode = '<STR_LIT:r>' ) : <EOL> """<STR_LIT>""" <EOL> return self . _callex ( open , self . strpath , mode ) <EOL> def listdir ( self , fil = None , sort = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( fil , str ) : <EOL> fil = common . fnmatch ( fil ) <EOL> res = [ ] <EOL> for name in self . _callex ( os . listdir , self . strpath ) : <EOL> childurl = self . join ( name ) <EOL> if fil is None or fil ( childurl ) : <EOL> res . append ( childurl ) <EOL> if callable ( sort ) : <EOL> res . sort ( sort ) <EOL> elif sort : <EOL> res . sort ( ) <EOL> return res <EOL> def size ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . stat ( ) . size <EOL> def mtime ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . stat ( ) . mtime <EOL> def copy ( self , target , archive = False ) : <EOL> """<STR_LIT>""" <EOL> assert not archive , "<STR_LIT>" <EOL> if self . check ( file = <NUM_LIT:1> ) : <EOL> if target . check ( dir = <NUM_LIT:1> ) : <EOL> target = target . join ( self . basename ) <EOL> assert self != target <EOL> copychunked ( self , target ) <EOL> else : <EOL> target . ensure ( dir = <NUM_LIT:1> ) <EOL> def rec ( p ) : <EOL> return p . check ( link = <NUM_LIT:0> ) <EOL> for x in self . visit ( rec = rec ) : <EOL> relpath = x . relto ( self ) <EOL> newx = target . join ( relpath ) <EOL> if x . check ( link = <NUM_LIT:1> ) : <EOL> newx . mksymlinkto ( x . readlink ( ) ) <EOL> elif x . check ( file = <NUM_LIT:1> ) : <EOL> copychunked ( x , newx ) <EOL> elif x . check ( dir = <NUM_LIT:1> ) : <EOL> newx . ensure ( dir = <NUM_LIT:1> ) <EOL> def rename ( self , target ) : <EOL> """<STR_LIT>""" <EOL> return self . _callex ( os . rename , str ( self ) , str ( target ) ) <EOL> def dump ( self , obj , bin = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> f = self . open ( '<STR_LIT:wb>' ) <EOL> try : <EOL> self . _callex ( py . std . cPickle . dump , obj , f , bin ) <EOL> finally : <EOL> f . close ( ) <EOL> def mkdir ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> p = self . join ( * args ) <EOL> self . _callex ( os . mkdir , str ( p ) ) <EOL> return p <EOL> def write ( self , content , mode = '<STR_LIT:wb>' ) : <EOL> """<STR_LIT>""" <EOL> s = str ( content ) <EOL> f = self . open ( mode ) <EOL> try : <EOL> f . write ( s ) <EOL> finally : <EOL> f . close ( ) <EOL> def _ensuredirs ( self ) : <EOL> parent = self . dirpath ( ) <EOL> if parent == self : <EOL> return self <EOL> if parent . check ( dir = <NUM_LIT:0> ) : <EOL> parent . _ensuredirs ( ) <EOL> if self . check ( dir = <NUM_LIT:0> ) : <EOL> try : <EOL> self . mkdir ( ) <EOL> except py . error . EEXIST : <EOL> if self . check ( dir = <NUM_LIT:0> ) : <EOL> raise <EOL> return self <EOL> def ensure ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> p = self . join ( * args ) <EOL> if kwargs . get ( '<STR_LIT>' , <NUM_LIT:0> ) : <EOL> return p . _ensuredirs ( ) <EOL> else : <EOL> p . dirpath ( ) . _ensuredirs ( ) <EOL> if not p . check ( file = <NUM_LIT:1> ) : <EOL> p . write ( "<STR_LIT>" ) <EOL> return p <EOL> def stat ( self ) : <EOL> """<STR_LIT>""" <EOL> stat = self . _callex ( os . stat , self . strpath ) <EOL> return self . _makestat ( stat ) <EOL> def lstat ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _makestat ( self . _callex ( os . lstat , self . strpath ) ) <EOL> def setmtime ( self , mtime = None ) : <EOL> """<STR_LIT>""" <EOL> if mtime is None : <EOL> return self . _callex ( os . utime , self . strpath , mtime ) <EOL> try : <EOL> return self . _callex ( os . utime , self . strpath , ( - <NUM_LIT:1> , mtime ) ) <EOL> except py . error . EINVAL : <EOL> return self . _callex ( os . utime , self . strpath , ( self . atime ( ) , mtime ) ) <EOL> def chdir ( self ) : <EOL> """<STR_LIT>""" <EOL> old = self . __class__ ( ) <EOL> self . _callex ( os . chdir , self . strpath ) <EOL> return old <EOL> def realpath ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __class__ ( os . path . realpath ( self . strpath ) ) <EOL> def atime ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . stat ( ) . atime <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . strpath <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . strpath <EOL> def pypkgpath ( self , pkgname = None ) : <EOL> """<STR_LIT>""" <EOL> pkgpath = None <EOL> for parent in self . parts ( reverse = True ) : <EOL> if pkgname is None : <EOL> if parent . check ( file = <NUM_LIT:1> ) : <EOL> continue <EOL> if parent . join ( '<STR_LIT>' ) . check ( ) : <EOL> pkgpath = parent <EOL> continue <EOL> return pkgpath <EOL> else : <EOL> if parent . basename == pkgname : <EOL> return parent <EOL> return pkgpath <EOL> def _prependsyspath ( self , path ) : <EOL> s = str ( path ) <EOL> if s != sys . path [ <NUM_LIT:0> ] : <EOL> sys . path . insert ( <NUM_LIT:0> , s ) <EOL> def pyimport ( self , modname = None , ensuresyspath = True ) : <EOL> """<STR_LIT>""" <EOL> if not self . check ( ) : <EOL> raise py . error . ENOENT ( self ) <EOL> pkgpath = None <EOL> if modname is None : <EOL> pkgpath = self . pypkgpath ( ) <EOL> if pkgpath is not None : <EOL> if ensuresyspath : <EOL> self . _prependsyspath ( pkgpath . dirpath ( ) ) <EOL> pkg = __import__ ( pkgpath . basename , None , None , [ ] ) <EOL> if hasattr ( pkg , '<STR_LIT>' ) : <EOL> modname = pkg . __pkg__ . getimportname ( self ) <EOL> assert modname is not None , "<STR_LIT>" % ( <EOL> pkg . __name__ , self ) <EOL> else : <EOL> names = self . new ( ext = '<STR_LIT>' ) . relto ( pkgpath . dirpath ( ) ) <EOL> names = names . split ( self . sep ) <EOL> modname = "<STR_LIT:.>" . join ( names ) <EOL> else : <EOL> if ensuresyspath : <EOL> self . _prependsyspath ( self . dirpath ( ) ) <EOL> modname = self . purebasename <EOL> mod = __import__ ( modname , None , None , [ '<STR_LIT>' ] ) <EOL> return mod <EOL> else : <EOL> try : <EOL> return sys . modules [ modname ] <EOL> except KeyError : <EOL> mod = py . std . new . module ( modname ) <EOL> mod . __file__ = str ( self ) <EOL> sys . modules [ modname ] = mod <EOL> try : <EOL> execfile ( str ( self ) , mod . __dict__ ) <EOL> except : <EOL> del sys . modules [ modname ] <EOL> raise <EOL> return mod <EOL> def _getpymodule ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . ext != '<STR_LIT>' : <EOL> return super ( LocalPath , self ) . _getpymodule ( ) <EOL> from py . __ . misc . buildcmodule import make_module_from_c <EOL> mod = make_module_from_c ( self ) <EOL> return mod <EOL> def _getpycodeobj ( self ) : <EOL> """<STR_LIT>""" <EOL> dotpy = self . check ( ext = '<STR_LIT>' ) <EOL> if dotpy : <EOL> my_magic = py . std . imp . get_magic ( ) <EOL> my_timestamp = int ( self . mtime ( ) ) <EOL> if __debug__ : <EOL> pycfile = self + '<STR_LIT:c>' <EOL> else : <EOL> pycfile = self + '<STR_LIT:o>' <EOL> try : <EOL> f = pycfile . open ( '<STR_LIT:rb>' ) <EOL> try : <EOL> header = f . read ( <NUM_LIT:8> ) <EOL> if len ( header ) == <NUM_LIT:8> : <EOL> magic , timestamp = py . std . struct . unpack ( '<STR_LIT>' , header ) <EOL> if magic == my_magic and timestamp == my_timestamp : <EOL> co = py . std . marshal . load ( f ) <EOL> path1 = co . co_filename <EOL> path2 = str ( self ) <EOL> if path1 == path2 : <EOL> return co <EOL> try : <EOL> if os . path . samefile ( path1 , path2 ) : <EOL> return co <EOL> except ( OSError , <EOL> AttributeError ) : <EOL> pass <EOL> finally : <EOL> f . close ( ) <EOL> except py . error . Error : <EOL> pass <EOL> s = self . read ( mode = '<STR_LIT>' ) + '<STR_LIT:\n>' <EOL> codeobj = compile ( s , str ( self ) , '<STR_LIT>' , generators . compiler_flag ) <EOL> if dotpy : <EOL> try : <EOL> f = pycfile . open ( '<STR_LIT:wb>' ) <EOL> f . write ( py . std . struct . pack ( '<STR_LIT>' , '<STR_LIT>' , - <NUM_LIT:1> ) ) <EOL> py . std . marshal . dump ( codeobj , f ) <EOL> f . flush ( ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> f . write ( py . std . struct . pack ( '<STR_LIT>' , my_magic , my_timestamp ) ) <EOL> f . close ( ) <EOL> except py . error . Error : <EOL> pass <EOL> return codeobj <EOL> def sysexec ( self , * argv ) : <EOL> """<STR_LIT>""" <EOL> from py . compat . subprocess import Popen , PIPE <EOL> argv = map ( str , argv ) <EOL> proc = Popen ( [ str ( self ) ] + list ( argv ) , stdout = PIPE , stderr = PIPE ) <EOL> stdout , stderr = proc . communicate ( ) <EOL> ret = proc . wait ( ) <EOL> if ret != <NUM_LIT:0> : <EOL> raise py . process . cmdexec . Error ( ret , ret , str ( self ) , <EOL> stdout , stderr , ) <EOL> return stdout <EOL> def sysfind ( cls , name , checker = None ) : <EOL> """<STR_LIT>""" <EOL> if os . path . isabs ( name ) : <EOL> p = py . path . local ( name ) <EOL> if p . check ( file = <NUM_LIT:1> ) : <EOL> return p <EOL> else : <EOL> if iswin32 : <EOL> paths = py . std . os . environ [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) <EOL> if '<STR_LIT>' not in paths and '<STR_LIT:.>' not in paths : <EOL> paths . append ( '<STR_LIT:.>' ) <EOL> try : <EOL> systemroot = os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> paths = [ re . sub ( '<STR_LIT>' , systemroot , path ) <EOL> for path in paths ] <EOL> tryadd = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> else : <EOL> paths = py . std . os . environ [ '<STR_LIT>' ] . split ( '<STR_LIT::>' ) <EOL> tryadd = ( '<STR_LIT>' , ) <EOL> for x in paths : <EOL> for addext in tryadd : <EOL> p = py . path . local ( x ) . join ( name , abs = True ) + addext <EOL> try : <EOL> if p . check ( file = <NUM_LIT:1> ) : <EOL> if checker : <EOL> if not checker ( p ) : <EOL> continue <EOL> return p <EOL> except py . error . EACCES : <EOL> pass <EOL> return None <EOL> sysfind = classmethod ( sysfind ) <EOL> def _gethomedir ( cls ) : <EOL> try : <EOL> x = os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> x = os . environ [ '<STR_LIT>' ] <EOL> return cls ( x ) <EOL> _gethomedir = classmethod ( _gethomedir ) <EOL> def get_temproot ( cls ) : <EOL> """<STR_LIT>""" <EOL> return py . path . local ( py . std . tempfile . gettempdir ( ) ) <EOL> get_temproot = classmethod ( get_temproot ) <EOL> def mkdtemp ( cls ) : <EOL> """<STR_LIT>""" <EOL> import tempfile <EOL> tries = <NUM_LIT:10> <EOL> for i in range ( tries ) : <EOL> dname = tempfile . mktemp ( ) <EOL> dpath = cls ( tempfile . mktemp ( ) ) <EOL> try : <EOL> dpath . mkdir ( ) <EOL> except ( py . error . EEXIST , py . error . EPERM , py . error . EACCES ) : <EOL> continue <EOL> return dpath <EOL> raise py . error . ENOENT ( dpath , "<STR_LIT>" % tries ) <EOL> mkdtemp = classmethod ( mkdtemp ) <EOL> def make_numbered_dir ( cls , prefix = '<STR_LIT>' , rootdir = None , keep = <NUM_LIT:3> , <EOL> lock_timeout = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> if rootdir is None : <EOL> rootdir = cls . get_temproot ( ) <EOL> def parse_num ( path ) : <EOL> """<STR_LIT>""" <EOL> bn = path . basename <EOL> if bn . startswith ( prefix ) : <EOL> try : <EOL> return int ( bn [ len ( prefix ) : ] ) <EOL> except ValueError : <EOL> pass <EOL> lastmax = None <EOL> while True : <EOL> maxnum = - <NUM_LIT:1> <EOL> for path in rootdir . listdir ( ) : <EOL> num = parse_num ( path ) <EOL> if num is not None : <EOL> maxnum = max ( maxnum , num ) <EOL> try : <EOL> udir = rootdir . mkdir ( prefix + str ( maxnum + <NUM_LIT:1> ) ) <EOL> except py . error . EEXIST : <EOL> if lastmax == maxnum : <EOL> raise <EOL> lastmax = maxnum <EOL> continue <EOL> break <EOL> if lock_timeout : <EOL> lockfile = udir . join ( '<STR_LIT>' ) <EOL> mypid = os . getpid ( ) <EOL> if hasattr ( lockfile , '<STR_LIT>' ) : <EOL> lockfile . mksymlinkto ( str ( mypid ) ) <EOL> else : <EOL> lockfile . write ( str ( mypid ) ) <EOL> def try_remove_lockfile ( ) : <EOL> if os . getpid ( ) != mypid : <EOL> return <EOL> try : <EOL> lockfile . remove ( ) <EOL> except py . error . Error : <EOL> pass <EOL> atexit . register ( try_remove_lockfile ) <EOL> if keep : <EOL> for path in rootdir . listdir ( ) : <EOL> num = parse_num ( path ) <EOL> if num is not None and num <= ( maxnum - keep ) : <EOL> lf = path . join ( '<STR_LIT>' ) <EOL> try : <EOL> t1 = lf . lstat ( ) . mtime <EOL> t2 = lockfile . lstat ( ) . mtime <EOL> if not lock_timeout or abs ( t2 - t1 ) < lock_timeout : <EOL> continue <EOL> except py . error . Error : <EOL> pass <EOL> try : <EOL> path . remove ( rec = <NUM_LIT:1> ) <EOL> except KeyboardInterrupt : <EOL> raise <EOL> except : <EOL> pass <EOL> try : <EOL> username = os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> try : <EOL> username = os . environ [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> username = '<STR_LIT>' <EOL> src = str ( udir ) <EOL> dest = src [ : src . rfind ( '<STR_LIT:->' ) ] + '<STR_LIT:->' + username <EOL> try : <EOL> os . unlink ( dest ) <EOL> except OSError : <EOL> pass <EOL> try : <EOL> os . symlink ( src , dest ) <EOL> except ( OSError , AttributeError ) : <EOL> pass <EOL> return udir <EOL> make_numbered_dir = classmethod ( make_numbered_dir ) <EOL> def copychunked ( src , dest ) : <EOL> chunksize = <NUM_LIT> <EOL> fsrc = src . open ( '<STR_LIT:rb>' ) <EOL> try : <EOL> fdest = dest . open ( '<STR_LIT:wb>' ) <EOL> try : <EOL> while <NUM_LIT:1> : <EOL> buf = fsrc . read ( chunksize ) <EOL> if not buf : <EOL> break <EOL> fdest . write ( buf ) <EOL> finally : <EOL> fdest . close ( ) <EOL> finally : <EOL> fsrc . close ( ) </s>
