bp_cli

49 """BitPim Command Line Interface implementation""" 50

51 - def __init__(self, arg, file_in, file_out, file_err, 52 config_filename=None, comm_port=None, 53 phone_model=None):

54 """Constructor 55 @param arg: command string including the command and its argument 56 @param file_in: input stream file object 57 @param file_out: output stream file object 58 @param file_err: error strean file object 59 @param config_filename: use this config file instead of the default 60 @param comm_port: string name of the comm port to use (default to config file) 61 @param phone_model: string phone model to use (default to config file) 62 """ 63 self.OK=False 64 self._inCLI=False 65 try: 66 _cmd_line=arg.split(None) 67 self.cmd=_cmd_line[0] 68 self.args=_cmd_line[1:] 69 self.config=bp_config.Config(config_filename) 70 _commport=comm_port if comm_port else self.config.Read("lgvx4400port", None) 71 _phonemodel=phone_model if phone_model else self.config.Read("phonetype", None) 72 if os.environ.get('PHONE_FS', None): 73 # running in BREW FS debug/sim mode 74 self.commport=None 75 else: 76 self.commport=commport.CommConnection(self, _commport) 77 try: 78 self.phonemodule=common.importas(phones.module(_phonemodel)) 79 except KeyError: 80 raise PhoneModelError 81 self.phone=self.phonemodule.Phone(self, self.commport) 82 self._rootdir='' 83 self._pwd=self._rootdir 84 self._in=file_in 85 self._out=file_out 86 self._err=file_err 87 # initialize the Brew file cache 88 com_brew.file_cache=com_brew.FileCache(self.config.Read('path', '')) 89 except common.CommsOpenFailure: 90 file_err.write('Error: Failed to open comm port %s\n'%_commport) 91 except PhoneModelError: 92 file_err.write('Error: Phone Model %s not available\n'%_phonemodel) 93 else: 94 self.OK=True

95 96 _phone_prefix='phone:' 97 _phone_prefix_len=len(_phone_prefix)

98 - def _parse_args(self, args, force_phonefs=False):

99 """Parse the args for a list of input and out args 100 @param args: input agrs 101 @returns: a dict that has 2 keys: 'source' and 'dest'. 'source' has a 102 list of files and/or directories. 'dest' is either a file or dir. 103 """ 104 _res=[] 105 for _item in args: 106 if _item.startswith(self._phone_prefix): 107 _dirname=_item[self._phone_prefix_len:] 108 _phonefs=True 109 else: 110 _dirname=_item 111 _phonefs=force_phonefs 112 if _phonefs: 113 if _dirname.startswith('/'): 114 # absolute path 115 _dirname=self.phone.join(self._rootdir, _dirname[1:]) 116 else: 117 _dirname=self.phone.join(self._pwd, _dirname) 118 # check for wildcards 119 global wildcards 120 _name=self.phone.basename(_dirname) 121 if wildcards.match(_name): 122 _dirname=self.phone.dirname(_dirname) 123 _wildcard=_name 124 else: 125 _wildcard=None 126 # correct path 127 if _phonefs: 128 _path=None 129 else: 130 _path=os.path.join(*_dirname.split('/')) 131 _res.append({ 'name': _dirname, 'phonefs': _phonefs, 132 'wildcard': _wildcard, 133 'path': _path }) 134 return _res

135

136 - def run(self, cmdline=None):

137 """Execute the specified command if specified, or the one 138 currently stored 139 @param cmdline: string command line 140 @returns: (T, None) if the command completed successfully, 141 (F, error code) otherwise 142 """ 143 if cmdline: 144 _cmdline=cmdline.split(None) 145 self.cmd=_cmdline[0] 146 self.args=_cmdline[1:] 147 _func=getattr(self, self.cmd, None) 148 if _func is None: 149 self._err.write('Error: invalid command: %s\n'%self.cmd) 150 return (False, InvalidCommand_Error) 151 return _func(self.args)

152

153 - def log(self, logstr):

154 pass

155

