0001 #!/usr/bin/env python 0002 ### BITPIM 0003 ### 0004 ### Copyright (C) 2007 Joe Pham <djpham@bitpim.org> 0005 ### 0006 ### This program is free software; you can redistribute it and/or modify 0007 ### it under the terms of the BitPim license as detailed in the LICENSE file. 0008 ### 0009 ### $Id: bp_cli.py 4599 2008-02-25 04:25:04Z djpham $ 0010 0011 """Provide Command Line Interface (CLI) functionality 0012 """ 0013 from __future__ import with_statement 0014 0015 # System modules 0016 import fnmatch 0017 import os 0018 import re 0019 0020 # BitPim modules 0021 import bp_config 0022 import phones.com_brew as com_brew 0023 import common 0024 import commport 0025 import phones 0026 0027 # Constants 0028 InvalidCommand_Error=1 0029 NotImplemented_Error=2 0030 InvalidDir_Error=3 0031 DirExists_Error=4 0032 0033 _commands=frozenset(('ls', 'll', 'cp', 'mkdir', 'cli', 'rm', 'rmdir')) 0034 wildcards=re.compile('.*[\*+|\?+]') 0035 0036 def valid_command(arg): 0037 """Check of this arg is a valid command or not 0038 @param arg: input arg, most likely passed from command line 0039 @returns: T if this is a valid command, F otherwise 0040 """ 0041 global _commands 0042 return arg.split(None)[0] in _commands 0043 0044 0045 class PhoneModelError(Exception): 0046 pass 0047 0048 class CLI(object): 0049 """BitPim Command Line Interface implementation""" 0050 0051 def __init__(self, arg, file_in, file_out, file_err, 0052 config_filename=None, comm_port=None, 0053 phone_model=None): 0054 """Constructor 0055 @param arg: command string including the command and its argument 0056 @param file_in: input stream file object 0057 @param file_out: output stream file object 0058 @param file_err: error strean file object 0059 @param config_filename: use this config file instead of the default 0060 @param comm_port: string name of the comm port to use (default to config file) 0061 @param phone_model: string phone model to use (default to config file) 0062 """ 0063 self.OK=False 0064 self._inCLI=False 0065 try: 0066 _cmd_line=arg.split(None) 0067 self.cmd=_cmd_line[0] 0068 self.args=_cmd_line[1:] 0069 self.config=bp_config.Config(config_filename) 0070 _commport=comm_port if comm_port else self.config.Read("lgvx4400port", None) 0071 _phonemodel=phone_model if phone_model else self.config.Read("phonetype", None) 0072 if os.environ.get('PHONE_FS', None): 0073 # running in BREW FS debug/sim mode 0074 self.commport=None 0075 else: 0076 self.commport=commport.CommConnection(self, _commport) 0077 try: 0078 self.phonemodule=common.importas(phones.module(_phonemodel)) 0079 except KeyError: 0080 raise PhoneModelError 0081 self.phone=self.phonemodule.Phone(self, self.commport) 0082 self._rootdir='' 0083 self._pwd=self._rootdir 0084 self._in=file_in 0085 self._out=file_out 0086 self._err=file_err 0087 # initialize the Brew file cache 0088 com_brew.file_cache=com_brew.FileCache(self.config.Read('path', '')) 0089 except common.CommsOpenFailure: 0090 file_err.write('Error: Failed to open comm port %s\n'%_commport) 0091 except PhoneModelError: 0092 file_err.write('Error: Phone Model %s not available\n'%_phonemodel) 0093 else: 0094 self.OK=True 0095 0096 _phone_prefix='phone:' 0097 _phone_prefix_len=len(_phone_prefix) 0098 def _parse_args(self, args, force_phonefs=False): 0099 """Parse the args for a list of input and out args 0100 @param args: input agrs 0101 @returns: a dict that has 2 keys: 'source' and 'dest'. 'source' has a 0102 list of files and/or directories. 'dest' is either a file or dir. 0103 """ 0104 _res=[] 0105 for _item in args: 0106 if _item.startswith(self._phone_prefix): 0107 _dirname=_item[self._phone_prefix_len:] 0108 _phonefs=True 0109 else: 0110 _dirname=_item 0111 _phonefs=force_phonefs 0112 if _phonefs: 0113 if _dirname.startswith('/'): 0114 # absolute path 0115 _dirname=self.phone.join(self._rootdir, _dirname[1:]) 0116 else: 0117 _dirname=self.phone.join(self._pwd, _dirname) 0118 # check for wildcards 0119 global wildcards 0120 _name=self.phone.basename(_dirname) 0121 if wildcards.match(_name): 0122 _dirname=self.phone.dirname(_dirname) 0123 _wildcard=_name 0124 else: 0125 _wildcard=None 0126 # correct path 0127 if _phonefs: 0128 _path=None 0129 else: 0130 _path=os.path.join(*_dirname.split('/')) 0131 _res.append({ 'name': _dirname, 'phonefs': _phonefs, 0132 'wildcard': _wildcard, 0133 'path': _path }) 0134 return _res 0135 0136 def run(self, cmdline=None): 0137 """Execute the specified command if specified, or the one 0138 currently stored 0139 @param cmdline: string command line 0140 @returns: (T, None) if the command completed successfully, 0141 (F, error code) otherwise 0142 """ 0143 if cmdline: 0144 _cmdline=cmdline.split(None) 0145 self.cmd=_cmdline[0] 0146 self.args=_cmdline[1:] 0147 _func=getattr(self, self.cmd, None) 0148 if _func is None: 0149 self._err.write('Error: invalid command: %s\n'%self.cmd) 0150 return (False, InvalidCommand_Error) 0151 return _func(self.args) 0152 0153 def log(self, logstr): 0154 pass 0155 0156 def logdata(self, logstr, logdata, dataclass=None, datatype=None): 0157 pass 0158 0159 def progress(self, pos, maxcnt, desc): 0160 "Update the progress meter" 0161 pass 0162 0163 def ls(self, args): 0164 """Do a directory listing 0165 @param args: string directory names 0166 @returns: (True, None) if successful, (False, error code) otherwise 0167 """ 0168 _src=self._parse_args(args, force_phonefs=True) 0169 if not _src: 0170 _src=[{ 'name': self._pwd, 'phonefs': True, 0171 'path': None, 'wildcard': None }] 0172 for _dir in _src: 0173 try: 0174 _dirlist=self.phone.getfilesystem(_dir['name']) 0175 self._out.write('%s:\n'%_dir['name']) 0176 for _,_file in _dirlist.items(): 0177 if _dir['wildcard'] is None or \ 0178 fnmatch.fnmatch(self.phone.basename(_file['name']), _dir['wildcard']): 0179 self._out.write('%s\n'%_file['name']) 0180 except (phones.com_brew.BrewNoSuchDirectoryException, 0181 phones.com_brew.BrewBadPathnameException): 0182 self._out.write('Error: cannot access %s: no such file or directory\n'%_dir['name']) 0183 self._out.write('\n') 0184 return (True, None) 0185 0186 def ll(self, args): 0187 """Do a long dir listing command 0188 @param args: string directory names 0189 @returns: (True, None) if successful, (False, error code) otherwise 0190 """ 0191 _src=self._parse_args(args, force_phonefs=True) 0192 if not _src: 0193 _src=[{ 'name': self._pwd, 'phonefs': True, 'path': None, 0194 'wildcard': None }] 0195 for _dir in _src: 0196 try: 0197 _dirlist=self.phone.getfilesystem(_dir['name']) 0198 self._out.write('%s:\n'%_dir['name']) 0199 _maxsize=0 0200 _maxdatelen=0 0201 for _,_file in _dirlist.items(): 0202 if _file.get('type', '')=='file': 0203 _maxsize=max(_maxsize, _file.get('size', 0)) 0204 _maxdatelen=max(_maxdatelen, len(_file.get('date', (0, ''))[1])) 0205 _formatstr='%%(dir)1s %%(size)%(maxsizelen)is %%(date)%(maxdatelen)is %%(name)s\n'%\ 0206 { 'maxdatelen': _maxdatelen, 0207 'maxsizelen': len(str(_maxsize)) } 0208 for _,_file in _dirlist.items(): 0209 if _dir['wildcard'] is None or \ 0210 fnmatch.fnmatch(self.phone.basename(_file['name']), _dir['wildcard']): 0211 _strdict={ 'name': _file['name'] } 0212 if _file['type']=='file': 0213 _strdict['dir']='' 0214 _strdict['size']=str(_file['size']) 0215 else: 0216 _strdict['dir']='d' 0217 _strdict['size']='' 0218 _strdict['date']=_file.get('date', (0, ''))[1] 0219 self._out.write(_formatstr%_strdict) 0220 except (phones.com_brew.BrewNoSuchDirectoryException, 0221 phones.com_brew.BrewBadPathnameException): 0222 self._out.write('Error: cannot access %s: no such file or directory\n'%_dir['name']) 0223 self._out.write('\n') 0224 return (True, None) 0225 0226 def _cpfilefromphone(self, filename, destdir): 0227 # copy a single file from the phone to the dest dir 0228 with open(os.path.join(destdir, self.phone.basename(filename)), 0229 'wb') as f: 0230 f.write(self.phone.getfilecontents(filename)) 0231 self._out.write('Copied file %(srcname)s to %(dirname)s\n'% { 'srcname': filename, 0232 'dirname': destdir}) 0233 def _cpdirfromphone(self, dirname, destdir, wildcard): 0234 # copy all files under a phone dir to the dest dir 0235 for _, _item in self.phone.listfiles(dirname).items(): 0236 if wildcard is None or \ 0237 fnmatch.fnmatch(self.phone.basename(_item['name']), wildcard): 0238 self._cpfilefromphone(_item['name'], destdir) 0239 0240 def _cpfromphone(self, args): 0241 # copy files from the phone 0242 _destdir=args[-1]['path'] 0243 if not os.path.isdir(_destdir): 0244 self._out.write('Error: %(dirname)s is not a valid local directory.\n'% {'dirname': _destdir }) 0245 return (False, InvalidDir_Error) 0246 for _item in args[:-1]: 0247 _name=_item['name'] 0248 if self.phone.isdir(_name): 0249 # this is a dir, cp all files under it 0250 self._cpdirfromphone(_name, _destdir, _item['wildcard']) 0251 elif self.phone.isfile(_name): 0252 # this is a file, just copy it 0253 self._cpfilefromphone(_name, _destdir) 0254 else: 0255 # not sure what it is 0256 self._out.write('Error: %(name)s does not exist\n'%{'name': _name}) 0257 return (True, None) 0258 0259 def _cpfiletophone(self, name, destdir, phonefs=False, force=False): 0260 # copy a file to the phone 0261 _filename=self.phone.join(destdir, 0262 self.phone.basename(name) if phonefs else \ 0263 os.path.basename(name)) 0264 if not force: 0265 # check if file already exists 0266 if self.phone.exists(_filename): 0267 # file exists, warn 0268 self._out.write('Phone file %(name)s exists, overwrite (y/n): '%\ 0269 { 'name': _filename }) 0270 if self._in.readline()[0].upper()!='Y': 0271 return 0272 if phonefs: 0273 # cp from phone FS to phone FS 0274 self.phone.writefile(_filename, 0275 self.phone.getfilecontents(name)) 0276 else: 0277 # local to phone FS 0278 with open(name, 'rb') as f: 0279 self.phone.writefile(_filename, 0280 f.read()) 0281 self._out.write('Copied %(filename)s to %(dirname)s.\n'%\ 0282 { 'filename': name, 0283 'dirname': destdir }) 0284 0285 def _cpdirtophone(self, dirname, destdir, phonefs=False): 0286 # cp a dir to the phone 0287 if phonefs: 0288 # phone FS dir 0289 for _, _item in self.phone.listfiles(dirname).items(): 0290 self._cpfiletophone(_item['name'], destdir, phonefs=True) 0291 else: 0292 # local dir 0293 for _item in os.listdir(dirname): 0294 if os.path.isfile(_item): 0295 self._cpfiletophone(_item, destdir, phonefs=False) 0296 0297 def _cptophone(self, args): 0298 # copy files to the phone 0299 _destdir=args[-1]['name'] 0300 if not self.phone.isdir(_destdir): 0301 self._out.write('Error: phone directory %(dirname)s is not exist.\n'%\ 0302 { 'dirname': _destdir }) 0303 return (False, InvalidDir_Error) 0304 for _item in args[:-1]: 0305 if _item['phonefs']: 0306 # this one on the phone 0307 _name=_item['name'] 0308 if self.phone.isdir(_name): 0309 self._cpdirtophone(_name, _destdir, phonefs=True) 0310 elif self.phone.isfile(_name): 0311 self._cpfiletophone(_name, _destdir, phonefs=True) 0312 else: 0313 self._out.write('Error: %(name)s does not exist.\n'%\ 0314 { 'name': _name }) 0315 else: 0316 # this one on the PC 0317 _name=_item['path'] 0318 if os.path.isdir(_name): 0319 self._cpdirtophone(_name, _destdir, phonefs=False) 0320 elif os.path.isfile(_name): 0321 self._cpfiletophone(_name, _destdir, phonefs=False) 0322 else: 0323 self._out.write('Error: %(name) does not exist.\n'%\ 0324 { 'name': _name }) 0325 return (True, None) 0326 0327 def cp(self, args): 0328 """Transfer files between the phone filesystem and local filesystem 0329 @param args: string dir names 0330 @returns: (True, None) if successful, (False, error code) otherwise 0331 """ 0332 _args=self._parse_args(args, force_phonefs=False) 0333 # The syntax of the last argument indicates the direction of the transfer 0334 # If the last arg is a phone FS: copy to the phone 0335 # If the last arg is not a phone FS: copy from the phone. 0336 if _args[-1]['phonefs']: 0337 # copy to the phone 0338 return self._cptophone(_args) 0339 else: 0340 # copy from the phone 0341 return self._cpfromphone(_args) 0342 0343 def mkdir(self, args): 0344 """Create one or more dirs on the phone FS. 0345 @param args: string dir names 0346 @returns: (True, None) if successful, (False, error code) otherwise 0347 """ 0348 _src=self._parse_args(args, force_phonefs=True) 0349 for _dir in _src: 0350 try: 0351 self.phone.mkdir(_dir['name']) 0352 except phones.com_brew.BrewDirectoryExistsException: 0353 self._out.write('Error: dir %(name)s exists.\n'% \ 0354 { 'name': _dir['name'] }) 0355 except phones.com_brew.BrewNoSuchDirectoryException: 0356 self._out.write('Error: Failed to create dir %(name)s.\n'%\ 0357 { 'name': _dir['name'] }) 0358 return (True, None) 0359 0360 def cli(self, _): 0361 """Run a primitive interactive CLI sesssion. 0362 @params _: don't care 0363 @returns: always (True, None) 0364 """ 0365 if self._inCLI: 0366 # we're in our shell, bail 0367 return (True, None) 0368 self._inCLI=True 0369 ## try: 0370 while True: 0371 self._out.write('BitPim>') 0372 _cmdline=self._in.readline() 0373 if _cmdline.startswith('exit'): 0374 break 0375 self.run(_cmdline) 0376 ## finally: 0377 self._inCLI=False 0378 return (True, None) 0379 0380 def cd(self, args): 0381 """Change the current working dir on the phone FS. 0382 @param args: dir name(s), only args[0] matters. 0383 @returns: (True, None) if successful, (False, error code) otherwise 0384 """ 0385 _dirs=self._parse_args(args, force_phonefs=True) 0386 if _dirs: 0387 _dirname=_dirs[0]['name'] 0388 if _dirname=='/': 0389 _dirname=self._rootdir 0390 if self.phone.exists(_dirname): 0391 self._pwd=_dirname 0392 else: 0393 self._out.write('Invalid dir: %s\n'%_dirname) 0394 return self.pwd(args) 0395 def cdu(self, args): 0396 """Change to current working dir on the phone up one level (parent). 0397 @param _: don't care 0398 @returns: (True, None) if successful, (False, error code) otherwise 0399 """ 0400 _updir=self.phone.dirname(self._pwd) 0401 if _updir: 0402 self._pwd=_updir 0403 return self.pwd(args) 0404 0405 def pwd(self, _): 0406 """Print the current working dir(cwd/pwd) on the phone FS. 0407 @params _: don't care 0408 @returns: (True, None) 0409 """ 0410 self._out.write("%(name)s\n"%{ 'name': self._pwd }) 0411 return (True, None) 0412 cwd=pwd 0413 0414 def rm(self, args): 0415 """delete one or more files on the phone FS. This command does not 0416 delete local files. 0417 @param args: file names 0418 @returns: (True, None) if successful, (False, error code) otherwise. 0419 """ 0420 _filenames=self._parse_args(args, force_phonefs=True) 0421 for _item in _filenames: 0422 _name=_item['name'] 0423 if self.phone.isfile(_name): 0424 self.phone.rmfile(_name) 0425 self._out.write('File %(name)s deleted\n'%{ 'name': _name }) 0426 else: 0427 self._out.write('Invalid file: %(name)s\n'%{ 'name': _name }) 0428 0429 def rmdir(self, args): 0430 """Delete one or more phone FS directories. This command does not 0431 delete local directories. 0432 @param args: dir names. 0433 @returns: (True, None) if successful, (False, error code) otherwise. 0434 """ 0435 _dirnames=self._parse_args(args, force_phonefs=True) 0436 for _item in _dirnames: 0437 _name=_item['name'] 0438 if self.phone.isdir(_name): 0439 # this is a dir, check for empty 0440 if self.phone.hassubdirs(_name) or \ 0441 self.phone.listfiles(_name): 0442 self._out.write('Dir %(name)s is not empty.\n'%{'name': _name }) 0443 else: 0444 self.phone.rmdirs(_name) 0445 self._out.write('Dir %(name)s deleted.\n'% { 'name': _name }) 0446 else: 0447 self._out.write('Invalid dir: %(name)s\n'%{ 'name': _name}) 0448
Generated by PyXR 0.9.4