pgap.py

#!/usr/bin/env python3
from __future__ importprint_function
importsys

min_python= (3,6)
try:
assert(sys.version_info>=min_python)
except:
fromplatformimportpython_version
print("Python version", python_version(), "is too old.")
print("Please use Python", ".".join(map(str,min_python)), "or later.")
sys.exit()

importargparse
importatexit
importcontextlib
importglob
importjson
importmultiprocessingasmp
importos
importplatform
importqueue
importre
importshutil
importsubprocess
importtarfile
importtime
importtempfile
importxml
importxml.dom.minidom


fromioimportopen
fromurllib.parseimporturlparse, urlencode
fromurllib.requestimporturlopen, urlretrieve, Request
fromurllib.errorimportHTTPError

defis_venv():
return (hasattr(sys, 'real_prefix') or
 (hasattr(sys, 'base_prefix') andsys.base_prefix!=sys.prefix))

classurlopen_progress:
timeout=60
retries=10

def__init__(self, url, quiet, teamcity):
self.url=url
self.bytes_so_far=0
self.urlopen()

self.quiet=quiet
self.teamcity=teamcity
ifteamcity:
self.EOL='\n'
else:
self.EOL='\r'
self.cur_row=-1
total_size=0
total_size=self.remote_file.getheader('Content-Length', 0) # More modern method

self.total_size=int(total_size)

defurlopen(self):
headers=dict()
ifself.bytes_so_far>0:
headers['Range'] ='bytes={}-'.format(self.bytes_so_far)
request=Request(self.url, headers=headers)
self.remote_file=urlopen(request, timeout=self.timeout)

defread(self, n=8388608):
delay=1
forattemptinrange(self.retries):
try:
ifself.remote_fileisNone:
self.urlopen()
buffer=self.remote_file.read(n)
ifnotbuffer:
ifnotself.quiet:
sys.stdout.write('\n')
return''
break
exceptExceptionasex:
self.remote_file=None
time.sleep(delay)
delay+=delay

self.bytes_so_far+=len(buffer)
percent=float(self.bytes_so_far) /self.total_size
percent=round(percent*100, 2)

do_print=True
ifself.teamcity:
do_print=False
row=int(percent)
ifrow>self.cur_row:
self.cur_row=row
do_print=True

ifdo_printandnotself.quiet:
sys.stderr.write("Downloaded %d of %d bytes (%0.2f%%)%s"% (self.bytes_so_far, self.total_size, percent, self.EOL))

returnbuffer

definstall_url(url, path, quiet, teamcity, guard_file):
basename=os.path.basename(urlparse(url).path)
try:
local_file=os.path.join(path, basename)
ifos.path.exists(local_file):
ifnotquiet:
print('Extracting local tarball: {}'.format(local_file))
fileobj=open(local_file, 'rb')
else:
ifnotquiet:
print('Downloading and extracting tarball: {}'.format(url))
fileobj=urlopen_progress(url, quiet, teamcity)
withtarfile.open(mode='r|*', fileobj=fileobj) astar:
tar.extractall(path=path)
except:
sys.stderr.write('''
ERROR: Failed to extract tarball; to install manually, try something like:
 curl -OLC - {}
 tar xvf {}
'''.format(url, basename))
raise
ifguard_file!=None:
open(guard_file, 'a').close()

defquiet_remove(filename):
withcontextlib.suppress(FileNotFoundError):
os.remove(filename)

deffind_failed_step(filename):
r=r"^\[(?P<time>[^\]]+)\] (?P<level>[^ ]+) \[(?P<source>[^ ]*) (?P<name>[^\]]*)\] (?P<status>.*)"
search=re.compile(r)
lines=open(filename, "r").readlines()
nameStarts= {}
start=-1
fornum, lineinenumerate(lines):
r=search.match(line)
ifr:
name=r.group("name")
ifnamenotinnameStarts:
nameStarts[name] =num

ifr.group("status") =="completed permanentFail":
start=nameStarts[name]
break
ifstart>-1:
print("Printing log starting from failed job:\n")
foriinrange(start, len(lines)):
print(lines[i], end="")
else:
print("Unable to find error in log file.")

defget_cpus(self):
if'SLURM_CPUS_PER_TASK'inos.environ:
returnint(os.environ['SLURM_CPUS_PER_TASK'])
elif'NSLOTS'inos.environ:
returnint(os.environ['NSLOTS'])
elifself.params.args.cpus:
returnself.params.args.cpus
else:
return0

classPipeline:

