diff --git a/test/test_siplib.py b/test/test_siplib.py
index 1efea2744c641e4bd5a32fecb71e8b09b1be7145..7a96efc82e963e544931d1f2d3f5334388238a9b 100755
--- a/test/test_siplib.py
+++ b/test/test_siplib.py
@@ -34,6 +34,9 @@ from lofar.lta.sip import siplib, validator, constants, ltasip
import os
import logging
logger = logging.getLogger(__name__)
+logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
+
+from lxml import etree
# d = os.path.dirname(os.path.realpath(__file__))
TMPFILE_PATH = "test_siplib.xml"
@@ -69,22 +72,123 @@ def create_basicdoc():
# project_telescope="LOFAR",
project_description = "awesome project",
project_coinvestigators = ["sidekick1", "sidekick2"],
- dataproduct = siplib.SimpleDataProduct(
- siplib.DataProductMap(
- type = "Unknown",
- identifier = dp_id,
- size = 1024,
- filename = "/home/paulus/test.h5",
- fileformat = "HDF5",
- storage_writer="Unknown",
- storage_writer_version="Unknown",
- process_identifier = pipe_id,
- checksum_md5 = "hash1",
- checksum_adler32 = "hash2",
- storageticket = "ticket"
- )
- )
- )
+ dataproduct = create_correlated_dataproduct() )
+
+
+def create_observation() -> siplib.Observation:
+ return siplib.Observation(observingmode="Interferometer",
+ instrumentfilter="10-70 MHz",
+ clock_frequency='160',
+ clock_frequencyunit="MHz",
+ stationselection="Core",
+ antennaset="HBA Zero",
+ timesystem="UTC",
+ stations=[siplib.Station.preconfigured("RS106", ["LBA"]),
+ siplib.Station.preconfigured("DE609", ["HBA"])],
+ numberofstations=5,
+ numberofsubarraypointings=5,
+ numberoftbbevents=5,
+ numberofcorrelateddataproducts=5,
+ numberofbeamformeddataproducts=5,
+ numberofbitspersample=5,
+ process_map=create_processmap(),
+ observationdescription="description",
+ channelwidth_frequency=160,
+ channelwidth_frequencyunit="MHz",
+ channelspersubband=5,
+ subarraypointings=[siplib.SubArrayPointing(
+ pointing=siplib.PointingAltAz(
+ az_angle=20,
+ az_angleunit="degrees",
+ alt_angle=30,
+ alt_angleunit="degrees",
+ equinox="SUN"
+ ),
+ beamnumber=5,
+ identifier=point_id,
+ measurementtype="All Sky",
+ targetname="Sun",
+ starttime="1980-03-23T10:20:15",
+ duration="P6Y3M10DT15H",
+ numberofprocessing=1,
+ numberofcorrelateddataproducts=2,
+ numberofbeamformeddataproducts=1,
+ relations=[siplib.ProcessRelation(
+ identifier=obs_id
+ )],
+ correlatorprocessing=siplib.CorrelatorProcessing(
+ integrationinterval=0.5,
+ integrationinterval_unit="ns",
+ channelwidth_frequency=160,
+ channelwidth_frequencyunit="MHz"
+ ),
+ coherentstokesprocessing=siplib.CoherentStokesProcessing(
+ rawsamplingtime=20,
+ rawsamplingtime_unit="ns",
+ timesamplingdownfactor=2,
+ samplingtime=10,
+ samplingtime_unit="ns",
+ stokes=["XX"],
+ numberofstations=1,
+ stations=[
+ siplib.Station.preconfigured("CS002", ["HBA0", "HBA1"])],
+ frequencydownsamplingfactor=2,
+ numberofcollapsedchannels=2,
+ channelwidth_frequency=160,
+ channelwidth_frequencyunit="MHz",
+ channelspersubband=122
+ ),
+ incoherentstokesprocessing=siplib.IncoherentStokesProcessing(
+ rawsamplingtime=20,
+ rawsamplingtime_unit="ns",
+ timesamplingdownfactor=2,
+ samplingtime=10,
+ samplingtime_unit="ns",
+ stokes=["XX"],
+ numberofstations=1,
+ stations=[
+ siplib.Station.preconfigured("CS003", ["HBA0", "HBA1"])],
+ frequencydownsamplingfactor=2,
+ numberofcollapsedchannels=2,
+ channelwidth_frequency=160,
+ channelwidth_frequencyunit="MHz",
+ channelspersubband=122
+ ),
+ flyseyeprocessing=siplib.FlysEyeProcessing(
+ rawsamplingtime=10,
+ rawsamplingtime_unit="ms",
+ timesamplingdownfactor=2,
+ samplingtime=2,
+ samplingtime_unit="ms",
+ stokes=["I"],
+ ),
+ nonstandardprocessing=siplib.NonStandardProcessing(
+ channelwidth_frequency=160,
+ channelwidth_frequencyunit="MHz",
+ channelspersubband=122
+ )
+ )],
+ transientbufferboardevents=["event1", "event2"]
+ )
+
+
+def create_correlated_dataproduct() -> siplib.CorrelatedDataProduct:
+ return siplib.CorrelatedDataProduct(
+ create_dataproductmap(),
+ subarraypointing_identifier=sap_id,
+ subband="1",
+ starttime="1980-03-23T10:20:15",
+ duration="P6Y3M10DT15H",
+ integrationinterval=10,
+ integrationintervalunit="ms",
+ central_frequency=160,
+ central_frequencyunit="MHz",
+ channelwidth_frequency=200,
+ channelwidth_frequencyunit="MHz",
+ channelspersubband=122,
+ stationsubband=2,
+ )
+
def create_processmap():
return siplib.ProcessMap(
@@ -299,23 +403,7 @@ class TestSIPlib(unittest.TestCase):
# add optional dataproduct item
logger.info("===\nAdded related correlated dataproduct:\n")
- logger.info(mysip.add_related_dataproduct(
- siplib.CorrelatedDataProduct(
- create_dataproductmap(),
- subarraypointing_identifier = sap_id,
- subband = "1",
- starttime = "1980-03-23T10:20:15",
- duration = "P6Y3M10DT15H",
- integrationinterval = 10,
- integrationintervalunit = "ms",
- central_frequency = 160,
- central_frequencyunit = "MHz",
- channelwidth_frequency = 200,
- channelwidth_frequencyunit = "MHz",
- channelspersubband = 122,
- stationsubband = 2,
- )
- ))
+ logger.info(mysip.add_related_dataproduct(create_correlated_dataproduct()))
# add optional dataproduct item
logger.info("===\nAdding related pixelmap dataproduct:\n")
@@ -414,102 +502,13 @@ class TestSIPlib(unittest.TestCase):
mysip.save_to_file(TMPFILE_PATH)
self.assertTrue(validator.validate(TMPFILE_PATH))
+
def test_observation(self):
mysip = create_basicdoc()
# add optional observation item
logger.info("===\nAdding observation:\n")
- logger.info(mysip.add_observation(siplib.Observation(observingmode = "Interferometer",
- instrumentfilter = "10-70 MHz",
- clock_frequency = '160',
- clock_frequencyunit = "MHz",
- stationselection = "Core",
- antennaset = "HBA Zero",
- timesystem = "UTC",
- stations = [siplib.Station.preconfigured("RS106", ["LBA"]),
- siplib.Station.preconfigured("DE609", ["HBA"])],
- numberofstations = 5,
- numberofsubarraypointings = 5,
- numberoftbbevents = 5,
- numberofcorrelateddataproducts = 5,
- numberofbeamformeddataproducts = 5,
- numberofbitspersample = 5,
- process_map = create_processmap(),
- observationdescription = "description",
- channelwidth_frequency = 160,
- channelwidth_frequencyunit = "MHz",
- channelspersubband = 5,
- subarraypointings = [siplib.SubArrayPointing(
- pointing = siplib.PointingAltAz(
- az_angle = 20,
- az_angleunit = "degrees",
- alt_angle = 30,
- alt_angleunit = "degrees",
- equinox = "SUN"
- ),
- beamnumber = 5,
- identifier = point_id,
- measurementtype = "All Sky",
- targetname = "Sun",
- starttime = "1980-03-23T10:20:15",
- duration = "P6Y3M10DT15H",
- numberofprocessing = 1,
- numberofcorrelateddataproducts = 2,
- numberofbeamformeddataproducts = 1,
- relations = [siplib.ProcessRelation(
- identifier = obs_id
- )],
- correlatorprocessing = siplib.CorrelatorProcessing(
- integrationinterval = 0.5,
- integrationinterval_unit = "ns",
- channelwidth_frequency = 160,
- channelwidth_frequencyunit = "MHz"
- ),
- coherentstokesprocessing = siplib.CoherentStokesProcessing(
- rawsamplingtime = 20,
- rawsamplingtime_unit = "ns",
- timesamplingdownfactor = 2,
- samplingtime = 10,
- samplingtime_unit = "ns",
- stokes = ["XX"],
- numberofstations = 1,
- stations = [siplib.Station.preconfigured("CS002", ["HBA0", "HBA1"])],
- frequencydownsamplingfactor = 2,
- numberofcollapsedchannels = 2,
- channelwidth_frequency = 160,
- channelwidth_frequencyunit = "MHz",
- channelspersubband = 122
- ),
- incoherentstokesprocessing = siplib.IncoherentStokesProcessing(
- rawsamplingtime = 20,
- rawsamplingtime_unit = "ns",
- timesamplingdownfactor = 2,
- samplingtime = 10,
- samplingtime_unit = "ns",
- stokes = ["XX"],
- numberofstations = 1,
- stations = [siplib.Station.preconfigured("CS003", ["HBA0", "HBA1"])],
- frequencydownsamplingfactor = 2,
- numberofcollapsedchannels = 2,
- channelwidth_frequency = 160,
- channelwidth_frequencyunit = "MHz",
- channelspersubband = 122
- ),
- flyseyeprocessing = siplib.FlysEyeProcessing(
- rawsamplingtime = 10,
- rawsamplingtime_unit = "ms",
- timesamplingdownfactor = 2,
- samplingtime = 2,
- samplingtime_unit = "ms",
- stokes = ["I"],
- ),
- nonstandardprocessing = siplib.NonStandardProcessing(
- channelwidth_frequency = 160,
- channelwidth_frequencyunit = "MHz",
- channelspersubband = 122
- )
- )],
- transientbufferboardevents = ["event1", "event2"]
- )))
+ mysip.add_observation(create_observation())
+ logger.info(mysip)
mysip.save_to_file(TMPFILE_PATH)
self.assertTrue(validator.validate(TMPFILE_PATH))
@@ -637,6 +636,43 @@ class TestSIPlib(unittest.TestCase):
mysip.save_to_file(TMPFILE_PATH)
self.assertTrue(validator.validate(TMPFILE_PATH))
+ def test_type_attributes(self):
+ '''Check if the types of complexTypes like CorrelatedDataproduct, Observation, etc are annotated using xml attributes.
+ The LTA catalogue code expects these attributes like xsi:type="sip:CorrelatedDataProduct" '''
+
+ mysip = create_basicdoc()
+ mysip.add_observation(create_observation())
+
+ xml_doc = mysip.get_prettyxml()
+ logger.info(xml_doc)
+
+ # The LTA catalogue code expects these attributes like xsi:type="sip:CorrelatedDataProduct"
+ # So, use etree XML DOM parsing for correct lookup of the interesting elements,
+ # but also check explicitly if the attribute as plain text is as the LTA code expects it.
+ # (There many ways to use a namespace, and parsers like etree have a tendency to interpret the plain text and substitute abbreviations with fqdn namespaces)
+ xml_root = etree.fromstring(xml_doc) # for DOM parsing
+ xml_lines = xml_doc.split('\n') # for plain text parsing
+
+ # check dataproduct type attribute via etree
+ dataproduct = xml_root.xpath('dataProduct')[0]
+ self.assertTrue('{http://www.w3.org/2001/XMLSchema-instance}type' in dataproduct.attrib)
+ self.assertEqual('sip:CorrelatedDataProduct', dataproduct.attrib['{http://www.w3.org/2001/XMLSchema-instance}type'])
+
+ # check dataproduct type attribute via plain text
+ dataproduct_line = [l for l in xml_lines if l.lstrip().startswith('<dataProduct ')][0]
+ self.assertTrue('xsi:type="sip:CorrelatedDataProduct"' in dataproduct_line)
+
+
+ # check observation type attribute via etree
+ observation = xml_root.xpath('observation')[0]
+ self.assertTrue('{http://www.w3.org/2001/XMLSchema-instance}type' in observation.attrib)
+ self.assertEqual('sip:Observation', observation.attrib['{http://www.w3.org/2001/XMLSchema-instance}type'])
+
+ # check observation type attribute via plain text
+ observation_line = [l for l in xml_lines if l.lstrip().startswith('<observation ')][0]
+ self.assertTrue('xsi:type="sip:Observation"' in observation_line)
+
+
# run tests if main
if __name__ == '__main__':
unittest.main()