156 - def logdata(self, logstr, logdata, dataclass=None, datatype=None):

157 pass

158

159 - def progress(self, pos, maxcnt, desc):

160 "Update the progress meter" 161 pass

162

163 - def ls(self, args):

164 """Do a directory listing 165 @param args: string directory names 166 @returns: (True, None) if successful, (False, error code) otherwise 167 """ 168 _src=self._parse_args(args, force_phonefs=True) 169 if not _src: 170 _src=[{ 'name': self._pwd, 'phonefs': True, 171 'path': None, 'wildcard': None }] 172 for _dir in _src: 173 try: 174 _dirlist=self.phone.getfilesystem(_dir['name']) 175 self._out.write('%s:\n'%_dir['name']) 176 for _,_file in _dirlist.items(): 177 if _dir['wildcard'] is None or \ 178 fnmatch.fnmatch(self.phone.basename(_file['name']), _dir['wildcard']): 179 self._out.write('%s\n'%_file['name']) 180 except (phones.com_brew.BrewNoSuchDirectoryException, 181 phones.com_brew.BrewBadPathnameException): 182 self._out.write('Error: cannot access %s: no such file or directory\n'%_dir['name']) 183 self._out.write('\n') 184 return (True, None)

185

186 - def ll(self, args):

187 """Do a long dir listing command 188 @param args: string directory names 189 @returns: (True, None) if successful, (False, error code) otherwise 190 """ 191 _src=self._parse_args(args, force_phonefs=True) 192 if not _src: 193 _src=[{ 'name': self._pwd, 'phonefs': True, 'path': None, 194 'wildcard': None }] 195 for _dir in _src: 196 try: 197 _dirlist=self.phone.getfilesystem(_dir['name']) 198 self._out.write('%s:\n'%_dir['name']) 199 _maxsize=0 200 _maxdatelen=0 201 for _,_file in _dirlist.items(): 202 if _file.get('type', '')=='file': 203 _maxsize=max(_maxsize, _file.get('size', 0)) 204 _maxdatelen=max(_maxdatelen, len(_file.get('date', (0, ''))[1])) 205 _formatstr='%%(dir)1s %%(size)%(maxsizelen)is %%(date)%(maxdatelen)is %%(name)s\n'%\ 206 { 'maxdatelen': _maxdatelen, 207 'maxsizelen': len(str(_maxsize)) } 208 for _,_file in _dirlist.items(): 209 if _dir['wildcard'] is None or \ 210 fnmatch.fnmatch(self.phone.basename(_file['name']), _dir['wildcard']): 211 _strdict={ 'name': _file['name'] } 212 if _file['type']=='file': 213 _strdict['dir']='' 214 _strdict['size']=str(_file['size']) 215 else: 216 _strdict['dir']='d' 217 _strdict['size']='' 218 _strdict['date']=_file.get('date', (0, ''))[1] 219 self._out.write(_formatstr%_strdict) 220 except (phones.com_brew.BrewNoSuchDirectoryException, 221 phones.com_brew.BrewBadPathnameException): 222 self._out.write('Error: cannot access %s: no such file or directory\n'%_dir['name']) 223 self._out.write('\n') 224 return (True, None)

225

226 - def _cpfilefromphone(self, filename, destdir):

227 # copy a single file from the phone to the dest dir 228 with open(os.path.join(destdir, self.phone.basename(filename)), 229 'wb') as f: 230 f.write(self.phone.getfilecontents(filename)) 231 self._out.write('Copied file %(srcname)s to %(dirname)s\n'% { 'srcname': filename, 232 'dirname': destdir})

233 - def _cpdirfromphone(self, dirname, destdir, wildcard):

234 # copy all files under a phone dir to the dest dir 235 for _, _item in self.phone.listfiles(dirname).items(): 236 if wildcard is None or \ 237 fnmatch.fnmatch(self.phone.basename(_item['name']), wildcard): 238 self._cpfilefromphone(_item['name'], destdir)

239

240 - def _cpfromphone(self, args):