defcleanup(self):
forfilein [self.yaml, self.submol]:
iffile!=None:
base=os.path.basename(file)
fullpath=os.path.join(self.params.outputdir, base)
ifos.path.exists(fullpath):
os.remove(fullpath)

def__init__(self, params, local_input, pipeline):
self.params=params
self.cwlfile=f"pgap/{pipeline}.cwl"
self.pipename=pipeline.upper()
self.pipeline=pipeline

self.data_dir=os.path.abspath(self.params.data_path)
self.input_dir=os.path.dirname(os.path.abspath(local_input))

# input file location inside docker instance:
self.input_file='/pgap/output/pgap_input.yaml'
submol=self.get_submol(local_input)
if ( submol!=None ):
self.submol=self.create_submolfile(submol, params.ani_output, params.ani_hr_output, params.args.auto_correct_tax)
else:
self.submol=None

add_std_validation_exemptions=False
args=self.params.args
if (args.genomeandargs.organism) orargs.asn1_input:
add_std_validation_exemptions=True
self.yaml=self.create_inputfile(local_input, add_std_validation_exemptions)
ifself.params.docker_typein ['singularity', 'apptainer']:
self.make_singularity_cmd()
elifself.params.docker_type=='podman':
self.make_podman_cmd()
else:
self.make_docker_cmd()

self.cmd.extend(['cwltool',
'--timestamps',
'--debug',
'--disable-color',
'--preserve-entire-environment',
'--outdir', '/pgap/output'
 ])

# Debug flags for cwltool
ifself.params.args.debug:
self.cmd.extend([
'--tmpdir-prefix', '/pgap/output/debug/tmpdir/',
'--leave-tmpdir',
'--tmp-outdir-prefix', '/pgap/output/debug/tmp-outdir/',
'--copy-outputs'])

self.cmd.extend([self.cwlfile, self.input_file])

defmake_docker_cmd(self):
self.cmd= [self.params.docker_cmd, 'run', '-i', '--rm' ]
self.cmd.extend(['--platform', 'linux/amd64'])

cpusEnv=get_cpus(self)
if (cpusEnv):
self.cmd.extend(['--cpus', str(get_cpus(self))])

ifself.params.no_internet:
self.cmd.extend(['--network=none'])

if (self.params.args.memory):
self.cmd.extend(['--memory', self.params.args.memory])

ifself.params.docker_user_remap:
self.cmd.extend(['--user', str(os.getuid()) +":"+str(os.getgid())])
self.cmd.extend([
'--volume', '{}:/pgap/input:ro,z'.format(self.data_dir),
'--volume', '{}:/pgap/user_input:z'.format(self.input_dir),
'--volume', '{}:/pgap/output:rw,z'.format(self.params.outputdir),
'--volume', '{}:{}:ro,z'.format(self.yaml, self.input_file ),
'--volume', '{}:/tmp:rw,z'.format(os.getenv("TMPDIR", "/tmp"))])

# Debug mount for docker image
ifself.params.args.debug:
log_dir=self.params.outputdir+'/debug/log'
os.makedirs(log_dir, exist_ok=True)
self.cmd.extend(['--volume', '{}:/log/srv:z'.format(log_dir)])
ifself.params.args.container_name:
self.cmd.extend(['--name', self.params.args.container_name])
self.cmd.append(self.params.docker_image)


defmake_podman_cmd(self):
self.cmd= [self.params.docker_cmd]
ifself.params.args.debug:
self.cmd.extend(['--log-level', 'debug'])
self.cmd.extend(['run', '-i', '--rm', '--privileged' ])

cpusEnv=get_cpus(self)
if (cpusEnv):
self.cmd.extend(['--cpus', str(get_cpus(self))])

ifself.params.no_internet:
self.cmd.extend(['--network=none'])

if (self.params.args.memory):
self.cmd.extend(['--memory', self.params.args.memory])

self.cmd.extend([
'--volume', '{}:/pgap/input:ro,Z'.format(self.data_dir),
'--volume', '{}:/pgap/user_input:Z'.format(self.input_dir),
'--volume', '{}:/pgap/output:rw,Z'.format(self.params.outputdir),
'--volume', '{}:{}:ro,Z'.format(self.yaml, self.input_file ),
'--volume', '{}:/tmp:rw'.format(os.getenv("TMPDIR", "/tmp"))])

# Debug mount for docker image
ifself.params.args.debug:
log_dir=self.params.outputdir+'/debug/log'
os.makedirs(log_dir, exist_ok=True)
self.cmd.extend(['--volume', '{}:/log/srv'.format(log_dir)])
ifself.params.args.container_name:
self.cmd.extend(['--name', self.params.args.container_name])
self.cmd.append(self.params.docker_image)

