Fix the mapping between CWL inputs, which are in JSON format, and Python...

Fix the mapping between CWL inputs, which are in JSON format, and Python types, for CWL steps that contain inline Python script.

The solution is to convert the CWL `$(inputs)` once to a Python `dict`, by using the `json.loads()` function, passing the `$(inputs)` as a raw Python string. Next, any input argument can be accessed as an entry in a Python `dict`.

Also, to avoid potential "Argument list too long" errors, all inline Python scripts that expand the `$(inputs)` argument are now automatically turned into a script *file*, instead of being passed as a (potentially huge) string on the command line.


Former-commit-id: 250aa9d1

lofar-cwl/steps/DP3.ParsetGenerator.cwl

@@ -4,51 +4,51 @@ cwlVersion: v1.1
 class: CommandLineTool
 id: generic_step
 
-requirements:
-  - class: InlineJavascriptRequirement
-
 baseCommand:
   - python3
+  - generic_step.py
 
-arguments:
- - prefix: -c
-   valueFrom: |
-        import sys
-
-        true = True
-        false = False
-        null = None
-        steps_list = $(inputs.steps)
-        steps_id_list = list(map(lambda item: item['step_id'], steps_list))
-
-        for step in steps_list:
-            try:
-                step_id = step.get('step_id')
-                step_type = step.get('step_type')
-                parameters = step.get('parameters')
-
-
-                line = '.'.join([step_id, 'type'])
-                line += '=%s' % step_type
-                print(line)
-
-                for parameter_name, parameter in parameters.items():
-                    if isinstance(parameter, dict):
-                        if parameter['class'] in ['File', 'Directory']:
-                            parameter = parameter['basename']
-                        else:
-                            raise NotImplemented()
-                    if parameter is None:
-                        continue
-                    line = '.'.join([step_id, parameter_name])
-                    line = '%s=%s' % (line, parameter)
-
-                    print(line)
-            except KeyError as e:
-                print('invalid input for step ', step, ':', e)
-                sys.exit(1)
-        last_step_line = 'steps=[%s,count]' % ', '.join(steps_id_list)
-        print(last_step_line)
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: generic_step.py
+        entry: |
+          import sys
+          import json
+
+          inputs = json.loads(r"""$(inputs)""")
+          steps_list = inputs['steps'] if inputs['steps'] else []
+          steps_id_list = list(map(lambda item: item['step_id'], steps_list))
+
+          for step in steps_list:
+              try:
+                  step_id = step.get('step_id')
+                  step_type = step.get('step_type')
+                  parameters = step.get('parameters')
+
+
+                  line = '.'.join([step_id, 'type'])
+                  line += '=%s' % step_type
+                  print(line)
+
+                  for parameter_name, parameter in parameters.items():
+                      if isinstance(parameter, dict):
+                          if parameter['class'] in ['File', 'Directory']:
+                              parameter = parameter['basename']
+                          else:
+                              raise NotImplemented()
+                      if parameter is None:
+                          continue
+                      line = '.'.join([step_id, parameter_name])
+                      line = '%s=%s' % (line, parameter)
+
+                      print(line)
+              except KeyError as e:
+                  print('invalid input for step ', step, ':', e)
+                  sys.exit(1)
+          last_step_line = 'steps=[%s,count]' % ', '.join(steps_id_list)
+          print(last_step_line)
 inputs:
   - id: steps
     type: Any[]?

lofar-cwl/steps/LoSoTo.Replicateonaxis.cwl

@@ -15,15 +15,18 @@ requirements:
       - entryname: "replicate_axis.py"
         entry: |
           import sys
+          import json
           from losoto.h5parm import openSoltab
           from losoto.operations import replicateonaxis
 
-          soltab_address = "$(inputs.soltab)"
+          inputs = json.loads(r"""$(inputs)""")
+
+          soltab_address = inputs['soltab']
           soltab = openSoltab(sys.argv[1], address=soltab_address)
 