241 # copy files from the phone 242 _destdir=args[-1]['path'] 243 if not os.path.isdir(_destdir): 244 self._out.write('Error: %(dirname)s is not a valid local directory.\n'% {'dirname': _destdir }) 245 return (False, InvalidDir_Error) 246 for _item in args[:-1]: 247 _name=_item['name'] 248 if self.phone.isdir(_name): 249 # this is a dir, cp all files under it 250 self._cpdirfromphone(_name, _destdir, _item['wildcard']) 251 elif self.phone.isfile(_name): 252 # this is a file, just copy it 253 self._cpfilefromphone(_name, _destdir) 254 else: 255 # not sure what it is 256 self._out.write('Error: %(name)s does not exist\n'%{'name': _name}) 257 return (True, None)

258

259 - def _cpfiletophone(self, name, destdir, phonefs=False, force=False):

260 # copy a file to the phone 261 _filename=self.phone.join(destdir, 262 self.phone.basename(name) if phonefs else \ 263 os.path.basename(name)) 264 if not force: 265 # check if file already exists 266 if self.phone.exists(_filename): 267 # file exists, warn 268 self._out.write('Phone file %(name)s exists, overwrite (y/n): '%\ 269 { 'name': _filename }) 270 if self._in.readline()[0].upper()!='Y': 271 return 272 if phonefs: 273 # cp from phone FS to phone FS 274 self.phone.writefile(_filename, 275 self.phone.getfilecontents(name)) 276 else: 277 # local to phone FS 278 with open(name, 'rb') as f: 279 self.phone.writefile(_filename, 280 f.read()) 281 self._out.write('Copied %(filename)s to %(dirname)s.\n'%\ 282 { 'filename': name, 283 'dirname': destdir })

284

285 - def _cpdirtophone(self, dirname, destdir, phonefs=False):

286 # cp a dir to the phone 287 if phonefs: 288 # phone FS dir 289 for _, _item in self.phone.listfiles(dirname).items(): 290 self._cpfiletophone(_item['name'], destdir, phonefs=True) 291 else: 292 # local dir 293 for _item in os.listdir(dirname): 294 if os.path.isfile(_item): 295 self._cpfiletophone(_item, destdir, phonefs=False)

296

297 - def _cptophone(self, args):

298 # copy files to the phone 299 _destdir=args[-1]['name'] 300 if not self.phone.isdir(_destdir): 301 self._out.write('Error: phone directory %(dirname)s is not exist.\n'%\ 302 { 'dirname': _destdir }) 303 return (False, InvalidDir_Error) 304 for _item in args[:-1]: 305 if _item['phonefs']: 306 # this one on the phone 307 _name=_item['name'] 308 if self.phone.isdir(_name): 309 self._cpdirtophone(_name, _destdir, phonefs=True) 310 elif self.phone.isfile(_name): 311 self._cpfiletophone(_name, _destdir, phonefs=True) 312 else: 313 self._out.write('Error: %(name)s does not exist.\n'%\ 314 { 'name': _name }) 315 else: 316 # this one on the PC 317 _name=_item['path'] 318 if os.path.isdir(_name): 319 self._cpdirtophone(_name, _destdir, phonefs=False) 320 elif os.path.isfile(_name): 321 self._cpfiletophone(_name, _destdir, phonefs=False) 322 else: 323 self._out.write('Error: %(name) does not exist.\n'%\ 324 { 'name': _name }) 325 return (True, None)

326

327 - def cp(self, args):

328 """Transfer files between the phone filesystem and local filesystem 329 @param args: string dir names 330 @returns: (True, None) if successful, (False, error code) otherwise 331 """ 332 _args=self._parse_args(args, force_phonefs=False) 333 # The syntax of the last argument indicates the direction of the transfer 334 # If the last arg is a phone FS: copy to the phone 335 # If the last arg is not a phone FS: copy from the phone. 336 if _args[-1]['phonefs']: 337 # copy to the phone 338 return self._cptophone(_args) 339 else: 340 # copy from the phone 341 return self._cpfromphone(_args)