defmake_singularity_cmd(self):
self.cmd= [self.params.docker_cmd, 'exec' ]

cpusEnv=get_cpus(self)
if (cpusEnv):
self.cmd.extend(['--cpus', str(get_cpus(self))])

ifself.params.no_internet:
self.cmd.extend(['--network=none'])

if (self.params.args.memory):
self.cmd.extend(['--memory', self.params.args.memory])

self.cmd.extend([
'--bind', '{}:/pgap/input:ro'.format(self.data_dir),
'--bind', '{}:/pgap/user_input'.format(self.input_dir),
'--bind', '{}:/pgap/output:rw'.format(self.params.outputdir),
'--bind', '{}:{}:ro'.format(self.yaml, self.input_file ),
'--bind', '{}:/tmp:rw'.format(os.getenv("TMPDIR", "/tmp"))])

# Debug mount for docker image
ifself.params.args.debug:
log_dir=self.params.outputdir+'/debug/log'
os.makedirs(log_dir, exist_ok=True)
self.cmd.extend(['--bind', '{}:/log/srv'.format(log_dir)])

ifself.params.args.container_path:
self.cmd.extend(["--pwd", "/pgap", self.params.docker_image])
else:
self.cmd.extend(["--pwd", "/pgap", "docker://"+self.params.docker_image])

defget_submol(self, local_input):
withopen(local_input, 'r') asfIn:
processing_submol=False
forlineinfIn:
ifline: # skip empty lines
if'submol:'inline: # we need to replace submol/location with new file
processing_submol=True
if'location:'inlineandprocessing_submol==True:
line=line.replace('location: ','')
submol_file=line
submol_file=submol_file.strip()
returnos.path.join(os.path.dirname(local_input), submol_file)
returnNone

defregexp_file(self, filename, field):

withopen(filename, 'r') asfIn:
forlineinfIn:
if(re.search(field,line)):
returnTrue
returnFalse
defget_genus_species(self, xml_file):
genus_species=None
try:
doc=xml.dom.minidom.parse(xml_file)
predicted_taxid=doc.getElementsByTagName('predicted-taxid')[0]
ifpredicted_taxid.getAttribute('confidence')=='HIGH':
genus_species=predicted_taxid.getAttribute('org-name')
except:
genus_species=None
returngenus_species

defcreate_submolfile(self, local_submol, ani_output, ani_hr_output, auto_correct_tax):
has_authors=self.regexp_file(local_submol, '^authors:')
has_contact_info=self.regexp_file(local_submol, '^contact_info:')
genus_species=None
ifauto_correct_tax:
ifani_output!=None: 
genus_species=self.get_genus_species(ani_output)
ifgenus_species!=None:
print('ANI analysis detected species "{}", and we will use it for PGAP'.format(genus_species))
else:
ifani_hr_output!=None:
chosen_ani_output_type=ani_hr_output
else:
chosen_ani_output_type=ani_output
print('ERROR: taxcheck failed to assign a species with high confidence, thus PGAP will not execute. See {}'.format(chosen_ani_output_type))

sys.exit(1)
withtempfile.NamedTemporaryFile(mode='w',
suffix=".yaml",
prefix="pgap_submol_",
dir=self.params.outputdir,
delete=False) asfOut:
yaml=os.path.basename(fOut.name)
withopen(local_submol, 'r') asfIn:
forlineinfIn:
ifline: # skip empty lines
ifauto_correct_taxandgenus_species!=Noneandre.match(r'\s+genus_species:', line):
print('replacing organism in line: {}'.format(line.rstrip()))
line=re.sub(r'genus_species:.*', "genus_species: '{}'".format(genus_species), line)
print('with organism: {}'.format(genus_species))
fOut.write(line.rstrip())
fOut.write(u'\n')
ifhas_authors==False:
fOut.write(u'authors:\n')
fOut.write(u' - author:\n')
#
# note: do not change these defaults, they are coded now 
# in standard diagnostics asnvalidate tool, that's how GenBank detects that users did not provide correct names
#
fOut.write(u" first_name: 'Firstname'\n")
fOut.write(u" last_name: 'Lastname'\n")
ifhas_contact_info==False:
fOut.write(u'contact_info:\n')
fOut.write(u" first_name: 'Firstname'\n")
fOut.write(u" last_name: 'Lastname'\n")
fOut.write(u" email: 'Email@address.net'\n")
fOut.write(u" organization: 'Organization'\n")
fOut.write(u" department: 'Department'\n")
fOut.write(u" phone: '301-555-0245'\n")
fOut.write(u" street: '100 Street St'\n")
fOut.write(u" city: 'City'\n")
fOut.write(u" postal_code: '12345'\n")
fOut.write(u" country: 'Country'\n")
fOut.flush()
returnyaml


