Документ взят из кэша поисковой машины. Адрес оригинального документа : http://www.stsci.edu/spst/UnixTransition/doc/cl_revision_letter_check.py Дата изменения: Fri Apr 8 12:46:10 2016 Дата индексирования: Mon Apr 11 06:50:48 2016 Кодировка: Поисковые слова: обвмадеойс нефептощи рпфплпч

#
#MODULE cl_revision_letter_check
#
#***********************************************************************
"""

**PURPOSE** --
Checks that the revision letter(s) associated with the command load(s)
(CL) are not used by command loads that have already been delivered
for similar timeframes.

**DEVELOPER** --
Danny Jones

**MODIFICATION HISTORY** --

o Initial implementation 2/2/16

"""
#***********************************************************************
from abstract_check import abstract_check
from checklist_init import PASS, FAIL

__version__ = '16.02.02'


class cl_revision_letter_check(abstract_check):
"""Make sure revision letter(s) for command load(s) differ from those already delivered
"""
def run(self):
import spss_sys_util
from time_util import spss_time
import stpydb

# Initialization
abstract_check.run(self)
self.pf_status = PASS
mscl_object = self.my_checklist.mscl
pass_sms_name = mscl_object.my_ms.get_ms_name()
spss_db = spss_sys_util.get_environ_variable("SPSS_DB")[0]
db_conn = stpydb.stpydb(dbmsName=spss_db)

# Determine whether the PASS run to be checked has already been
# delivered. If so, record the time the mission schedule was delivered
# as a reference time to search for products that may have conflicting
# revision letters. If not, use the current time as the reference time.
db_query = "select date_recv from pass_catalog " + \
" where pass_sms_nm = '%s'" % pass_sms_name.upper() + \
" and file_type = 'MS'"
db_conn.query(db_query)
query_result = []
get_deliveries_before = spss_time() # current time if not delivered
already_delivered = False
while db_conn.execute(query_result):
get_deliveries_before = query_result[0]
already_delivered = True

# Determine the products that may have conflicting revision letters.
# These are CL products (tagged in the pass_catalog relation as
# "MTL" for Mission Timeline) that were delivered before the reference
# time and belong to an SMS that ends after the reference time. This
# allows for the check to be made on PASS runs that have already
# been delivered (by eliminating any products that were delivered
# AFTER) and for intercept PASS runs (by including the currently
# executing PASS run along with any PASS runs that were previously
# delivered an are now deemed superseded.)
db_query = "select pass_sms_nm, pass_fil_dir" + \
" from pass_catalog" + \
" where date_recv < '%s'" % get_deliveries_before + \
" and end_time >= '%s'" % get_deliveries_before + \
" and file_type = 'MTL'"
db_conn.query(db_query)
query_result = []
delivered_pass_names = set()
delivered_cl_dirs = []
while db_conn.execute(query_result):
delivered_pass_names.add(query_result[0])
delivered_cl_dirs.append(query_result[1])
#db_conn.close()

# Find the unique revision letters used by every SPC file in the
# command load director(ies), separating them into 2 sets--one
# for the products being checked, the other for the previously
# delivered products.
cl_dirs = [cl.get_output_directory() for cl in mscl_object.my_cls]
curr_rev_letters = self.revision_letter_set(cl_dirs)
delivered_rev_letters = self.revision_letter_set(delivered_cl_dirs)

# An intersection of these sets show the same revision letter being
# used. This may confuse those responsible for uplinking the CLs
# to the spacecraft, so this would FAIL the test. Also fail the test
# if no command loads have been generated for the run being checked.
if (not curr_rev_letters) or \
(curr_rev_letters.intersection(delivered_rev_letters)):
self.pf_status = FAIL

# Return the pass/fail status write diagnostic information to
# the established output file.
with open(self.outfile, "wt") as outfile:
if not curr_rev_letters:
outfile.write("ERROR - no SPC files found for command loads " +
"associated with PASS SMS name %s." % pass_sms_name)
elif already_delivered:
outfile.write("%s has already been delivered at %s.\n" %
(pass_sms_name, get_deliveries_before))
outfile.write("Products whose revision letters may have " +
"conflicted at that time belonged to:\n")
else:
outfile.write("Products that may have conflicting revision " +
"letters belong to:\n")
outfile.write(" %s\n" % ", ".join(delivered_pass_names))
outfile.write("Revision letter(s) used by those products: %s\n" %
", ".join(sorted(delivered_rev_letters)))
outfile.write("Revision letter(s) used by current products: %s\n" %
", ".join(sorted(curr_rev_letters)))
if self.pf_status == PASS:
outfile.write("\nCheck passed.\n")
else:
outfile.write("\nCheck failed.\n")
return self.pf_status

def revision_letter_set(self, cl_dirs):
""" Given a sequence of command load directory names, return the
set of unique revision letters used to generate the filenames
of the SPC files in those directories. The revision letter is the
last character before the ".spc" extension.
"""

import glob
import os
rev_letters = set()
for cl_dir in cl_dirs:
spc_files = glob.glob(os.path.join(cl_dir, "*.spc"))
rev_letters.update([os.path.splitext(fname)[0][-1].upper()
for fname in spc_files])
return rev_letters


def run(pass_sms_name):
"""Run the check stand alone.
"""
import passchecklist
clist = passchecklist.Pass(pass_sms_name, 1)
check = cl_revision_letter_check(clist)
print "\n Output sent to file " + check.outfile
check.run()

# to run on command line
if __name__ == '__main__':
import sys
run(*sys.argv[1:])