Merge pull request #47 from LSSTDESC/issue/46/photometric_errors_pipe…

…line

Issue/46/photometric errors pipeline

src/rail/cli/pipe_commands.py

@@ -17,16 +17,16 @@ def pipe_cli() -> None:
 
 @pipe_cli.command(name="inspect")
 @pipe_options.config_file()
-def inspect(config_file: str) -> int:
+def inspect_command(config_file: str) -> int:
     """Inspect a rail pipeline project config"""
     return pipe_scripts.inspect(config_file)
 
 
-@pipe_cli.command()
+@pipe_cli.command(name="build")
 @pipe_options.config_file()
 @pipe_options.flavor()
-def build_pipelines(config_file: str, **kwargs: Any) -> int:
-    """Build the ceci pipeline configuraiton files"""
+def build_command(config_file: str, **kwargs: Any) -> int:
+    """Build the ceci pipeline configuration files"""
     project = RailProject.load_config(config_file)
     flavors = project.get_flavor_args(kwargs.pop('flavor'))
     iter_kwargs = project.generate_kwargs_iterable(flavor=flavors)
@@ -36,13 +36,13 @@ def build_pipelines(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command()
+@pipe_cli.command(name="reduce")
 @pipe_options.config_file()
 @pipe_options.input_tag()
 @pipe_options.input_selection()
 @pipe_options.selection()
 @pipe_options.run_mode()
-def reduce_roman_rubin(config_file: str, **kwargs: Any) -> int:
+def reduce_command(config_file: str, **kwargs: Any) -> int:
     """Reduce the roman rubin simulations for PZ analysis"""
     project = RailProject.load_config(config_file)
     selections = project.get_selection_args(kwargs.pop('selection'))
@@ -55,23 +55,49 @@ def reduce_roman_rubin(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="truth-to-observed")
+@pipe_cli.command(name="subsample")
 @pipe_options.config_file()
 @pipe_options.selection()
 @pipe_options.flavor()
+@pipe_options.label()
 @pipe_options.run_mode()
-def truth_to_observed_pipeline(config_file: str, **kwargs: Any) -> int:
-    """Run the truth-to-observed analysis pipeline"""
+def subsample_command(config_file: str, **kwargs: Any) -> int:
+    """Make a training or test data set by randomly selecting objects"""
+    """Make a training data set by randomly selecting objects"""
     project = RailProject.load_config(config_file)
     flavors = project.get_flavor_args(kwargs.pop('flavor'))
     selections = project.get_selection_args(kwargs.pop('selection'))
     iter_kwargs = project.generate_kwargs_iterable(flavor=flavors, selection=selections)
     ok = 0
-    pipeline_name = "truth_to_observed"
+    for kw in iter_kwargs:
+        ok |= pipe_scripts.subsample_data(project, **kw, **kwargs)
+    return ok
+
 
-    pipeline_catalog_config = pipe_scripts.TruthToObservedPipelineCatalogConfiguration(
-        project, source_catalog_tag='reduced', sink_catalog_tag='degraded',
+@pipe_cli.group(name="run")
+def run_group() -> None:
+    """Run a pipeline"""
+
+
+@run_group.command(name="phot-errors")
+@pipe_options.config_file()
+@pipe_options.selection()
+@pipe_options.flavor()
+@pipe_options.run_mode()
+def photmetric_errors_pipeline(config_file: str, **kwargs: Any) -> int:
+    """Run the photometric errors analysis pipeline"""
+    project = RailProject.load_config(config_file)
+    flavors = project.get_flavor_args(kwargs.pop('flavor'))
+    selections = project.get_selection_args(kwargs.pop('selection'))
+    iter_kwargs = project.generate_kwargs_iterable(flavor=flavors, selection=selections)
+    ok = 0
+    pipeline_name = "photometric_errors"
+    pipeline_info = project.get_pipeline(pipeline_name)
+    input_catalog_name = pipeline_info['InputCatalogTag']
+    pipeline_catalog_config = pipe_scripts.PhotmetricErrorsPipelineCatalogConfiguration(
+        project, source_catalog_tag=input_catalog_name, sink_catalog_tag='degraded',
     )
+
     for kw in iter_kwargs:
         ok |= pipe_scripts.run_pipeline_on_catalog(
             project, pipeline_name,
@@ -81,22 +107,21 @@ def truth_to_observed_pipeline(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="spec-selection")
+@run_group.command(name="truth-to-observed")
 @pipe_options.config_file()
 @pipe_options.selection()
 @pipe_options.flavor()
 @pipe_options.run_mode()
-def spectroscopic_selection_pipeline(config_file: str, **kwargs: Any) -> int:
-    """Run the spectroscopic selection analysis pipeline"""
+def truth_to_observed_pipeline(config_file: str, **kwargs: Any) -> int:
+    """Run the truth-to-observed data pipeline"""
     project = RailProject.load_config(config_file)
     flavors = project.get_flavor_args(kwargs.pop('flavor'))
     selections = project.get_selection_args(kwargs.pop('selection'))
     iter_kwargs = project.generate_kwargs_iterable(flavor=flavors, selection=selections)
     ok = 0
-    pipeline_name = "spec_selection"
+    pipeline_name = "truth_to_observed"
     pipeline_info = project.get_pipeline(pipeline_name)
     input_catalog_name = pipeline_info['InputCatalogTag']
-
     pipeline_catalog_config = pipe_scripts.SpectroscopicPipelineCatalogConfiguration(
         project,
         source_catalog_tag=input_catalog_name,
@@ -113,7 +138,7 @@ def spectroscopic_selection_pipeline(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="blending")
+@run_group.command(name="blending")
 @pipe_options.config_file()
 @pipe_options.selection()
 @pipe_options.flavor()
@@ -143,30 +168,43 @@ def blending_pipeline(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="subsample")
+@run_group.command(name="spec-selection")
 @pipe_options.config_file()
-@pipe_options.flavor()
 @pipe_options.selection()
-@pipe_options.label()
+@pipe_options.flavor()
 @pipe_options.run_mode()
-def subsample_data(config_file: str, **kwargs: Any) -> int:
-    """Make a training data set by randomly selecting objects"""
+def spectroscopic_selection_pipeline(config_file: str, **kwargs: Any) -> int:
+    """Run the spectroscopic selection data pipeline"""
     project = RailProject.load_config(config_file)
     flavors = project.get_flavor_args(kwargs.pop('flavor'))
     selections = project.get_selection_args(kwargs.pop('selection'))
     iter_kwargs = project.generate_kwargs_iterable(flavor=flavors, selection=selections)
     ok = 0
+    pipeline_name = "spec_selection"
+    pipeline_info = project.get_pipeline(pipeline_name)   
+    input_catalog_name = pipeline_info['InputCatalogTag']    
+    pipeline_catalog_config = pipe_scripts.SpectroscopicPipelineCatalogConfiguration(
+        project,
+        source_catalog_tag=input_catalog_name,
+        sink_catalog_tag='degraded',
+        source_catalog_basename="output_dereddener_errors.pq",
+    )
+
     for kw in iter_kwargs:
-        ok |= pipe_scripts.subsample_data(project, **kw, **kwargs)
+        ok |= pipe_scripts.run_pipeline_on_catalog(
+            project, pipeline_name,
+            pipeline_catalog_config,
+            **kw, **kwargs,
+        )
     return ok
 
 
-@pipe_cli.command(name="inform")
+@run_group.command(name="inform")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()
 @pipe_options.run_mode()
-def inform(config_file: str, **kwargs: Any) -> int:
+def inform_single(config_file: str, **kwargs: Any) -> int:
     """Run the inform pipeline"""
     pipeline_name = "inform"
     project = RailProject.load_config(config_file)
@@ -183,7 +221,7 @@ def inform(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="estimate")
+@run_group.command(name="estimate")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()
@@ -205,7 +243,7 @@ def estimate_single(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="evaluate")
+@run_group.command(name="evaluate")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()
@@ -227,7 +265,7 @@ def evaluate_single(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="pz")
+@run_group.command(name="pz")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()
@@ -249,7 +287,7 @@ def pz_single(config_file: str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="tomography")
+@run_group.command(name="tomography")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()
@@ -271,7 +309,7 @@ def tomography_single(config_file : str, **kwargs: Any) -> int:
     return ok
 
 
-@pipe_cli.command(name="sompz")
+@run_group.command(name="sompz")
 @pipe_options.config_file()
 @pipe_options.flavor()
 @pipe_options.selection()

src/rail/cli/pipe_scripts.py

@@ -201,6 +201,21 @@ def get_convert_commands(self, sink_dir: str) -> list[list[str]]:
         return convert_commands
 
 
+class PhotmetricErrorsPipelineCatalogConfiguration(PipelineCatalogConfiguration):
+
+    def get_convert_commands(self, sink_dir: str) -> list[list[str]]:
+        convert_command = [
+            "tables-io",
+            "convert",
+            "--input",
+            f"{sink_dir}/output_dereddener_errors.pq",
+            "--output",
+            f"{sink_dir}/output.hdf5",
+        ]
+        convert_commands = [convert_command]
+        return convert_commands
+
+
 class SpectroscopicPipelineCatalogConfiguration(PipelineCatalogConfiguration):
 
     def get_convert_commands(self, sink_dir: str) -> list[list[str]]:

src/rail/cli/reduce_roman_rubin_data.py

@@ -88,12 +88,12 @@
 
 
 def reduce_roman_rubin_data(
-    project,
-    input_tag,
-    input_selection,
-    selection,
-    run_mode=RunMode.bash,
-):
+    project: RailProject,
+    input_tag: str,
+    input_selection: str,
+    selection: str|None,
+    run_mode: RunMode=RunMode.bash,
+) -> int:
 
     source_catalogs = []
     sink_catalogs = []
@@ -120,7 +120,6 @@ def reduce_roman_rubin_data(
                 iteration_vars[i]: iteration_args[i]
                 for i in range(len(iteration_vars))
             }
-            input_selection = iteration_kwargs.pop('input_selection', None)
             source_catalog = project.get_catalog(input_tag, selection=input_selection, **iteration_kwargs)
             sink_catalog = project.get_catalog('reduced', selection=selection, **iteration_kwargs)
             sink_dir = os.path.dirname(sink_catalog)

tests/example.yaml

@@ -1,5 +1,5 @@
 # These are used to make all the other paths
-CommonPaths:  
+CommonPaths:
     root: /sdf/data/rubin/shared/pz
     scratch_root: "{root}"
     project: eac_test
@@ -33,12 +33,20 @@ Files:
     train_file_zCOSMOS_100k:
         NumObjects: 100000
         Seed: 4321
-        PathTemplate: "{catalogs_dir}/train/{project}_{selection}_zCOSMOS_100k.hdf5"    
+        PathTemplate: "{catalogs_dir}/train/{project}_{selection}_zCOSMOS_100k.hdf5"
         SourceFileBasename: output_select_zCOSMOS.pq
 
 # These are ceci pipelines that we will be running
-Pipelines: 
+Pipelines:
     truth_to_observed:
+        PipelineClass: rail.pipelines.degradation.truth_to_observed.TruthToObservedPipeline
+        CatalogTag: roman_rubin
+        InputCatalogTag: reduced
+        kwargs:
+            error_models: ErrorModels
+            selectors: SpecSelections
+            blending: true
+    photometric_errors:
         PipelineClass: rail.pipelines.degradation.apply_phot_errors.ApplyPhotErrorsPipeline
         CatalogTag: roman_rubin
         InputCatalogTag: reduced
@@ -60,7 +68,7 @@ Pipelines:
         CatalogTag: roman_rubin
         kwargs:
             algorithms: PZAlgorithms
-        InputFileTags: 
+        InputFileTags:
             input:
                 flavor: baseline
                 tag: train
@@ -141,7 +149,6 @@ SpecSelections:
 #        Module: rail.creation.degraders.spectroscopic_selections
 
 
-
 # These describe all the algorithms that estimate PZ
 PZAlgorithms:
     trainz:
@@ -185,6 +192,7 @@ Summarizers:
         Summarize: PointEstHistMaskedSummarizer
         Module: rail.estimation.algos.point_est_hist
 
+
 # The describe the error models we use in the truth_to_observed pipeline
 ErrorModels:
     lsst:
@@ -256,4 +264,3 @@ Flavors:
             inform:
                 inform_gpz:
                     gpz_method: GL
-