defcreate_inputfile(self, local_input, add_std_validation_exemptions):
withtempfile.NamedTemporaryFile(mode='w',
suffix=".yaml",
prefix="pgap_input_",
dir=self.params.outputdir,
delete=False) asfOut:
yaml=fOut.name
withopen(local_input, 'r') asfIn:
processing_submol=False
processing_fasta=False
processing_entries=False
processing_submol_json=False
forlineinfIn:
ifline: # skip empty lines
if'submol:'inline: # we need to replace submol/location with new file
processing_submol=True
processing_fasta=False
processing_entries=False
processing_submol_json=False
if'location:'inlineandprocessing_submol:
processing_submol=False
processing_entries=False
processing_submol_json=False
pos=line.index('location: ')
line=' '*pos+u'location: '+self.submol+'\n'
if'fasta:'inline: # we need to copy fasta input to output_dir
processing_submol=False
processing_fasta=True
processing_entries=False
processing_submol_json=False
if'entries:'inline: # we need to copy entries input to output_dir
processing_submol=False
processing_fasta=False
processing_submol_json=False
processing_entries=True
if'submol_block_json:'inline: 
processing_submol=False
processing_fasta=False
processing_submol_json=True
processing_entries=False
if'location:'inline:
local_input_dir=os.path.dirname(os.path.abspath(local_input))
ifprocessing_fasta:
processing_fasta=False
input_fasta_location=None
ifself.params.args.genome:
input_fasta_location=self.params.args.genome
else: 
match=re.search(r'location:\s+(\S+)', line)
input_fasta_location=os.path.join(local_input_dir, match.group(1))
copy_genome_to_workspace(input_fasta_location, self.params.outputdir)
elifprocessing_entries:
processing_entries=False
match=re.search(r'location:\s+(\S+)', line)
input_entries_location=os.path.join(local_input_dir, match.group(1))
copy_genome_to_workspace(input_entries_location, self.params.outputdir)
elifprocessing_submol_json:
processing_submol_json=False
match=re.search(r'location:\s+(\S+)', line)
input_submol_json_location=os.path.join(local_input_dir, match.group(1))
copy_genome_to_workspace(input_submol_json_location, self.params.outputdir)

