Changeset 29140

Timestamp:

Sep 10, 2010, 9:39:28 AM (16 years ago)

Author:

Serge CHASTEL

Message:

Unstable. Added logging (no tused yet. p2 is currently very slow

Location:

branches/sc_branches/psps_testing

Files:

• psi/psi_inquisitor.py (modified) (4 diffs)
• testers/fits/p2.py (modified) (1 diff)
• testers/fits_file.py (modified) (3 diffs)
• utilities/psps_logger.py (added)
• utilities/util.py (added)

branches/sc_branches/psps_testing/psi/psi_inquisitor.py

@@ -2 +2 @@
 from suds.xsd.doctor import Import, ImportDoctor
 from suds import sudsobject
+from utilities.util import convert
+import unicodedata
 
 class PsiInquisitor:
@@ -42 +44 @@
         >>> # we cannot guarantee the order in result
         >>> print result['items']
-        [{u'[frameID]': u'105439'}]
+        [{'frameID': 105439}]
         >>> print result['fields']
-        [u'[frameID]']
+        ['[frameID]']
         >>> print result['types']
-        {u'[frameID]': u'Integer'}
+        {'[frameID]': 'Integer'}
         >>> # This test has to be completed with exhaustive ones...
         >>> # e.g. unit tests against the csr data base.
@@ -62 +64 @@
         line_count = 0
         for line in lines:
+            # Unicode normalization comes from
+            # http://www.peterbe.com/plog/unicode-to-ascii. I don't
+            # know where the NFKD means...
+            line = unicodedata.normalize('NFKD', line).encode('ascii','ignore')
             if first_line_not_seen:
                 # Build up the field list and the types dictionary
@@ -85 +91 @@
                 element = {}
                 for i in range(len(values)):
-                    element[contents['fields'][i]] = values[i]
+                    field_name = contents['fields'][i].replace('[', 
+                                                               '').replace(']', 
+                                                                           '')
+                    element[field_name] = convert(types_dictionary[contents['fields'][i]],
+                                                  values[i])
                 contents['items'].append(element)
         return contents

branches/sc_branches/psps_testing/testers/fits/p2.py

@@ -1 @@
-
+from psi.psi_inquisitor import PsiInquisitor
+import pyfits
+from StringIO import StringIO
+from utilities.util import match
 
 class P2FitsTester:
     """
+    The content of a P2 frame data is detailed in PSDC-940-006-01,
+    paragraph 3.6.3, item 1).
+
     This is the class that tests the contents of a P2 FITS file
-    provided by IPP. The contents can be checked against the database
+    provided by IPP. The contents are checked against the database
     through SOAP queries.
+
+    Note that the purpose of this class is NOT to test the FITS
+    validity (which can be checked using some pyfits options).
     """
-    def __init__(self, fitsFile):
-        self.fits = fitsFile
+    def __init__(self, fits_file, psi_configuration):
+        """
+        Creates a P2FitsTester object.
+
+        >>> import pyfits
+        >>> fits_file = pyfits.open('data/psut/ok/fits/p2_was_00105439.FITS')
+        >>> psi_configuration = 'psi.web01_configuration'
+        >>> p2ft = P2FitsTester(fits_file, psi_configuration)
+        """
+        self.fits = fits_file
+        self.psi_inquisitor = PsiInquisitor(psi_configuration)
 
     def test(self):
-        return ("BARF", None)
+        """
+        Tests a FITS file supposed to contain a P2 frame data set.
+
+        >>> import pyfits
+        >>> fits_file = pyfits.open('data/psut/ok/fits/p2_was_00105439.FITS')
+        >>> psi_configuration = 'psi.web01_configuration'
+        >>> p2ft = P2FitsTester(fits_file, psi_configuration)
+        >>> (message, values) = p2ft.test()
+        >>> print message # doctest:+ELLIPSIS
+        PSDC-940-006-01, 3.6.3, 1): OK
+        PSDC-940-006-01, 3.6.6.8: OK
+        PSDC-940-006-01, TBD_test_detections_frames: OK (frame 00 - SELECT * FROM ImageMeta WHERE imageID = 10543901)
+        PSDC-940-006-01, TBD_test_detections_frames: OK (frame 01 - SELECT * FROM ImageMeta WHERE imageID = 10543902)
+        ...
+        PSDC-940-006-01, TBD_test_detections_frames: OK (frame 57 - SELECT * FROM ImageMeta WHERE imageID = 10543976)
+        >>> print values
+        None
+        """
+        report = StringIO()
+        # 1. There is no image for the PrimaryHDU (at 0)
+        (messages, other) = self.test_primary()
+        report.write(messages)
+        # 2. Look at the FrameMeta information
+        (messages, values) = self.test_frame_meta()
+        report.write('\n' + messages)
+        if values is None:
+            return (report.getvalue(), None)
+        # 3. There should be values['ota'] sequences of [ImageMeta,
+        # Detection, SkinnyObject, ObjectCalColor] frames
+        frameID = values['frameID']
+        nOTA = values['nOTA']
+        if len(self.fits) != 4*nOTA + 2:
+            report.write('\nInvalid number of OTA: from FITS file: %d / from FrameMeta %d'
+                         % (len(self.fits)-2/4, nOTA))
+            return (report.getvalue(), None)
+        for ota_index in range(nOTA):
+            (messages, values) = self.test_detections_frames(frameID, ota_index) 
+            report.write('\n' + messages)
+        return (report.getvalue(), None)
+
+    def test_detections_frames(self, frameID, ota_index):
+        """
+        Tests the contents of the 4-frame sequence [ImageMeta,
+        Detection, SkinnyObject, ObjectCalColor].
+        Those contents can be queried by: TODO
+        
+        """
+        report = StringIO()
+        # Check that the 4 sequence is correct
+        requirement = 'PSDC-940-006-01, TBD_test_detections_frames: '
+        fits_index = 4*ota_index+2
+        if self.fits[fits_index].name != 'IMAGEMETA':
+            return (requirement + 'KO (Not an ImageMeta frame)', None)
+        if self.fits[fits_index+1].name != 'DETECTION':
+            return (requirement + 'KO (Not a Detection frame)', None)
+        if self.fits[fits_index+2].name != 'SKINNYOBJECT':
+            return (requirement + 'KO (Not a SkinnyObject frame)', None)
+        if self.fits[fits_index+3].name != 'OBJECTCALCOLOR':
+            return (requirement + 'KO (Not a ObjectCalColor frame)', None)
+        # Check ImageMeta values (!We get imageID from the FITS file
+        # since they are not sorted)
+        imageID = self.fits[fits_index].data.field('imageID')[0]
+        query = 'SELECT * FROM ImageMeta WHERE imageID = %s' % str(imageID)
+        answer = self.psi_inquisitor.query(query)
+        (message, status) = match(self.fits[fits_index].data, answer, requirement)
+        if status != 0:
+            return (message + ' (frame %02d - %s)' % (ota_index, query), None)
+        report.write(message)
+        # Extract the number of detections
+        nDetect = int(self.fits[fits_index].data.field('nDetect')[0])
+        # Check if we have the same number from PSI
+        query = 'SELECT count(*) AS HowMany FROM Detection WHERE imageID = %s' % str(imageID)
+        howMany = self.psi_inquisitor.query(query)['items'][0]['HowMany']
+        if nDetect != howMany:
+            report.write('\nGot %d Detection items from PSI instead of %d for image %s'
+                         % (howMany, nDetect, str(imageID)) )
+        # Get the nDetect detections (they are assumed to be sorted in the FITS file)
+        query = 'SELECT * FROM Detection WHERE imageID = %s ORDER BY detectID' % str(imageID)
+        answer = self.psi_inquisitor.query(query)
+        for detection_index in range(nDetect):
+            (message, status) = match(self.fits[fits_index+1].data, answer, 
+                                      requirement, detection_index)
+            report.write('\n' + message)
+        return (report.getvalue() + ' (frame %02d - %s)' % (ota_index, query), None)
+
+    def test_frame_meta(self):
+        """
+        Tests the contents of the FrameMeta table. 
+        - 48 columns x 1 row
+        - This table contains values that are useful, namely: frameID,
+          nOTA (number of OTA).
+        - The data can be found in PSI using the query:
+                   SELECT * FROM FrameMeta WHERE frameID = <frameID>;
+        """
+        requirement = 'PSDC-940-006-01, 3.6.6.8: '
+        if self.fits[1].header['NAXIS'] != 2:
+            return (requirement + 'KO (NAXIS != 2)', None)
+        if self.fits[1].header['NAXIS2'] != 1:
+            return (requirement + 'KO (NAXIS2 != 1)', None)
+        if self.fits[1].header['TFIELDS'] != 48:
+            return (requirement + 'KO (TFIELDS != 48)', None)
+        frameID = self.fits[1].data.field('frameID')[0]
+        query = "SELECT * FROM FrameMeta WHERE frameID = %d" % frameID
+        answer = self.psi_inquisitor.query(query)
+        (message, status) = match(self.fits[1].data, answer, requirement)
+        if status != 0:
+            return (message, None)
+        else:
+            return (message, 
+                    {'frameID': int(self.fits[1].data.field('frameID')[0]),
+                     'nOTA': int(self.fits[1].data.field('nOTA')[0]) })
+    
+    def test_primary(self):
+        """
+        Check that the primary header i.e. fits[0] has no dimension.
+        """
+        requirement = 'PSDC-940-006-01, 3.6.3, 1): '
+        if self.fits[0].header['NAXIS'] != 0:
+            return (requirement + 'KO', None)
+        # Add more tests?
+        return (requirement + 'OK', None)
+
+if __name__ == '__main__':
+    import doctest
+    doctest.testmod()

branches/sc_branches/psps_testing/testers/fits_file.py

@@ -40 +40 @@
         FileManipulation.system('/bin/rm -rf %s' % FitsFileTester._TMPDIR)
 
-    def __init__(self, fitsname, basename, tempDir, fitsType):
+    def __init__(self, fitsname, basename, tempDir, fitsType, 
+                 psi_configuration = 'psi.web01_configuration'):
         """
         Creates an instance of a FitsFileTester
@@ -58 +59 @@
         self.fitsType = fitsType
         self.fitsFile = pyfits.open(self.filename)
+        self.psi_configuration = psi_configuration
 
     def __del__(self):
@@ -90 +92 @@
         """
         if self.fitsType == "P2":
-            return P2FitsTester(self.fitsFile).test()
+            return P2FitsTester(self.fitsFile, self.psi_configuration).test()
         else:
             raise Exception("Unsupported FITS type [%s]" % self.fitsType)