def version(self):
"""Print out the version number and exit"""
self.printout(version.__version__)
python类__version__()的实例源码
def __call__(self, parser, namespace, values, option_string):
try:
import version
print('kdtool version {}, {}.'.format(
version.__version__, version.__info__))
except ImportError:
print('kdtool, development build or manual installation.')
exit(0)
# Create our argument parser.
def main():
puts(colored.green("Pgrepup %s" % __version__))
try:
dispatch(__doc__)
except KeyboardInterrupt:
puts("\n" + colored.red('Execution aborted'))
exit(0)
def main():
parser = argparse.ArgumentParser(prog='zibra', formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("-v", "--version", help="Installed zibra version",
action="version",
version="%(prog)s " + str(version.__version__))
subparsers = parser.add_subparsers(title='[sub-commands]', dest='command', parser_class=ArgumentParserWithDefaults)
# extract
parser_extract = subparsers.add_parser('extract',
help='Create an empty poredb database')
parser_extract.add_argument('directory', metavar='directory',
help='The name of the database.')
parser_extract.add_argument('--basecaller', metavar='basecaller',
default='ONT Albacore Sequencing Software',
help='The name of the basecaller')
parser_extract.set_defaults(func=run_subtool)
# callers
parser_extract = subparsers.add_parser('basecaller', help='Display basecallers in files')
parser_extract.add_argument('directory', metavar='directory', help='Directory of FAST5 files.')
parser_extract.set_defaults(func=run_subtool)
# demultiplex
parser_demultiplex = subparsers.add_parser('demultiplex', help='Run demultiplex')
parser_demultiplex.add_argument('fasta', metavar='fasta', help='Undemultiplexed FASTA file.')
parser_demultiplex.add_argument('--threads', type=int, default=8, help='Number of threads')
parser_demultiplex.add_argument('--prefix', help='Prefix for demultiplexed files')
parser_demultiplex.add_argument('--no-remove-directory', dest='no_remove_directory', action='store_true')
parser_demultiplex.set_defaults(func=run_subtool)
# minion
parser_minion = subparsers.add_parser('minion', help='Run demultiplex')
parser_minion.add_argument('scheme', metavar='scheme', help='The name of the scheme.')
parser_minion.add_argument('sample', metavar='sample', help='The name of the sample.')
parser_minion.add_argument('--normalise', dest='normalise', type=int, default=100, help='Normalise down to moderate coverage to save runtime.')
parser_minion.add_argument('--threads', type=int, default=8, help='Number of threads')
parser_minion.add_argument('--scheme-directory', metavar='scheme_directory', default='/zibra/zika-pipeline/schemes', help='Default scheme directory')
parser_minion.add_argument('--max-haplotypes', type=int, default=1000000, metavar='max_haplotypes', help='max-haplotypes value for nanopolish')
parser_minion.add_argument('--read-file', metavar='read_file', help='Use alternative FASTA/FASTQ file to <sample>.fasta')
parser_minion.add_argument('--skip-nanopolish', action='store_true')
parser_minion.set_defaults(func=run_subtool)
# import
"""parser_import = subparsers.add_parser('import',
help='Import files into a poredb database')
parser_import.add_argument('db', metavar='DB',
help='The poredb database.')
parser_import.add_argument('fofn', metavar='FOFN',
help='A file containing a list of file names.')
parser_import.add_argument('--alternate-path', metavar='alternate_path')
parser_import.set_defaults(func=run_subtool)
"""
args = parser.parse_args()
if args.quiet:
logger.setLevel(logging.ERROR)
args.func(parser, args)
def main():
"""Main entry point"""
# Build default paths for files.
dirs = appdirs.AppDirs("hangupsbot", "hangupsbot")
default_log_path = os.path.join(dirs.user_data_dir, "hangupsbot.log")
default_cookies_path = os.path.join(dirs.user_data_dir, "cookies.json")
default_config_path = os.path.join(dirs.user_data_dir, "config.json")
default_memory_path = os.path.join(dirs.user_data_dir, "memory.json")
# Configure argument parser
parser = argparse.ArgumentParser(
prog="hangupsbot",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("-d", "--debug", action="store_true",
help=_("log detailed debugging messages"))
parser.add_argument("--log", default=default_log_path,
help=_("log file path"))
parser.add_argument("--cookies", default=default_cookies_path,
help=_("cookie storage path"))
parser.add_argument("--memory", default=default_memory_path,
help=_("memory storage path"))
parser.add_argument("--config", default=default_config_path,
help=_("config storage path"))
parser.add_argument("--retries", default=5, type=int,
help=_("Maximum disconnect / reconnect retries before "
"quitting"))
parser.add_argument("--version", action="version",
version="%(prog)s {}".format(version.__version__),
help=_("show program\"s version number and exit"))
args = parser.parse_args()
# Create all necessary directories.
for path in [args.log, args.cookies, args.config, args.memory]:
directory = os.path.dirname(path)
if directory and not os.path.isdir(directory):
try:
os.makedirs(directory)
except OSError as err:
sys.exit(_("Failed to create directory: %s"), err)
# If there is no config file in user data directory, copy default one there
if not os.path.isfile(args.config):
try:
shutil.copy(
os.path.abspath(os.path.join(os.path.dirname(sys.argv[0]),
"config.json")),
args.config)
except (OSError, IOError) as err:
sys.exit(_("Failed to copy default config file: %s"), err)
configure_logging(args)
# initialise the bot
bot = HangupsBot(args.cookies, args.config, args.memory, args.retries)
# start the bot
bot.run()
def __init__(self, user_list, subreddit_list, queue, unfinished_downloads_list):
"""
Class that does the main part of the work for the program. This class contains the praw instance that is used
for actually extracting the content from reddit. When an instance is created all settings parameters must be
supplied to the instance. This class also calls the necessary functions of the other classes and completes the
downloads.
:param user_list: The actual list of User objects contained in the user ListModel class displayed in the GUI
:param subreddit_list: The actual list of Subreddit objects contained in the subreddit ListModel class displayed
in the GUI. If this class is initialized to only run a user list, this parameter is None
:param queue: The queue that text is added to in order to update the GUI
The rest of teh parameters are all configuration options that are set in the settings dialog
"""
super().__init__()
self._r = praw.Reddit(user_agent='python:DownloaderForReddit:%s (by /u/MalloyDelacroix)' % __version__,
client_id='frGEUVAuHGL2PQ', client_secret=None)
self.settings_manager = Core.Injector.get_settings_manager()
self.user_list = user_list
self.subreddit_list = subreddit_list
self.queue = queue
self.validated_objects = Queue()
self.validated_subreddits = []
self.failed_downloads = []
self.downloaded_users = {}
self.unfinished_downloads = []
self.user_run = True if self.user_list is not None else False
self.post_limit = self.settings_manager.post_limit
self.save_path = self.settings_manager.save_directory
self.subreddit_sort_method = self.settings_manager.subreddit_sort_method
self.subreddit_sort_top_method = self.settings_manager.subreddit_sort_top_method
self.restrict_date = self.settings_manager.restrict_by_date
self.restrict_by_score = self.settings_manager.restrict_by_score
self.restrict_score_method = self.settings_manager.score_limit_operator
self.restrict_score_limit = self.settings_manager.post_score_limit
self.unfinished_downloads_list = unfinished_downloads_list
self.load_undownloaded_content = self.settings_manager.save_undownloaded_content
self.queued_posts = Queue()
self.run = True
self.start_extractor()
self.start_downloader()
self.download_number = 0