fOut.write(line.rstrip())
fOut.write(u'\n')
fOut.write(u'supplemental_data: { class: Directory, location: /pgap/input }\n')
if (self.params.report_usage!='none'):
fOut.write(u'report_usage: {}\n'.format(self.params.report_usage))
if (self.params.ignore_all_errors=='true'):
fOut.write(u'ignore_all_errors: {}\n'.format(self.params.ignore_all_errors))
if (self.params.no_internet=='true'):
fOut.write(u'no_internet: {}\n'.format(self.params.no_internet))
uuidfile=self.params.outputdir+"/uuid.txt"
ifos.path.exists(uuidfile) andos.stat(uuidfile).st_size!=0:
fOut.write(u'make_uuid: false\n')
fOut.write(u'uuid_in: { class: File, location: /pgap/output/uuid.txt }\n')
ifadd_std_validation_exemptions:
ifself.pipeline=='wf_common':
fOut.write(f"""
xpath_fail_initial_asndisc: >
 //*[@severity="FATAL"
 and not(contains(@name, "CITSUBAFFIL_CONFLICT"))
 ]
xpath_fail_initial_asnvalidate: >
 //*[
 ( @severity="ERROR" or @severity="REJECT" )
 and not(contains(@code, "SEQ_DESCR_BadOrgMod")) 
 and not(contains(@code, "SEQ_PKG_NucProtProblem")) 
 and not(contains(@code, "SEQ_DESCR_BacteriaMissingSourceQualifier"))
 ]
xpath_fail_final_asnvalidate: >
 //*[( @severity="ERROR" or @severity="REJECT" )
 and not(contains(@code, "GENERIC_MissingPubRequirement"))
 and not(contains(@code, "SEQ_DESCR_BadOrgMod")) 
 and not(contains(@code, "SEQ_DESCR_BacteriaMissingSourceQualifier"))
 and not(contains(@code, "SEQ_DESCR_Chromosomepath"))
 and not(contains(@code, "SEQ_DESCR_MissingLineage"))
 and not(contains(@code, "SEQ_DESCR_NoTaxonID"))
 and not(contains(@code, "SEQ_DESCR_UnwantedCompleteFlag"))
 and not(contains(@code, "SEQ_FEAT_ShortIntron"))
 and not(contains(@code, "SEQ_INST_InternalNsInSeqRaw"))
 and not(contains(@code, "SEQ_INST_ProteinsHaveGeneralID"))
 and not(contains(@code, "SEQ_PKG_ComponentMissingTitle"))
 and not(contains(@code, "SEQ_PKG_NucProtProblem")) 

 ]
""")
else:
fOut.write(f"""
xpath_fail_initial_asnvalidate: >
 //*[
 ( @severity="ERROR" or @severity="REJECT" )
 and not(contains(@code, "GENERIC_MissingPubRequirement")) 
 and not(contains(@code, "GENERIC_BadSubmissionAuthorName")) 
 and not(contains(@code, "SEQ_DESCR_ChromosomeLocation")) 
 and not(contains(@code, "SEQ_DESCR_MissingLineage")) 
 and not(contains(@code, "SEQ_DESCR_NoTaxonID")) 
 and not(contains(@code, "SEQ_DESCR_OrganismIsUndefinedSpecies"))
 and not(contains(@code, "SEQ_DESCR_StrainWithEnvironSample"))
 and not(contains(@code, "SEQ_DESCR_BacteriaMissingSourceQualifier"))
 and not(contains(@code, "SEQ_DESCR_UnwantedCompleteFlag")) 
 and not(contains(@code, "SEQ_FEAT_BadCharInAuthorLastName")) 
 and not(contains(@code, "SEQ_FEAT_ShortIntron")) 
 and not(contains(@code, "SEQ_INST_InternalNsInSeqRaw")) 
 and not(contains(@code, "SEQ_INST_ProteinsHaveGeneralID")) 
 and not(contains(@code, "SEQ_PKG_NucProtProblem")) 
 and not(contains(@code, "SEQ_PKG_ComponentMissingTitle")) 
 ]
xpath_fail_final_asnvalidate: >
 //*[( @severity="ERROR" or @severity="REJECT" )
 and not(contains(@code, "GENERIC_MissingPubRequirement")) 
 and not(contains(@code, "GENERIC_BadSubmissionAuthorName")) 
 and not(contains(@code, "SEQ_DESCR_ChromosomeLocation")) 
 and not(contains(@code, "SEQ_DESCR_MissingLineage")) 
 and not(contains(@code, "SEQ_DESCR_NoTaxonID")) 
 and not(contains(@code, "SEQ_DESCR_OrganismIsUndefinedSpecies"))
 and not(contains(@code, "SEQ_DESCR_StrainWithEnvironSample"))
 and not(contains(@code, "SEQ_DESCR_BacteriaMissingSourceQualifier"))
 and not(contains(@code, "SEQ_DESCR_UnwantedCompleteFlag")) 
 and not(contains(@code, "SEQ_FEAT_BadCharInAuthorLastName")) 
 and not(contains(@code, "SEQ_FEAT_ShortIntron")) 
 and not(contains(@code, "SEQ_INST_InternalNsInSeqRaw")) 
 and not(contains(@code, "SEQ_INST_ProteinsHaveGeneralID")) 
 and not(contains(@code, "SEQ_PKG_ComponentMissingTitle")) 
 and not(contains(@code, "SEQ_PKG_NucProtProblem")) 
 ]
""")
fOut.flush()
returnyaml

defreport_output_files(self, output, output_files):
# output_files = [
# {"file": "", "remove": True},
# ]
foroutput_pairinoutput_files:
fullname=os.path.join(output, output_pair["file"])
ifos.path.exists(fullname) andos.path.getsize(fullname) >0:
print(f'FILE: {output_pair["file"]}')
withopen(fullname, 'r') asf:
print(f.read())
ifoutput_pair["remove"]:
os.remove(fullname)


deflaunch(self):
cwllog=self.params.outputdir+'/cwltool.log'
withopen(cwllog, 'a', encoding="utf-8") asf:
# Show original command line in log
cmdline="Original command: "+" ".join(sys.argv)
f.write(cmdline)
f.write("\n\n")
# Show docker command line in log
cmdline="Docker command: "+" ".join(self.cmd)
f.write(cmdline)
f.write("\n\n")
# Show YAML file in the log
f.write("--- Start YAML Input ---\n") 
withopen(self.yaml, 'r') asfIn:
forlineinfIn:
f.write(line)
f.write("--- End YAML Input ---\n\n")

