Updating documentation and library to include support for hats

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"
    ]
@@ -635,7 +649,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 +717,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 +741,19 @@
     "- `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.\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",
+    "```"
    ]
   },
   {
@@ -739,7 +764,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",
@@ -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",
+    "```"
    ]
   },
   {
@@ -1081,7 +1117,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)"
    ]

src/pzserver/process.py

@@ -6,6 +6,9 @@
 
 FONTCOLORERR = "\033[38;2;255;0;0m"
 FONTCOLOREND = "\033[0m"
+PROCESS_CONFIG_KEYS = (
+    "output_format",
+)
 
 
 class Process:
@@ -28,6 +31,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 +127,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 +185,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 +211,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 +362,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,

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",
+    }