342

343 - def mkdir(self, args):

344 """Create one or more dirs on the phone FS. 345 @param args: string dir names 346 @returns: (True, None) if successful, (False, error code) otherwise 347 """ 348 _src=self._parse_args(args, force_phonefs=True) 349 for _dir in _src: 350 try: 351 self.phone.mkdir(_dir['name']) 352 except phones.com_brew.BrewDirectoryExistsException: 353 self._out.write('Error: dir %(name)s exists.\n'% \ 354 { 'name': _dir['name'] }) 355 except phones.com_brew.BrewNoSuchDirectoryException: 356 self._out.write('Error: Failed to create dir %(name)s.\n'%\ 357 { 'name': _dir['name'] }) 358 return (True, None)

359

360 - def cli(self, _):

361 """Run a primitive interactive CLI sesssion. 362 @params _: don't care 363 @returns: always (True, None) 364 """ 365 if self._inCLI: 366 # we're in our shell, bail 367 return (True, None) 368 self._inCLI=True 369 ## try: 370 while True: 371 self._out.write('BitPim>') 372 _cmdline=self._in.readline() 373 if _cmdline.startswith('exit'): 374 break 375 self.run(_cmdline) 376 ## finally: 377 self._inCLI=False 378 return (True, None)

379

380 - def cd(self, args):

381 """Change the current working dir on the phone FS. 382 @param args: dir name(s), only args[0] matters. 383 @returns: (True, None) if successful, (False, error code) otherwise 384 """ 385 _dirs=self._parse_args(args, force_phonefs=True) 386 if _dirs: 387 _dirname=_dirs[0]['name'] 388 if _dirname=='/': 389 _dirname=self._rootdir 390 if self.phone.exists(_dirname): 391 self._pwd=_dirname 392 else: 393 self._out.write('Invalid dir: %s\n'%_dirname) 394 return self.pwd(args)

395 - def cdu(self, args):

396 """Change to current working dir on the phone up one level (parent). 397 @param _: don't care 398 @returns: (True, None) if successful, (False, error code) otherwise 399 """ 400 _updir=self.phone.dirname(self._pwd) 401 if _updir: 402 self._pwd=_updir 403 return self.pwd(args)

404

405 - def pwd(self, _):

406 """Print the current working dir(cwd/pwd) on the phone FS. 407 @params _: don't care 408 @returns: (True, None) 409 """ 410 self._out.write("%(name)s\n"%{ 'name': self._pwd }) 411 return (True, None)

412 cwd=pwd 413

414 - def rm(self, args):

415 """delete one or more files on the phone FS. This command does not 416 delete local files. 417 @param args: file names 418 @returns: (True, None) if successful, (False, error code) otherwise. 419 """ 420 _filenames=self._parse_args(args, force_phonefs=True) 421 for _item in _filenames: 422 _name=_item['name'] 423 if self.phone.isfile(_name): 424 self.phone.rmfile(_name) 425 self._out.write('File %(name)s deleted\n'%{ 'name': _name }) 426 else: 427 self._out.write('Invalid file: %(name)s\n'%{ 'name': _name })

428

429 - def rmdir(self, args):

430 """Delete one or more phone FS directories. This command does not 431 delete local directories. 432 @param args: dir names. 433 @returns: (True, None) if successful, (False, error code) otherwise. 434 """ 435 _dirnames=self._parse_args(args, force_phonefs=True) 436 for _item in _dirnames: 437 _name=_item['name'] 438 if self.phone.isdir(_name): 439 # this is a dir, check for empty 440 if self.phone.hassubdirs(_name) or \ 441 self.phone.listfiles(_name): 442 self._out.write('Dir %(name)s is not empty.\n'%{'name': _name }) 443 else: 444 self.phone.rmdirs(_name) 445 self._out.write('Dir %(name)s deleted.\n'% { 'name': _name }) 446 else: 447 self._out.write('Invalid dir: %(name)s\n'%{ 'name': _name})

Source Code for Module bp_cli