Changeset 29140 for branches/sc_branches/psps_testing/testers/fits/p2.py

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

File:

• branches/sc_branches/psps_testing/testers/fits/p2.py (modified) (1 diff)

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()