-          axisReplicate = "$(inputs.axisReplicate)"
-          fromCell = "$(inputs.fromCell !== null ? inputs.fromCell : 0)"
-          updateWeights = "$(inputs.updateWeights? inputs.updateWeights : 'True')"
+          axisReplicate = inputs['axisReplicate']
+          fromCell = inputs['fromCell'] if inputs['fromCell'] else 0
+          updateWeights = inputs['updateWeights']
 
           replicateonaxis.run(soltab, axisReplicate=axisReplicate,
                                       fromCell=fromCell,

steps/check_unflagged_fraction.cwl

@@ -5,6 +5,7 @@ $namespaces:
 id: check_unflagged_fraction
 baseCommand:
   - python3
+  - check_fraction.py
 inputs:
     - id: msin
       type: Directory
@@ -15,35 +16,39 @@ inputs:
       type: float?
       default: 0.5
       doc: Fraction of flagged data in a MS to be acceptable
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entry: $(inputs.msin)
+        writable: true
+      - entryname: check_fraction.py
+        entry: |
+          import sys
+          import json
+          import shutil
+          import os
+
+          from check_unflagged_fraction import main as check_unflagged_fraction
+
+          ms = sys.argv[1]
+          inputs = json.loads(r"""$(inputs)""")
+          min_fraction = inputs['min_fraction']
 
-label: check_unflagged_fraction
-arguments:
-  - '-c'
-  - |
-    import sys
-    import json
-    import shutil
-    import os
+          output = check_unflagged_fraction(ms, min_fraction=min_fraction, print_fraction=True)
 
-    from check_unflagged_fraction import main as check_unflagged_fraction
+          filename           = os.path.basename(output['flagged'])
+          unflagged_fraction = output['unflagged_fraction']
 
-    ms = sys.argv[1]
-    min_fraction = float($(inputs.min_fraction))
-    
-    output = check_unflagged_fraction(ms, min_fraction=min_fraction, print_fraction=True)
+          if filename != 'None':
+              shutil.move(filename, filename.lstrip('out_'))
 
-    filename           = os.path.basename(output['flagged'])
-    unflagged_fraction = output['unflagged_fraction']
+          cwl_output = {}
+          cwl_output['unflagged_fraction'] = unflagged_fraction
 
-    if filename != 'None':
-        shutil.move(filename, filename.lstrip('out_'))
-        
-    cwl_output = {}
-    cwl_output['unflagged_fraction'] = unflagged_fraction
+          with open('./out.json', 'w') as fp:
+              json.dump(cwl_output, fp)
 
-    with open('./out.json', 'w') as fp:
-        json.dump(cwl_output, fp)
-        
 outputs:
   - id: msout
     type: Directory[]
@@ -59,16 +64,8 @@ outputs:
     type: File[]
     outputBinding:
       glob: 'check_unflagged_fraction*.log'
-        
-requirements:
-  - class: InlineJavascriptRequirement
-  - class: InitialWorkDirRequirement
-    listing:
-      - entry: $(inputs.msin)
-        writable: true
-
 hints:
   - class: DockerRequirement
     dockerPull: lofareosc/prefactor3-cwl
 stdout: check_unflagged_fraction.log
-stderr: check_unflagged_fraction_err.log
+stderr: check_unflagged_fraction_err.log
\ No newline at end of file

steps/compare_station_list.cwl

@@ -5,6 +5,7 @@ $namespaces:
 id: compare_station_list
 baseCommand:
   - python3
+  - script.py
 inputs:
     - id: msin
       type: Directory[]
@@ -22,27 +23,29 @@ inputs:
       type: string?
       default: '*&'
       doc: Filter these baselines for the comparison
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: script.py
+        entry: |
+          import sys
+          import json
+          from compareStationList import main as compareStationList
 
-label: compareStationList.py
-arguments:
-  - '-c'
-  - |
-    import sys
-    import json
-    from compareStationList import main as compareStationList
-    
-    mss = sys.argv[1:]
-    h5parmdb = $(inputs.h5parmdb)['path']
-    solset_name = '$(inputs.solset_name)'
-    filter = '$(inputs.filter)'
+          mss = sys.argv[1:]
+          inputs = json.loads(r"""$(inputs)""")
+          h5parmdb = inputs['h5parmdb']['path']
+          solset_name = inputs['solset_name']
+          filter = inputs['filter']
 
-    output = compareStationList(mss, h5parmdb, solset_name, filter)
+          output = compareStationList(mss, h5parmdb, solset_name, filter)
 
-    filter_out = output['filter']
-    cwl_output = {"filter_out": filter_out}
+          filter_out = output['filter']
+          cwl_output = {"filter_out": filter_out}
 
-    with open('./out.json', 'w') as fp:
-        json.dump(cwl_output, fp)
+          with open('./out.json', 'w') as fp:
+              json.dump(cwl_output, fp)
 outputs:
   - id: filter_out
     type: string
@@ -54,12 +57,8 @@ outputs:
     type: File[]
     outputBinding:
       glob: 'compareStationList*.log'
-        
-requirements:
-  - class: InlineJavascriptRequirement
-
 hints:
   - class: DockerRequirement
     dockerPull: lofareosc/prefactor3-cwl
 stdout: compareStationList.log
-stderr: compareStationList_err.log
+stderr: compareStationList_err.log
\ No newline at end of file

steps/filter_ms_group.cwl

@@ -42,31 +42,32 @@ requirements:
         import json
         import os
 
-        group_id = "$(inputs.group_id)"
+        inputs = json.loads(r"""$(inputs)""")
+        group_id = inputs['group_id']
         json_file = sys.argv[1]
         ms_list = sys.argv[2:]
 
         ms_by_name = { ms.split(os.path.sep)[-1]:
                        {'class':'Directory', 'path': ms} for ms in ms_list}
-                       
+
         output_file = 'selected_ms.json'
-        
+
         with open(json_file, 'r') as f_stream:
             selected_ms = json.load(f_stream)[group_id]
-            
+
         selected_ms = [os.path.basename(ms_name) for ms_name in selected_ms]
         cwl_output  = {'selected_ms': selected_ms}
 
         with open('./out.json', 'w') as fp:
             json.dump(cwl_output, fp)
-            
+
         selected_ms = [ms_by_name[ms_name] for ms_name in selected_ms if ms_name != 'dummy.ms']
-            
-        
+
+
         with open(output_file, 'w') as f_stream:
             json.dump(selected_ms, f_stream)
-              
-        
+
+
 
 hints:
   - class: InitialWorkDirRequirement
@@ -75,4 +76,4 @@ hints:
         writable: true
 
 stdout: filter_ms_by_group.log
-stderr: filter_ms_by_group_err.log
+stderr: filter_ms_by_group_err.log
\ No newline at end of file

steps/findRefAnt_join.cwl

@@ -19,7 +19,7 @@ inputs:
       type: string?
       default: 'NONE'
       doc: Provide state information for collecting antenna statistics
-      
+
 label: findRefAnt_join
 
 outputs:
@@ -36,8 +36,8 @@ outputs:
   - id: logfile
     type: File
     outputBinding:
-      glob: findRefAnt.log    
-        
+      glob: findRefAnt.log
+
 requirements:
   - class: InlineJavascriptRequirement
   - class: InitialWorkDirRequirement
@@ -50,12 +50,13 @@ requirements:
             import json
             import re
             import ast
-    
+
+            inputs = json.loads(r"""$(inputs)""")
             with open('input.json', 'r') as f_stream:
                 flagged_fraction_dict_list = json.load(f_stream)
-            filter_station = '$(inputs.filter_station)'
+            filter_station = inputs['filter_station']
             no_station_selected = True
-    
+
             while no_station_selected:
                 print('Applying station filter ' + str(filter_station))
                 flagged_fraction_data = {}
@@ -77,7 +78,7 @@ requirements:
 
             flagged_fraction_list = []
             sorted_stations = sorted(flagged_fraction_data.keys())
-            
+
             flagged_fraction_antenna = {}
 
             for antenna in sorted_stations:
@@ -93,15 +94,15 @@ requirements:
             refant = flagged_fraction_data[min_flagged_fraction][0]
             print('Selected station ' + str(refant) + ' as reference antenna. Fraction of flagged data is ' + '{:>3}'.format('{:.1f}'.format(min_flagged_fraction) + '%'))
 
-            flagged_fraction_antenna['state'] = '$(inputs.state)'
-            
+            flagged_fraction_antenna['state'] = inputs['state']
+
             cwl_output = {'refant': str(refant)}
 
             with open('./out.json', 'w') as fp:
                 json.dump(cwl_output, fp)
-                
+
             with open('./flagged_fraction_antenna.json', 'w') as fp:
                 json.dump(flagged_fraction_antenna, fp)
-            
+
 stdout: findRefAnt.log
 stderr: findRefAnt_err.log

steps/find_skymodel_cal.cwl

@@ -7,7 +7,7 @@ baseCommand:
   - python3
 inputs:
     - id: msin
-      type: 
+      type:
        - Directory
        - Directory[]
       doc: MS containing the calibrator
@@ -45,8 +45,8 @@ outputs:
   - id: logfile
     type: File
     outputBinding:
-      glob: find_skymodel_cal.log    
-        
+      glob: find_skymodel_cal.log
+
 requirements:
   - class: InlineJavascriptRequirement
   - class: InitialWorkDirRequirement
@@ -56,36 +56,37 @@ requirements:
             import sys
             import os
             import shutil
-            null = None
-            
+            import json
+
             from unittest.mock import MagicMock
             sys.modules['lofarpipe.support.data_map'] = MagicMock()
-            
-            
+
+
             from find_skymodel_cal import main as find_skymodel
-            import json
+
             mss = sys.argv[1:]
-            
-            skymodels = $(inputs.skymodels)
-            max_separation_arcmin = $(inputs.max_separation_arcmin)
+            inputs = json.loads(r"""$(inputs)""")
+
+            skymodels = inputs['skymodels']
+            max_separation_arcmin = inputs['max_separation_arcmin']
             max_separation_arcmin = 1.0 if max_separation_arcmin is None else max_separation_arcmin
-            extension = "$(inputs.skymodels_extension)"
+            extension = inputs['skymodels_extension']
             if skymodels is None:
                 skymodels = os.path.expandvars("$PREFACTOR_DATA_ROOT/skymodels")
             else:
                 skymodels = skymodels["path"]
             output = {}
-            if extension != 'null':
+            if extension is not None:
                 output = find_skymodel(mss, skymodels, extension, max_separation_arcmin=max_separation_arcmin)
             else:
                 output = find_skymodel(mss, skymodels, max_separation_arcmin=max_separation_arcmin)
-            
+
             skymodel_path = output['SkymodelCal']
             skymodel_name = output['SkymodelName']
             skymodel_path = shutil.copy(skymodel_path, os.getcwd())
             cwl_output = {"file":{'class': 'File', 'path': skymodel_path},
                           "skymodel_name": skymodel_name}
-            
+
             with open('./out.json', 'w') as fp:
                 json.dump(cwl_output, fp)
 

steps/find_skymodel_target.cwl

@@ -5,6 +5,7 @@ $namespaces:
 id: find_skymodel_target_py
 baseCommand:
   - python3
+  - find_sky.py
 inputs:
     - id: msin
       type: Directory[]
@@ -26,35 +27,37 @@ inputs:
       type: boolean?
       doc: Download a new skymodel if given path is empty
       default: true
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: find_sky.py
+        entry: |
+          import sys
+          import shutil
+          import os
+          import json
+
+          from download_skymodel_target import main as download_skymodel_target
+
+          mss = sys.argv[1:]
+          inputs = json.loads(r"""$(inputs)""")
 
-label: find_skymodel_target.py
-arguments:
-  - '-c'
-  - |
-    import sys
-    import shutil
-    import os
-    null = None
+          SkymodelPath = inputs['SkymodelPath']
+          if SkymodelPath is None:
+              SkymodelPath = os.getcwd() + "/target.skymodel"
+          else:
+              SkymodelPath = SkymodelPath["path"]
 
-    from download_skymodel_target import main as download_skymodel_target
+          Radius = inputs['Radius']
+          Source = inputs['Source']
+          DoDownload = str(inputs['DoDownload'])
 
-    mss = sys.argv[1:]
+          output = download_skymodel_target(mss, SkymodelPath, Radius, DoDownload, Source)
 
-    SkymodelPath = $(inputs.SkymodelPath)
-    if SkymodelPath is None:
-        SkymodelPath = os.getcwd() + "/target.skymodel"
-    else:
-        SkymodelPath = SkymodelPath["path"]
+          if "target.skymodel" not in SkymodelPath:
+             shutil.copyfile(SkymodelPath, "target.skymodel")
 
-    Radius = $(inputs.Radius)
-    Source = "$(inputs.Source)"
-    DoDownload = "$(inputs.DoDownload)"
-    
-    output = download_skymodel_target(mss, SkymodelPath, Radius, DoDownload, Source)
-    
-    if "target.skymodel" not in SkymodelPath:
-       shutil.copyfile(SkymodelPath, "target.skymodel")
-   
 outputs:
   - id: skymodel
     type: File?
@@ -63,13 +66,9 @@ outputs:
   - id: logfile
     type: File
     outputBinding:
-      glob: find_skymodel_target.log    
-        
-requirements:
-  - class: InlineJavascriptRequirement
-
+      glob: find_skymodel_target.log
 hints:
   - class: DockerRequirement
     dockerPull: lofareosc/prefactor3-cwl
 
-stdout: find_skymodel_target.log
+stdout: find_skymodel_target.log
\ No newline at end of file

steps/identify_bad_antennas_join.cwl

@@ -5,6 +5,7 @@ $namespaces:
 id: identify_bad_antennas_join
 baseCommand:
   - python3
+  - indentify_bad_antennas.py
 inputs:
     - id: flaggedants
       type: string[]?
@@ -14,30 +15,34 @@ inputs:
       type: string?
       default: '*&'
       doc: Filter these baselines for the comparison
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: indentify_bad_antennas.py
+        entry: |
+          import sys
+          import json
 
-label: identifyBadAntennas_join
-arguments:
-  - '-c'
-  - |
-
-    import sys
-    import json
-
-    flaggedants = $(inputs.flaggedants)
-    filter = '$(inputs.filter)'
+          inputs = json.loads(r"""$(inputs)""")
+          flaggedants = inputs['flaggedants']
+          filter = inputs['filter']
 
 
-    flaggedants_list = [ flaggedant.split(',') for flaggedant in flaggedants ]
-    flagged_antenna_list = set.intersection(*map(set, flaggedants_list))
+          flaggedants_list = [ flaggedant.split(',') for flaggedant in flaggedants ]
+          if flaggedants_list:
+              flagged_antenna_list = set.intersection(*map(set, flaggedants_list))
+          else:
+              flagged_antenna_list = set()
 
-    for flagged_antenna in flagged_antenna_list:
-        if flagged_antenna != '':
-            filter += ';!' + flagged_antenna + '*&&*'
+          for flagged_antenna in flagged_antenna_list:
+              if flagged_antenna != '':
+                  filter += ';!' + flagged_antenna + '*&&*'
 
-    cwl_output  = {"filter": filter}
+          cwl_output  = {"filter": filter}
 
-    with open('./out.json', 'w') as fp:
-        json.dump(cwl_output, fp)
+          with open('./out.json', 'w') as fp:
+              json.dump(cwl_output, fp)
 
 outputs:
   - id: filter_out
@@ -50,9 +55,5 @@ outputs:
     type: File
     outputBinding:
       glob: identifyBadAntennas.log
-
-requirements:
-  - class: InlineJavascriptRequirement
-
 stdout: identifyBadAntennas.log
 stderr: identifyBadAntennas_err.log

steps/sort_times_into_freqGroups.cwl

@@ -5,6 +5,7 @@ $namespaces:
 id: sort_times_into_freqGroups
 baseCommand:
   - python3
+  - sort_times.py
 inputs:
     - id: msin
       type: Directory[]
@@ -35,42 +36,42 @@ inputs:
       type: boolean?
       default: True
       doc: Add dummy file-names for missing frequencies, so that NDPPP can fill the data with flagged dummy data.
+requirements:
+  - class: InlineJavascriptRequirement
+  - class: InitialWorkDirRequirement
+    listing:
+      - entryname: sort_times.py
+        entry: |
+          import sys
+          import json
+          from sort_times_into_freqGroups_CWL import main as sort_times_into_freqGroups
+
+          mss = sys.argv[1:]
+          inputs = json.loads(r"""$(inputs)""")
+
+          numbands = inputs['numbands']
+          stepname = inputs['stepname']
+          NDPPPfill = inputs['NDPPPfill']
+          mergeLastGroup = inputs['mergeLastGroup']
+          truncateLastSBs = inputs['truncateLastSBs']
+          firstSB = inputs['firstSB']
+
+          output = sort_times_into_freqGroups(mss, numbands, NDPPPfill, stepname, mergeLastGroup, truncateLastSBs, firstSB)
 
-label: sort_times_into_freqGroups.py
-arguments:
-  - '-c'
-  - |
-    import sys
-    import json
-    from sort_times_into_freqGroups_CWL import main as sort_times_into_freqGroups
-    null = None
-    false = False
-    true = True
-    
-    mss = sys.argv[1:]
-    numbands = int($(inputs.numbands))
-    stepname = "$(inputs.stepname)"
-    NDPPPfill = $(inputs.NDPPPfill)
-    mergeLastGroup = $(inputs.mergeLastGroup)
-    truncateLastSBs = $(inputs.truncateLastSBs)
-    firstSB = $(inputs.firstSB)
+          filenames  = output['filenames']
+          groupnames = output['groupnames']
+          total_bandwidth = output['total_bandwidth']
 
-    output = sort_times_into_freqGroups(mss, numbands, NDPPPfill, stepname, mergeLastGroup, truncateLastSBs, firstSB)
+          cwl_output = {}
+          cwl_output['groupnames'] = groupnames
+          cwl_output['total_bandwidth'] = total_bandwidth
 
-    filenames  = output['filenames']
-    groupnames = output['groupnames']
-    total_bandwidth = output['total_bandwidth']
-    
-    cwl_output = {}
-    cwl_output['groupnames'] = groupnames
-    cwl_output['total_bandwidth'] = total_bandwidth
+          with open('./filenames.json', 'w') as fp:
+              json.dump(filenames, fp)
+
+          with open('./out.json', 'w') as fp:
+              json.dump(cwl_output, fp)
 
-    with open('./filenames.json', 'w') as fp:
-        json.dump(filenames, fp)
-        
-    with open('./out.json', 'w') as fp:
-        json.dump(cwl_output, fp)
-        
 outputs:
   - id: filenames
     type: File
@@ -91,13 +92,9 @@ outputs:
   - id: logfile
     type: File
     outputBinding:
-      glob: sort_times_into_freqGroups.log    
-        
-requirements:
-  - class: InlineJavascriptRequirement
-
+      glob: sort_times_into_freqGroups.log
 hints:
   - class: DockerRequirement
     dockerPull: lofareosc/prefactor3-cwl
 stdout: sort_times_into_freqGroups.log
-stderr: sort_times_into_freqGroups_err.log
+stderr: sort_times_into_freqGroups_err.log
\ No newline at end of file