try:
proc=subprocess.Popen(self.cmd, stdout=f, stderr=subprocess.STDOUT)
proc.wait()
finally:
ifproc.returncode==None:
print('\nAbnormal termination, stopping all processes.')
proc.terminate()
elifproc.returncode==0:
ifself.pipename!="WF_COMMON":
print(f'{self.pipename} completed successfully.')
else:
print(f'{self.pipename} failed, docker exited with rc =', proc.returncode)
find_failed_step(cwllog)
output_files= [
 {"file": "final_asndisc_diag.xml", "remove": True},
 {"file": "final_asnval_diag.xml", "remove": True},
 {"file": "initial_asndisc_diag.xml", "remove": True},
 {"file": "initial_asnval_diag.xml", "remove": True}
 ]
self.report_output_files(self.params.outputdir, output_files)
returnproc.returncode

classSetup:

def__init__(self, args):
self.args=args
self.ani_output=None
# human readable ANI output file
self.ani_hr_output=None
self.branch=self.get_branch()
self.repo=self.get_repo()
self.registry_docker_hub_version="v2"

ifplatform.system() =='Windows':
self.install_dir=os.environ.get('PGAP_INPUT_DIR',os.environ['USERPROFILE']+'/.pgap')
else:
self.install_dir=os.environ.get('PGAP_INPUT_DIR',os.environ['HOME']+'/.pgap')

self.local_version=self.get_local_version()
ifself.args.no_internet:
self.remote_versions= [self.local_version]
else:
self.remote_versions=self.get_remote_versions()
self.report_usage=self.get_report_usage()
self.ignore_all_errors=self.get_ignore_all_errors()
self.no_internet=self.get_no_internet()
self.timeout=self.get_timeout()
self.check_status()
ifargs.version:
sys.exit(0)
if (args.list):
self.list_remote_versions()
return
self.use_version=self.get_use_version()
ifself.args.container_path:
self.docker_image=self.args.container_path
else:
self.docker_image="ncbi/{}:{}".format(self.repo, self.use_version)
self.data_path='{}/input-{}'.format(self.install_dir, self.use_version)
self.test_genomes_path='{}/test_genomes-{}'.format(self.install_dir, self.use_version)
self.outputdir=self.get_output_dir()
self.get_docker_info()
ifself.docker_type=='podman':
# see PGAPX-1073
self.docker_image="docker.io/ncbi/{}:{}".format(self.repo, self.use_version)

self.update_self()
if (self.local_version!=self.use_version) ornotself.check_install_data():
self.update()



defget_branch(self):
if (self.args.dev):
return"dev"
if (self.args.test):
return"test"
if (self.args.prod):
return"prod"
return""

defget_repo(self):
ifself.branch=="":
return"pgap"
return"pgap-"+self.branch

defget_local_version(self):
filename=self.install_dir+"/VERSION"
ifos.path.isfile(filename):
withopen(filename, encoding='utf-8') asf:
returnf.read().strip()
returnNone


defget_remote_versions(self):
versions= []
if (self.get_branch()):
#print("Checking docker hub for latest version.")
url=f'https://registry.hub.docker.com/{self.registry_docker_hub_version}/repositories/ncbi/{self.repo}/tags'
response=urlopen(url)
json_resp=json.loads(response.read().decode())
ifself.registry_docker_hub_version=='v2':
# looks like these are already sorted in right order in this version 
forresultinjson_resp['results']:
versions.append(result['name'])
elifself.registry_docker_hub_version=='v1':
foriinreversed(json_resp):
versions.append(i['name'])
else:
raiseException(f'INTERNAL ERROR: chosen docker hub registry version {self.registry_docker_hub_version} is not supporter.')
else:
#print("Checking github releases for latest version.")
response=urlopen('https://api.github.com/repos/ncbi/pgap/releases/latest')
latest=json.loads(response.read().decode())['tag_name']
versions.append(latest)
returnversions

defcheck_status(self):
ifself.local_version==None:
print("The latest version of PGAP is {}, you have nothing installed locally.".format(self.get_latest_version()))
returnFalse
ifself.args.no_internet:
print("--no-internet flag enabled, not checking remote versions.")
returnTrue
ifself.local_version==self.get_latest_version():
ifself.branch=="":
print("PGAP version {} is up to date.".format(self.local_version))
else:
print("PGAP from {} branch, version {} is up to date.".format(self.branch, self.local_version))
returnTrue
print("The latest version of PGAP is {}, you are using version {}, please update.".format(self.get_latest_version(), self.local_version))
returnTrue

deflist_remote_versions(self):
print("Available versions:")
foriinself.remote_versions:
print("\t", i)

defget_latest_version(self):
returnself.remote_versions[0]

