From 84f482f3507db3cbac5a030057509748207be01b Mon Sep 17 00:00:00 2001 From: Luigi Lucas de Carvalho Silva Date: Tue, 14 Jul 2026 11:03:38 -0300 Subject: [PATCH 1/3] Updating notebook for new flags_translation upload possibility --- docs/notebooks/pzserver_tutorial.ipynb | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/docs/notebooks/pzserver_tutorial.ipynb b/docs/notebooks/pzserver_tutorial.ipynb index a8bf94a..b05c1d0 100644 --- a/docs/notebooks/pzserver_tutorial.ipynb +++ b/docs/notebooks/pzserver_tutorial.ipynb @@ -635,7 +635,7 @@ "\n", "The pipeline **Combine Redshift Catalogs (CRC)** combines multiple redshift catalogs into a single table and registers it as a new data product on the PZ Server. It was designed to help **aggregate multiple samples from individual surveys** into a single catalog before they are associated with LSST data through spatial cross-matching.\n", "\n", - "It **optionally deduplicates** catalog entries based on the ```z_flag_homogenized``` and ```instrument_type_homogenized``` columns, which can be automatically generated for some specific supported surveys (the ```survey_name``` column is required in this case) or manually embedded by the user in their catalogs. See the **documentation at https://docs.linea.org.br/en/sci-platforms/pz_server_crc.html** and the **currently supported automatic translations at https://github.com/linea-it/pzserver_pipelines/blob/main/combine_redshift_dedup/flags_translation.yaml**. \n", + "It **optionally deduplicates** catalog entries based on the ```z_flag_homogenized``` and ```instrument_type_homogenized``` columns, which can be automatically generated for some specific supported surveys (the ```survey_name``` column is required in this case) or manually embedded by the user in their catalogs. See the **documentation at https://docs.linea.org.br/en/sci-platforms/pz_server_crc.html** and the **currently supported automatic translations at [flags_translation.yaml](https://github.com/linea-it/pzserver_combine_redshift_dedup/blob/main/flags_translation.yaml)**. \n", "\n", "> **Note:** all the **SITCOMTN-154 reference redshift catalogs are supported**, considering that they have a flag column which is quality like, that is, with float values between 0 and 1.\n", "\n", @@ -703,8 +703,7 @@ "source": [ "View the current pipeline parameters and modify them if desired.\n", "\n", - "> **Note:** For now, the ```flags_translation.yaml``` file needs to be in a location that the ```app.orch``` application can access, in the LIneA storage system. We don't recommend changing it for now. \n", - "> **Future work:** Users will be able to upload customized versions ```flags_translation.yaml``` to change default's translation dictionary. More detailed instructions on how to do this will be available soon.\n", + "> **Note:** By default, CRC uses the ```flags_translation.yaml``` registered in the pipeline configuration. The library also supports uploading a local custom flags translation file. To customize it, download the default [flags_translation.yaml](https://github.com/linea-it/pzserver_combine_redshift_dedup/blob/main/flags_translation.yaml), edit the translation values as needed, and set ```flags_translation_file``` to the local file path, for example: ```\"flags_translation_file\": \"./flags_translation.yaml\"```. When a local path is provided, the file is uploaded together with the process submission.\n", " " ] }, @@ -728,7 +727,9 @@ "- `concatenate_and_mark_duplicates`\n", "- `concatenate_and_remove_duplicates`\n", " \n", - "The `z_flag_homogenized_value_to_cut` defines a cut in the homogenized flag column greater or equal than its value." + "The `z_flag_homogenized_value_to_cut` defines a cut in the homogenized flag column greater or equal than its value.\n", + "\n", + "The example below uses the default flags translation file. If you want to use a customized local file, add `\"flags_translation_file\": \"./flags_translation.yaml\"` to `cfg`, replacing the path with the location of your edited YAML file." ] }, { @@ -739,7 +740,6 @@ "source": [ "cfg = {\n", " \"combine_type\": \"concatenate_and_mark_duplicates\",\n", - " \"flags_translation_file\": \"/data/apps/app.orch/production/pipelines/combine_redshift_dedup/flags_translation.yaml\",\n", " \"z_flag_homogenized_value_to_cut\": 3.0\n", "}\n", "\n", @@ -1081,7 +1081,8 @@ " 'output_catalog_name': 'tsm_example'\n", " }, \n", " 'dereddening': 'sfd', \n", - " 'convert_flux_to_mag': True}\n", + " 'convert_flux_to_mag': True,\n", + " 'random_seed': 42}\n", "\n", "tsm.set_config(config)" ] From fef4820f32a80ff7d966a7bea76a3e7040bd6db2 Mon Sep 17 00:00:00 2001 From: Luigi Lucas de Carvalho Silva Date: Tue, 14 Jul 2026 11:15:33 -0300 Subject: [PATCH 2/3] Support HATS output config in process submissions --- docs/notebooks/pzserver_tutorial.ipynb | 42 ++++++++++++++++++++-- src/pzserver/process.py | 29 +++++++++++++-- tests/pzserver/test_process.py | 49 ++++++++++++++++++++++++++ 3 files changed, 114 insertions(+), 6 deletions(-) create mode 100644 tests/pzserver/test_process.py diff --git a/docs/notebooks/pzserver_tutorial.ipynb b/docs/notebooks/pzserver_tutorial.ipynb index b05c1d0..3f70816 100644 --- a/docs/notebooks/pzserver_tutorial.ipynb +++ b/docs/notebooks/pzserver_tutorial.ipynb @@ -370,6 +370,8 @@ "source": [ "The default method to upload a data product to the PZ Server is using the upload form on the website. Alternatively, the `pzserver` Python library can be used to send data products to the host service.\n", "\n", + "PZ Server also supports products stored as [HATS](https://hats.readthedocs.io/en/latest/) catalogs. HATS is a hierarchical sky-partitioned catalog format authored by [LINCC Frameworks](https://lsstdiscoveryalliance.org/programs/lincc-frameworks/) and commonly used with LSDB for scalable catalog operations. To upload a HATS product, first compress the HATS collection directory as a `.zip` or `.tar` archive and use that archive as the `main_file`.\n", + "\n", "The first step is to prepare a dictionary with the relevant information about your data product. For example:" ] }, @@ -393,6 +395,18 @@ "cell_type": "markdown", "metadata": {}, "source": [ + "For a HATS product, the minimum change is to point `main_file` to the compressed HATS archive. The remaining metadata follows the same upload pattern:\n", + "\n", + "```python\n", + "data_to_upload = {\n", + " \"name\": \"example HATS upload via lib\",\n", + " \"product_type\": \"redshift_catalog\",\n", + " \"release\": None,\n", + " \"main_file\": \"./examples/my_hats_catalog.zip\", # or .tar\n", + " \"auxiliary_files\": []\n", + "}\n", + "```\n", + "\n", "Then, execute the `upload` method from the `pz_server` class, providing the product details dictionary as a `kwargs` argument." ] }, @@ -613,7 +627,7 @@ "source": [ "# PZ Server Pipelines \n", "\n", - "In addition to PZ-related data hosting and curation services, PZ Server also provides tools to help users prepare training data for PZ algorithms. The pipeline *Training Set Maker* uses the data partitioning method [HATS](https://hats.readthedocs.io/en/stable/) and the Python framework [LSDB](https://docs.lsdb.io/en/stable/) (both developed by [LINCC](https://lsstdiscoveryalliance.org/programs/lincc/)) as the cross-matching back-end engine, coupled with a user interface on the PZ Server website connected to IDAC-Brazil's high-performance computing infrastructure. \n", + "In addition to PZ-related data hosting and curation services, PZ Server also provides tools to help users prepare training data for PZ algorithms. The pipeline *Training Set Maker* uses [HATS](https://hats.readthedocs.io/en/latest/) catalogs and the Python framework [LSDB](https://docs.lsdb.io/en/stable/) as the cross-matching back-end engine, coupled with a user interface on the PZ Server website connected to IDAC-Brazil's high-performance computing infrastructure. HATS is authored by [LINCC Frameworks](https://lsstdiscoveryalliance.org/programs/lincc-frameworks/), which also develops LSDB.\n", "\n", "With *Training Set Maker*, users can create training sets by matching objects from a reference redshift catalog available on the server with objects from an LSST Object catalog. The reference catalog can be previously uploaded by the user or created as a combination of multiple redshift catalogs by another pipeline, called *Combine Redshift Catalogs*.\n" ] @@ -729,7 +743,17 @@ " \n", "The `z_flag_homogenized_value_to_cut` defines a cut in the homogenized flag column greater or equal than its value.\n", "\n", - "The example below uses the default flags translation file. If you want to use a customized local file, add `\"flags_translation_file\": \"./flags_translation.yaml\"` to `cfg`, replacing the path with the location of your edited YAML file." + "The example below uses the default flags translation file. If you want to use a customized local file, add `\"flags_translation_file\": \"./flags_translation.yaml\"` to `cfg`, replacing the path with the location of your edited YAML file.\n", + "\n", + "CRC can also write the output product as a HATS catalog. The minimum change is to set the process output format to `\"hats\"`:\n", + "\n", + "```python\n", + "cfg = {\n", + " \"combine_type\": \"concatenate_and_mark_duplicates\",\n", + " \"z_flag_homogenized_value_to_cut\": 3.0,\n", + " \"output_format\": \"hats\"\n", + "}\n", + "```" ] }, { @@ -1065,8 +1089,20 @@ " * **output_catalog_name:** name assigned to the resulting training set.\n", "* **dereddening** – dust map to use for flux dereddening (e.g., sfd).\n", "* **convert_flux_to_mag** – if True, converts fluxes to magnitudes before output.\n", + "* **output_format** – optional output product format. Use `\"hats\"` to register the resulting training set as a HATS catalog.\n", + "\n", + "These settings are passed directly to ```tsm.set_config(config)``` before submitting the process. The example below writes the default tabular output; to request HATS output, add `\"output_format\": \"hats\"` to `config`:\n", "\n", - "These settings are passed directly to ```tsm.set_config(config)``` before submitting the process." + "```python\n", + "config = {\n", + " \"flux_type\": \"cmodel\",\n", + " \"crossmatch\": {\"n_neighbors\": 1, \"radius_arcsec\": 1.0},\n", + " \"dereddening\": \"sfd\",\n", + " \"convert_flux_to_mag\": True,\n", + " \"random_seed\": 42,\n", + " \"output_format\": \"hats\"\n", + "}\n", + "```" ] }, { diff --git a/src/pzserver/process.py b/src/pzserver/process.py index 6f38dae..36302fb 100644 --- a/src/pzserver/process.py +++ b/src/pzserver/process.py @@ -6,6 +6,12 @@ FONTCOLORERR = "\033[38;2;255;0;0m" FONTCOLOREND = "\033[0m" +PROCESS_CONFIG_KEYS = ( + "output_dir", + "output_name", + "output_format", + "output_root_dir", +) class Process: @@ -28,6 +34,7 @@ def __init__(self, pipeline, name, api): self.comment = None self.pipeline = Pipeline(pipeline, self.__api) self.__config = self.pipeline.parameters + self.__process_config = {} self.__inputs = [] self.__process = None self.__upload = None @@ -123,13 +130,27 @@ def config(self): """ return self.__config + @property + def process_config(self): + """Gets process-level config overrides.""" + return self.__process_config + + @property + def used_config(self): + """Gets the complete config payload for process submission.""" + return {"param": self.config, **self.process_config} + def set_config(self, config): """Set config Args: config (dict): config """ - self.__config.update(config) + for key, value in config.items(): + if key in PROCESS_CONFIG_KEYS: + self.__process_config[key] = value + else: + self.__config[key] = value def get_product(self, product_id=None, internal_name=None): """_summary_ @@ -167,6 +188,8 @@ def summary(self, extra_info=None): print("-" * 30) print(f"{self.pipeline.display_name}: {self.name}") print(f"Configuration: {self.config}") + if self.process_config: + print(f"Process options: {self.process_config}") if isinstance(extra_info, list): for line in extra_info: @@ -191,7 +214,7 @@ def run(self): data_process = { "display_name": self.name, - "used_config": {"param": self.config}, + "used_config": self.used_config, "pipeline": self.pipeline.pipeline_id, "inputs": self.inputs, } @@ -342,7 +365,7 @@ def run(self): data_process = { "pipeline": self.pipeline.pipeline_id, - "used_config": {"param": self.config}, + "used_config": self.used_config, "display_name": self.name, "release": self.release.get("id"), "inputs": self.inputs, diff --git a/tests/pzserver/test_process.py b/tests/pzserver/test_process.py new file mode 100644 index 0000000..d020637 --- /dev/null +++ b/tests/pzserver/test_process.py @@ -0,0 +1,49 @@ +from pzserver.process import Process + + +class FakePipeline: + def __init__(self, name, api): + self.pipeline_id = 7 + self.display_name = name + self.parameters = {"combine_type": "concatenate"} + self.acceptable_product_types = () + self.output_product_type = {} + + +class FakeApi: + def __init__(self): + self.started_process = None + + def start_process(self, data, files=None): + self.started_process = data + return {"id": 11, "upload": 22} + + def get(self, entity, _id): + if entity == "products": + return { + "id": _id, + "display_name": "output", + "internal_name": "22_output", + } + if entity == "processes": + return {"status": "Pending"} + return {} + + +def test_process_level_config_is_sent_outside_param(monkeypatch): + monkeypatch.setattr("pzserver.process.Pipeline", FakePipeline) + api = FakeApi() + process = Process("combine_redshift_dedup", "crc hats", api) + + process.set_config( + { + "combine_type": "concatenate_and_mark_duplicates", + "output_format": "hats", + } + ) + process.run() + + assert api.started_process["used_config"] == { + "param": {"combine_type": "concatenate_and_mark_duplicates"}, + "output_format": "hats", + } From f9962ce190b87edcb25460fbffc3edc593f6e9c4 Mon Sep 17 00:00:00 2001 From: Luigi Lucas de Carvalho Silva Date: Tue, 14 Jul 2026 11:39:42 -0300 Subject: [PATCH 3/3] Ommiting other outputs options --- src/pzserver/process.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/pzserver/process.py b/src/pzserver/process.py index 36302fb..dbd47c2 100644 --- a/src/pzserver/process.py +++ b/src/pzserver/process.py @@ -7,10 +7,7 @@ FONTCOLORERR = "\033[38;2;255;0;0m" FONTCOLOREND = "\033[0m" PROCESS_CONFIG_KEYS = ( - "output_dir", - "output_name", "output_format", - "output_root_dir", )