defget_use_version(self):
ifself.args.use_version:
returnself.args.use_version
if (self.local_version==None) orself.args.update:
returnself.get_latest_version()
returnself.local_version

defget_output_dir(self):
returnos.path.abspath(self.args.output)

defget_docker_info(self):
docker_type_alternatives= ['docker', 'podman', 'singularity', 'apptainer']
ifself.args.docker:
self.docker_cmd=shutil.which(self.args.docker)
else:
fordockerindocker_type_alternatives:
self.docker_cmd=shutil.which(docker)
ifself.docker_cmd!=None:
break
ifself.docker_cmd==None:
sys.exit("Docker not found.")

version=subprocess.run([self.docker_cmd, '--version'], check=True, stdout=subprocess.PIPE, stdin=subprocess.DEVNULL).stdout.decode('utf-8')
self.docker_type=version.split(maxsplit=1)[0].lower()
ifself.docker_typenotindocker_type_alternatives:
self.docker_type=os.path.basename(os.path.realpath(self.docker_cmd)).lower()
ifself.docker_typenotindocker_type_alternatives:
self.docker_type='docker'
print('WARNING: {} ({}) support as Docker alternative has not been tested'.format(version, self.docker_cmd))
self.docker_user_remap= (self.docker_type=='docker'andplatform.system() !="Windows")

defget_report_usage(self):
if (self.args.report_usage_true):
return'true'
if (self.args.report_usage_false):
return'false'
return'none'

defget_ignore_all_errors(self):
if (self.args.ignore_all_errors):
return'true'
else:
return'false'

defget_no_internet(self):
if (self.args.no_internet):
return'true'
else:
return'false'

defget_timeout(self):
defstr2sec(s):
returnsum(x*int(t) forx, tin
zip(
 [1, 60, 3600, 86400],
reversed(s.split(":"))
 ))
returnstr2sec(self.args.timeout)

defupdate(self):
print(f"installation directory: {self.install_dir}")
subprocess.run(["/bin/df", "-k", self.install_dir], stdin=subprocess.DEVNULL)
self.update_self()
threads=list()
docker_thread=mp.Process(target=self.install_docker, name='docker image pull')
docker_thread.start()
threads.append(docker_thread)
# precreate the directory where the tarfile will be unloaded.
os.makedirs(f"{self.install_dir}/input-{self.use_version}/uniColl_path", exist_ok=True)
self.install_data(threads)
genomes_thread=mp.Process(target=self.install_test_genomes, name='test genomes installation')
genomes_thread.start()
threads.append(genomes_thread)
global_exit_value=0
forthreadinthreads:
thread.join()
ifthread.exitcode!=0:
global_exit_value=thread.exitcode
ifglobal_exit_value!=0:
raiseException(f'installation of some or all of components failed. Please remove {self.data_path}, {self.install_dir}/test_genomes, {self.test_genomes_path} directories and try again.')
self.write_version()

definstall_docker(self):
ifself.docker_typein ['singularity', 'apptainer']:
sif=self.docker_image.replace("ncbi/pgap:", "pgap_") +".sif"
try:
subprocess.run([self.docker_cmd, 'sif', 'list', sif],
check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL)
print("Singularity sif files exists, not updating.")
return
exceptsubprocess.CalledProcessError:
pass
docker_url="docker://"+self.docker_image
else:
docker_url=self.docker_image
print('Downloading (as needed) Docker image {}'.format(docker_url))
r=None;
try:
r=subprocess.run([self.docker_cmd, 'pull', docker_url], check=True, stdin=subprocess.DEVNULL)
#print(r)
exceptsubprocess.CalledProcessError:
print(r)
sys.exit(1)

# This and install data should probably be refactored
defcheck_install_data(self):
ifself.use_version>"2019-11-25.build4172":
ifself.args.ani:
packages= ['ani', 'pgap']
elifself.args.ani_only:
packages= ['ani']
else:
packages= ['pgap']
forpackageinpackages:
guard_file=f"{self.install_dir}/input-{self.use_version}/.{package}_complete"
ifnotos.path.isfile(guard_file):
returnFalse
returnTrue

definstall_data(self, threads):
suffix=""
ifself.branch!="":
suffix=self.branch+"."

ifself.args.ani:
packages= ['ani', 'pgap']
elifself.args.ani_only:
packages= ['ani']
else:
packages= ['pgap']
forpackageinpackages:
guard_file=f"{self.install_dir}/input-{self.use_version}/.{package}_complete"
ifpackage=="pgap":
remote_path=f"https://s3.amazonaws.com/pgap/input-{self.use_version}.{suffix}tgz"
else:
remote_path=f"https://s3.amazonaws.com/pgap/input-{self.use_version}.{suffix}{package}.tgz"
ifnotos.path.isfile(guard_file):
url_thread=mp.Process(target=install_url, name='{package} installation',args=(remote_path, self.install_dir, self.args.quiet, self.args.teamcity, guard_file, ))
url_thread.start()
threads.append(url_thread)
else:
print(f"Skipping already installed tarball: {remote_path}")

definstall_test_genomes(self):
defget_suffix(branch):
ifbranch=="":
return""
return"."+self.branch

guard_file=f"{self.test_genomes_path}/.complete"
ifnotos.path.isfile(guard_file):
quiet_remove("test_genomes")
URL='https://s3.amazonaws.com/pgap-data/test_genomes-{}{}.tgz'.format(self.use_version,get_suffix(self.branch))
print('Installing PGAP test genomes')
print(self.test_genomes_path)
print(URL)
install_url(URL, self.install_dir, self.args.quiet, self.args.teamcity, guard_file=None)
open(guard_file, 'a').close()

defupdate_self(self):
ifself.args.teamcity:
print("Not trying to update self, because the --teamcity flag is enabled.")
# Never update self when running teamcity
return

ifself.args.no_self_up:
print("Not trying to update self, because the --no-self-update flag is enabled.")
# Useful when locally editing and testing this file.
return

cur_file=sys.argv[0]
ifself.branch=="":
#ver = self.use_version
ver="prod"
else:
ver=self.branch
url=f"https://github.com/ncbi/pgap/raw/{ver}/scripts/pgap.py"
request=Request(url)

try:
withurlopen(request, timeout=self.timeout) asresponse:
new_pgap=response.read()
withopen(cur_file, "rb") asf:
old_pgap=f.read()

ifnew_pgap!=old_pgap:
print(f"Attempting to update <{cur_file}> ...", end='')
withopen(cur_file, "wb") asf:
f.write(new_pgap)
print("updated successfully.")
print("Please restart update.")
sys.exit()

exceptExceptionasexc:
print(exc)
print(f"Failed to update {cur_file}, ignoring")
print(f"Something has gone wrong, please manually download: {url}")
sys.exit()

defwrite_version(self):
filename=self.install_dir+"/VERSION"
withopen(filename, 'w', encoding='utf-8') asf:
f.write(u'{}\n'.format(self.use_version))


defremove_empty_files(rootdir):
forfinos.listdir(rootdir):
fullname=os.path.join(rootdir, f)
ifos.path.isfile(fullname) andos.path.getsize(fullname) ==0:
quiet_remove(fullname)

defcopy_genome_to_workspace(genome, original_workspace):
os.makedirs(original_workspace, exist_ok=True) 

# Check if the input file actually exists
ifnotos.path.exists(genome):
print(f"Error: The input genome file:{genome} does not exist.")
sys.exit(1) # Exit the script with an error code

filename=os.path.basename(genome)
new_genome_path=os.path.join(original_workspace, filename)

# Check if the file with the same name already exists in the workspace
ifos.path.exists(new_genome_path):
returnfilename

try:
# Attempt to copy the file
shutil.copy2(genome, new_genome_path)
exceptFileNotFoundError:
print(f"Error: The genome file {genome} does not exist. Can't copy it to {new_genome_path}")
sys.exit(1) # Exit the script with an error code

returnfilename

defcreate_simple_yaml_files(fasta_location, genus_species, output_dir):
# Note: The args are not validated here, as they are validated when the generated YAML files are ingested in the pipeline.
output_submol_yaml_filename=os.path.join(output_dir, 'submol.yaml')
withopen(output_submol_yaml_filename, 'w') asf:
f.write(f'''\
organism:
 genus_species: "{genus_species}"
'''
 )
output_input_yaml_filename=os.path.join(output_dir, 'input.yaml')
withopen(output_input_yaml_filename, 'w') asf:
f.write(f'''\
fasta:
 class: File
 location: {fasta_location}
submol:
 class: File
 location: submol.yaml
'''
 )

returnos.path.abspath(output_input_yaml_filename)

defvalidate_prefix(prefix):

"""
 Validates the given prefix to ensure it can be used in a filename on Linux, macOS, and Windows.

 Exits the program with an error message if the prefix is not valid.

 Valid Prefix:
 - Contains only alphanumeric characters, underscores, or hyphens.
 e.g., "my_prefix", "prefix123", "123_prefix", "prefix-123"

 Invalid Prefix:
 - Contains any characters other than alphanumeric characters, underscores, or hyphens.
 e.g., "my prefix", "prefix#", "prefix@", "prefix!"