### Clone and Install hats-import from Source

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/contributing.md

Clone the repository and install the package with its development dependencies.

```bash
git clone https://github.com/astronomy-commons/hats-import
cd hats-import/
```

```bash
chmod +x .setup_dev.sh
./.setup_dev.sh
```

--------------------------------

### Setup: Create Toy Dataset

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Generates a toy dataset split across multiple CSV files and a temporary directory to simulate input for the HATS import pipeline. This setup is used to demonstrate the reducing stage.

```python
import tempfile
from pathlib import Path

import numpy as np
import pandas as pd

# Create toy dataset - split into 3 files to demonstrate reducing stage
# When we process multiple input files, objects going to the same destination
# pixel will create multiple shards that need to be combined during reducing.
toy_data_1 = pd.DataFrame({"id": [0, 1, 2], "ra": [10, 10, 20], "dec": [10, 60, 80]})

toy_data_2 = pd.DataFrame({"id": [3, 4], "ra": [-10, 100], "dec": [80, -80]})

toy_data_3 = pd.DataFrame({"id": [5, 6, 7], "ra": [200, 200, 15], "dec": [0, 60, 70]})

# Create a temporary directory for our demo
demo_dir = tempfile.TemporaryDirectory(prefix="hats_demo_")
demo_path = Path(demo_dir.name)

# Write to separate CSV files
input_csv_1 = demo_path / "toy_catalog_1.csv"
input_csv_2 = demo_path / "toy_catalog_2.csv"
input_csv_3 = demo_path / "toy_catalog_3.csv"

toy_data_1.to_csv(input_csv_1, index=False)
toy_data_2.to_csv(input_csv_2, index=False)
toy_data_3.to_csv(input_csv_3, index=False)

# Combine for display
toy_data = pd.concat([toy_data_1, toy_data_2, toy_data_3], ignore_index=True)
input_files = [input_csv_1, input_csv_2, input_csv_3]

print("Toy Dataset (split across 3 files):")
print(toy_data)
print(f"\nFile 1: {input_csv_1.name} ({len(toy_data_1)} rows)")
print(f"File 2: {input_csv_2.name} ({len(toy_data_2)} rows)")
print(f"File 3: {input_csv_3.name} ({len(toy_data_3)} rows)")
```

--------------------------------

### Collection Properties File Example

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/collections.md

Shows an example of a collection.properties file, defining the primary catalog, margins, and indexes for the collection.

```default
# HATS Collection
obs_collection = gaia_dr3
hats_primary_table_url = catalog
all_margins = gaia_dr3_5arcs gaia_dr3_100arcs
default_margin = gaia_dr3_5arcs
all_indexes = designation gaia_dr3_designation
```

--------------------------------

### Install Ray

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/dask_on_ray.md

Install the Ray library using pip. This is a prerequisite for using Dask-on-Ray.

```bash
pip install ray
```

--------------------------------

### Progress Reporting Example

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/index_table.md

This is an example of how progress bars are displayed during pipeline execution for mapping, reducing, and finishing stages.

```text
Mapping  : 100%|██████████| 2352/2352 [9:25:00<00:00, 14.41s/it]
Reducing : 100%|██████████| 2385/2385 [00:43<00:00, 54.47it/s]
Finishing: 100%|██████████| 4/4 [00:03<00:00,  1.15it/s]
```

--------------------------------

### Collection Pipeline Setup with Builder Pattern

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/collections.md

Demonstrates how to set up a collection pipeline using the CollectionArguments builder pattern in Python. This includes specifying output details, catalog arguments, and margin/index arguments.

```python
from hats_import import CollectionArguments

args = (
    CollectionArguments(
        output_artifact_name="my-catalog",
        output_path="/path/to/place/collection/",
        ## dask arguments go here
        ## progress reporting arguments go here
    )
    .catalog(
        input_path="/path/to/csv/files",
        file_reader="csv",
        ra_column="ra",
        dec_column="dec",
        sort_columns="source_id",
    )
    .add_margin(margin_threshold=5.0, is_default=True)
    .add_margin(margin_threshold=100.0)
    .add_index(indexing_column="designation", include_healpix_29=False)
)
```

--------------------------------

### Minimal ImportArguments Setup

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Instantiates ImportArguments with essential parameters for catalog import, including input/output paths and column names. Use this for basic catalog imports.

```python
from hats_import.catalog.arguments import ImportArguments

args = ImportArguments(
    sort_columns="ObjectID",
    ra_column="ObjectRA",
    dec_column="ObjectDec",
    input_path="./my_data",
    file_reader="csv",
    output_artifact_name="test_cat",
    output_path="./output",
)
```

--------------------------------

### Initialize Ray and Enable Dask-on-Ray

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/dask_on_ray.md

Initialize a Ray client within a context manager, enable Dask-on-Ray, and then create a Dask client. This setup allows Dask tasks to run on Ray workers. Ensure to disable Dask-on-Ray when done.

```python
import ray
from dask.distributed import Client
from ray.util.dask import disable_dask_on_ray, enable_dask_on_ray

from hats_import.pipeline import pipeline_with_client

with ray.init(
    num_cpus=args.dask_n_workers,
    _temp_dir=args.dask_tmp,
):
    enable_dask_on_ray()

    with Client(
        local_directory=args.dask_tmp,
        n_workers=args.dask_n_workers
    ) as client:
        pipeline_with_client(args, client)

    disable_dask_on_ray()
```

--------------------------------

### Collection Directory Structure Example

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/collections.md

Illustrates a typical directory structure for a catalog collection, including the primary catalog, margin tables, and index tables.

```default
gaia_dr3 /
|-- collection.properties
|-- gaia_dr3 /
| | -- hats.properties
| + -- . . .
|-- gaia_dr3_5arcs /
| | -- hats.properties
| + -- . . .
|-- gaia_dr3_100arcs /
| | -- hats.properties
| + -- . . .
+-- gaia_dr3_designation /
    |-- hats.properties
    + -- . . .
```

--------------------------------

### Dask Client Setup for Pipeline

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/hipscat_conversion.md

When you need more control over Dask, create your own client and pass it to the pipeline. This is useful for custom Dask configurations, such as SLURM worker pools.

```python
from dask.distributed import Client
from hats_import.pipeline import pipeline_with_client

args = ConversionArguments(...)
with Client('scheduler:port') as client:
    pipeline_with_client(args, client)
```

--------------------------------

### Run Package Unit Tests

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/contributing.md

Execute the project's unit tests using pytest to verify installation and code integrity.

```bash
python -m pytest
```

--------------------------------

### Create and Activate Conda Environment

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/getting_started.md

Use conda to create a new virtual environment for installing hats-import and its dependencies. Ensure you activate the environment before proceeding.

```console
>> conda create -n <env_name> python=3.12
>> conda activate <env_name>
```

--------------------------------

### Custom Starr File Reader Implementation

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/reference/file_readers.md

Example of a custom InputReader subclass for a fictional Starr file format. It demonstrates handling 'read_columns' for optimized RA/Dec reads and iterating through the full file using chunking.

```python
class StarrReader(InputReader):
    """Class for fictional Starr file format."""
    def __init__(self, chunksize=500_000, **kwargs):
        self.chunksize = chunksize
        self.kwargs = kwargs

    def read(self, input_file, read_columns=None):
        if read_columns:
           ## shortcut - just need positions
           starr_manifest = starr_io.read_manifest(input_file, **self.kwargs)
           ra_array, dec_array = starr_manifest.read_positions()
           yield pd.Dataframe({"ra" : ra_array, "dec" : dec_array})
        else:
           ## get all of the fields
           starr_file = starr_io.read_table(input_file, **self.kwargs)
           for smaller_table in starr_file.to_batches(max_chunksize=self.chunksize):
              smaller_table = join_to_starr_data(smaller_table)
              yield smaller_table.to_pandas()
```

--------------------------------

### Perform Import with Schema File

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/unequal_schema.ipynb

This code demonstrates how to successfully import data from CSV files with unequal schemas by providing an explicit Parquet schema file. This resolves the casting errors encountered in the previous example.

```python
tmp_path = tempfile.TemporaryDirectory()
args = ImportArguments(
    output_artifact_name="mixed_csv_good",
    input_file_list=[
        os.path.join(mixed_schema_csv_dir, "input_01.csv"),
        os.path.join(mixed_schema_csv_dir, "input_02.csv"),
    ],
    output_path=tmp_path.name,
    highest_healpix_order=1,
    file_reader=get_file_reader("csv", schema_file=mixed_schema_csv_parquet),
    use_schema_file=mixed_schema_csv_parquet,
)
with Client(n_workers=1, threads_per_worker=1) as client:
    pipeline_with_client(args, client)
```

--------------------------------

### Create Temporary Directory and Sample Parquet File Path

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/estimate_pixel_threshold.ipynb

Sets up a temporary directory and defines the path for the sample parquet file. Remember to change the `sample_parquet_file` path if you are not using a temporary directory.

```python
import os
import tempfile

tmp_path = tempfile.TemporaryDirectory()
sample_parquet_file = os.path.join(tmp_path.name, "sample.parquet")
```

--------------------------------

### Initialize Output Path with Universal Pathlib

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/margin_cache.md

When writing to cloud storage or using filesystem credentials, initialize `output_path` using `universal_pathlib` utilities.

```python
output_path = universal_pathlib.Path("s3://my-bucket/my-catalogs")
```

--------------------------------

### Build Sphinx Documentation Locally

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/contributing.md

Build the project's documentation locally using Sphinx. Navigate to the docs directory first.

```bash
cd docs
make html
```

--------------------------------

### Define SkyMapper Collection Paths

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/pre_executed/backport_collection.ipynb

Set variables for a specific catalog collection, such as SkyMapper DR4. This example shows a different path and catalog subdirectory.

```python
collection_path = "/data3/epyc/data3/hats/catalogs/skymapper/sky_mapper_dr4"
catalog_subdir = "catalog"

## This is a human-readable name of the collection, often the survey or data release.
collection_name = "sky_mapper_dr4"
```

--------------------------------

### Initialize Verification Arguments

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/verification.md

Instantiate VerificationArguments with input and output paths. Ensure input_catalog_path is correctly set for cloud storage using universal_pathlib if necessary.

```python
from hats_import.verification.arguments import VerificationArguments

args = VerificationArguments(
    input_catalog_path="./my_data/my_catalog",
    output_path="./output",
)
```

--------------------------------

### Prepare Output Directory for Bad Schemas

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Initializes the output directory for malformed schemas. It removes any existing directory to ensure a clean state and raises an error if the parent directory is not empty.

```python
output_dataset_path = Path(".") / "bad_schemas" / "dataset"
remove_directory(output_dataset_path)

# Existing files may result in unexpected metadata output.
if output_dataset_path.parent.exists() and any(output_dataset_path.parent.iterdir()):
    raise FileExistsError("bad_schemas directory exists and is not empty. Remove it and try again.")

# We will create the following files using input_frag
ffrag_out = output_dataset_path / frag_key
fextra_col = ffrag_out.with_suffix(".extra_column.parquet")
fmissing_col = ffrag_out.with_suffix(".missing_column.parquet")
fwrong_types = ffrag_out.with_suffix(".wrong_dtypes.parquet")
fwrong_metadata = ffrag_out.with_suffix(".wrong_metadata.parquet")

ffrag_out.parent.mkdir(parents=True, exist_ok=True)
```

--------------------------------

### Creating a Thumbnail File

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Set `create_thumbnail` to `True` to optionally generate a `dataset/data_thumbnail.parquet` file containing the first row of each data partition as a representative sample.

```python
create_thumbnail=True
```

--------------------------------

### Prepare Dataset for Validation Failures

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Sets up a directory structure and copies initial files to simulate a dataset that will fail validation. It prepares files for recording in metadata, including one that will be missing and one that will have extra rows.

```python
output_dataset_path = Path(".") / "wrong_files_and_rows" / "dataset"
remove_directory(output_dataset_path.parent)

# Existing files may result in unexpected metadata output.
if output_dataset_path.parent.exists() and any(output_dataset_path.parent.iterdir()):
    raise FileExistsError("wrong_files_and_rows directory exists and is not empty. Remove it and try again.")

# We will create the following files using input_frag
ffrag_out = output_dataset_path / frag_key
fmissing_file = ffrag_out.with_suffix(".missing_file.parquet")
fextra_file = ffrag_out.with_suffix(".extra_file.parquet")
fextra_rows = ffrag_out.with_suffix(".extra_rows.parquet")

ffrag_out.parent.mkdir(parents=True, exist_ok=True)

# Copy metadata files that we will not alter
shutil.copy(input_dataset_path.parent / "properties", output_dataset_path.parent / "properties")
shutil.copy(input_dataset_path / "_common_metadata", output_dataset_path / "_common_metadata")
```

```python
# Make a direct copy of input_frag for all files that will be recorded in the _metadata file
for file_out in [ffrag_out, fmissing_file, fextra_rows]:
    shutil.copy(input_frag.path, file_out)

# Write _metadata
collect_and_write_metadata(output_dataset_path)
```

```python
# Mangle the dataset.

# Add a file
shutil.copy(input_frag.path, fextra_file)

# Remove a file
fmissing_file.unlink()

# Add rows to an existing file
new_tbl = pa.concat_tables([input_tbl, input_tbl.take([1, 2, 3, 4])])
pq.write_table(new_tbl, fextra_rows)
```

--------------------------------

### Set Up HATS Import Pipeline with Dask Client

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/getting_started.md

Instantiate argument containers and pass them to the pipeline control. Running within a main guard and using a dask Client can help manage threading issues.

```python
from dask.distributed import Client
from hats_import.pipeline import pipeline_with_client

def main():
    args = ...
    with Client(
        n_workers=10,
        threads_per_worker=1,
        ...
    ) as client:
        pipeline_with_client(args, client)

if __name__ == '__main__':
    main()
```

--------------------------------

### Instantiate Custom Starr Reader for Import Pipeline

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/reference/file_readers.md

Shows how to instantiate the custom StarrReader and pass it to the ImportArguments for the catalog import pipeline. Ensure the input_path correctly locates your files.

```python
args = ImportArguments(
    ...
    ## Locates files like "/directory/to/files/**starr"
    input_path="/directory/to/files/",
    ## NB - you need the parens here!
    file_reader=StarrReader(),
)
```

--------------------------------

### Create Collection Properties

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/pre_executed/backport_collection.ipynb

Instantiate `CollectionProperties` with catalog metadata and write it to a properties file in the specified collection path.

```python
from hats.catalog.dataset.collection_properties import CollectionProperties

info = {"obs_collection": collection_name}
info["hats_primary_table_url"] = catalog_subdir
if margin_paths:
    info["all_margins"] = margin_paths
if default_margin:
    info["default_margin"] = default_margin

if index_paths:
    info["all_indexes"] = index_paths

properties = CollectionProperties(**info)
properties.to_properties_file(collection_path)
```

--------------------------------

### Display Initial Directory Structure

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Calls the utility function to show the directory structure before any pipeline stages are applied. This serves as a baseline for observing changes.

```python
# Show initial directory structure
show_directory_tree(demo_path, "SETUP")
```

--------------------------------

### Import Libraries and Initialize Client

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Imports necessary libraries and initializes a Dask client for distributed computing. Sets up temporary directories and paths for data processing.

```python
import shutil
import tempfile
from pathlib import Path

import lsdb
import pyarrow as pa
import pyarrow.dataset as pds
import pyarrow.parquet as pq
from dask.distributed import Client
from hats.io.file_io import remove_directory

from hats_import import pipeline_with_client, ImportArguments

tmp_path = tempfile.TemporaryDirectory()
tmp_dir = tmp_path.name

hats_import_dir = "."
client = Client(n_workers=1, threads_per_worker=1, local_directory=tmp_dir)
```

--------------------------------

### Initialize IndexArguments

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/index_table.md

Instantiate IndexArguments with essential parameters for the indexing pipeline. Specify input catalog path, the column to index, and output details.

```python
from hats_import.index.arguments import IndexArguments

args = IndexArguments(
    input_catalog_path="./my_data/my_catalog",
    indexing_column="target_id",
    output_path="./output",
    output_artifact_name="my_catalog_target_id",
)
```

--------------------------------

### Copy Input Fragment and Write Initial Metadata

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Copies the input fragment to create base files for malformed data and writes the initial metadata file. This sets up the structure before introducing specific errors.

```python
# Make a direct copy of input_frag for all files that will be recorded in the _metadata file
for file_out in [ffrag_out, fmissing_col, fextra_col, fwrong_types]:
    shutil.copy(input_frag.path, file_out)

# Write a _metadata that has the correct schema except for file-level metadata
metadata = input_tbl.schema.metadata or {}
metadata.update({b"extra key": b"extra value"})
collect_and_write_metadata(output_dataset_path, schema=input_tbl.schema.with_metadata(metadata))
```

--------------------------------

### Print HATS Catalog File Explanations

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Prints explanations for various files within the HATS catalog, including partition info, properties, pixel parquet files, metadata, and the point map.

```python
print("=== CATALOG FILES EXPLAINED ===")
print()

print("1. partition_info.csv")
print("   - Lists all HEALPix pixels in the catalog")
print("   - Shows the directory structure (Norder=/Npix=)")
print("   Content:")
partition_df = pd.read_csv(partition_info_file)
print(partition_df.to_string(index=False))
print()

print("2. properties")
print("   - Catalog metadata (name, type, row count, schema)")
print("   Content (excerpt):")
with open(properties_file, "r") as f:
    for i, line in enumerate(f):
        if i < 10:
            print(f"   {line.rstrip()}")
print("   ...")
print()

print("3. Pixel parquet files (Norder=/Npix=/Npix=.parquet)")
print("   - Actual catalog data, one file per HEALPix pixel")
print("   - Organized in directories by order and pixel")
print("   - Sorted by _healpix_29 spatial index")
print()
for hp_pixel, file_path, count in reduced_files:
    print(f"   {file_path.relative_to(output_path)} ({count} rows):")
    df = pd.read_parquet(file_path)
    print(df.to_string(index=False))
    print()

print("4. _metadata & _common_metadata")
print("   - Parquet metadata for the entire catalog")
print("   - _common_metadata: schema shared by all files")
print("   - _metadata: references to all parquet files")
print()

print("5. point_map.fits")
print("   - FITS skymap showing object distribution")
print(f"   - Resolution: HEALPix order {mapping_healpix_order}")
print(f"   - Non-empty pixels: {len(non_empty_pixels)}")
```

--------------------------------

### Import AllWISE Catalog Data

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/public/allwise.md

This script demonstrates how to set up and run an import pipeline for the AllWISE catalog using `pandas` and `hats_import`. It configures a `CsvReader` with specific chunk sizes, separators, and column mappings, and utilizes a schema file for type handling. Ensure `allwise_types.csv` and `allwise_schema.parquet` are accessible.

```python
import pandas as pd

import hats_import.pipeline as runner
from hats_import.catalog.arguments import ImportArguments
from hats_import.catalog.file_readers import CsvReader

# Load the column names and types from a side file.
type_frame = pd.read_csv("allwise_types.csv")
type_map = dict(zip(type_frame["name"], type_frame["type"]))

args = ImportArguments(
    output_artifact_name="allwise",
    input_path="/path/to/allwise/",
    file_reader=CsvReader(
        chunksize=250_000,
        header=None,
        sep="|",
        column_names=type_frame["name"].values.tolist(),
        type_map=type_map,
    ),
    use_schema_file="allwise_schema.parquet",
    ra_column="ra",
    dec_column="dec",
    sort_columns="source_id",
    pixel_threshold=1_000_000,
    highest_healpix_order=7,
    output_path="/path/to/catalogs/",
)
runner.pipeline(args)
```

--------------------------------

### Initialize MarginCacheArguments

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/margin_cache.md

Instantiate MarginCacheArguments with essential paths and parameters for generating a margin cache. Ensure input_catalog_path and output_path are correctly specified.

```python
from hats_import.margin_cache.margin_cache_arguments import MarginCacheArguments

args = MarginCacheArguments(
    input_catalog_path="./my_data/my_catalog",
    output_path="./output",
    margin_threshold=10.0,
    output_artifact_name="my_catalog_10arcs",
)
```

--------------------------------

### Inspect Parquet Schema

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/unequal_schema.ipynb

This snippet shows how to load a Parquet file and print its schema using `pyarrow.parquet`. This is useful for understanding the expected data types before an import.

```python
import pyarrow.parquet as pq

mixed_schema_csv_parquet = "../../tests/data/mixed_schema/schema.parquet"

parquet_file = pq.ParquetFile(mixed_schema_csv_parquet)
print(parquet_file.schema)
```

--------------------------------

### Display Directory Tree for Finishing Stage

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This snippet visualizes the directory structure created during the finishing stage of the HATS catalog import. It requires the `show_directory_tree` function to be available.

```python
show_directory_tree(demo_path, "FINISHING")
```

--------------------------------

### Run Pre-commit Checks

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/contributing.md

Execute the pre-commit hooks to check staged code for quality and formatting issues.

```bash
pre-commit
```

--------------------------------

### Create String Divisions for Gaia DR3

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/index_table.md

Generate sample division hints for string identifiers, specifically for Gaia DR3 data. This uses string prefixes to help Dask optimize the indexing process.

```python
divisions = [f"Gaia DR3 {i}" for i in range(10_000, 99_999, 12)]
divisions.append("Gaia DR3 999999988604363776")
```

--------------------------------

### Format and Display Schema Sizes

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/estimate_pixel_threshold.ipynb

Formats schema sizes as percentages and displays them in a sorted DataFrame. Requires pandas and assumes 'sizes' and 'parquet_file' are defined.

```python
percents = [f"{s/sizes.sum()*100:.1f}" for s in sizes]
pd.DataFrame({"name": parquet_file.schema.names, "size": sizes.astype(int), "percent": percents}).sort_values(
    "size", ascending=False
)
```

--------------------------------

### Specify Output Path

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Define the base directory for your catalogs using `output_path`. The full path will be `output_path/output_artifact_name`.

```python
output_path='/path/to/catalogs'
```

--------------------------------

### Load and Verify Catalog Collection

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/pre_executed/backport_collection.ipynb

Load the newly created catalog collection using `lsdb.read_hats` and assert that it is recognized as a collection.

```python
import lsdb

new_collection = lsdb.read_hats(collection_path)
assert new_collection.hc_collection
```

--------------------------------

### Display Directory Tree for Splitting Stage

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This command visualizes the directory structure created during the splitting stage of the pipeline. It helps in verifying the output and understanding the organization of the intermediate shard files.

```python
show_directory_tree(demo_path, "SPLITTING")
```

--------------------------------

### Read and Join Catalogs, then Write Nested Catalog

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Reads object and source catalogs using `lsdb.read_hats`, performs a nested join based on 'id' and 'object_id', and then writes the resulting nested catalog to disk using `lsdb.io.to_hats`.

```python
small_sky_object = lsdb.read_hats("small_sky_object_catalog", columns="all")
small_sky_source = lsdb.read_hats("small_sky_source_catalog")
small_sky_nested = small_sky_object.join_nested(
    small_sky_source, left_on="id", right_on="object_id", nested_column_name="lc"
)
lsdb.io.to_hats(
    small_sky_nested,
    base_catalog_path="small_sky_nested_catalog",
    catalog_name="small_sky_nested_catalog",
    histogram_order=5,
    overwrite=True,
)
```

--------------------------------

### Write Partition Info, Parquet Metadata, Point Map, and Properties, then Validate Catalog

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This script performs the final steps of the HATS catalog import. It writes partition information, Parquet metadata, a FITS skymap, and catalog properties. Finally, it validates the entire catalog. Ensure necessary imports and variables like `destination_pixel_map`, `output_path`, `raw_histogram`, `ra_column`, and `dec_column` are defined.

```python
from hats.catalog import PartitionInfo, TableProperties
from hats.io.parquet_metadata import write_parquet_metadata
from hats.io import file_io as io
from hats.io.validation import is_valid_catalog

print(f"=== FINISHING STAGE ===")
print(f"Writing catalog metadata and validating...\n")

# 1. Write partition_info.csv
print("1. Writing partition_info.csv...")
partition_info = PartitionInfo.from_healpix(list(destination_pixel_map.keys()))
partition_info_file = paths.get_partition_info_pointer(output_path)
partition_info.write_to_file(partition_info_file)
print(f"   → {partition_info_file.relative_to(demo_path)}")
print(f"   Content:")
print(partition_info.as_dataframe().to_string(index=False))
print()

# 2. Write Parquet metadata
print("2. Writing Parquet metadata files...")
parquet_rows = write_parquet_metadata(output_path)
print(f"   → _metadata")
print(f"   → _common_metadata")
print(f"   Validated {parquet_rows} total rows")

# Read schema
nested_schema = npd.read_parquet(paths.get_common_metadata_pointer(output_path))
column_names = list(nested_schema.columns) + nested_schema.get_subcolumns()
print(f"   Columns: {column_names}")
print()

# 3. Write point_map.fits
print("3. Writing point_map.fits...")
io.write_fits_image(raw_histogram, paths.get_point_map_file_pointer(output_path))
print(f"   → {paths.get_point_map_file_pointer(output_path).relative_to(demo_path)}")
print(f"   (FITS skymap of {len(non_empty_pixels)} non-empty pixels at order {mapping_healpix_order})")
print()

# 4. Write catalog properties
print("4. Writing catalog properties...")
total_rows = sum(destination_pixel_map.values())
catalog_info = TableProperties(
    catalog_name="toy_catalog",
    catalog_type="object",
    total_rows=total_rows,
    ra_column=ra_column,
    dec_column=dec_column,
)
catalog_info.to_properties_file(output_path)
# The properties file is simply 'properties' in the catalog base directory
properties_file = output_path / "properties"
print(f"   → {properties_file.relative_to(demo_path)}")
print(f"   Total rows: {total_rows}")
print(f"   Highest order: {partition_info.get_highest_order()}")
print()

# 5. Validate catalog
print("5. Validating catalog...")
is_valid = is_valid_catalog(output_path)
print(f"   ✓ Catalog is valid: {is_valid}")
print()

print("=== IMPORT COMPLETE ===")
print(f"\nCatalog location: {output_path}")
```

--------------------------------

### Specify Output Path

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/margin_cache.md

Set the `output_path` to the base directory where margin data should be written. The full path will be `output_path/output_artifact_name`.

```python
output_path="/path/to/your/catalogs"
```

--------------------------------

### Load Input Data for Malformed Catalogs

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Loads the initial dataset and identifies a specific fragment for use in generating malformed files. Ensure the input directory and dataset structure are as expected.

```python
input_dataset_path = Path(hats_import_dir) / "small_sky_object_catalog" / "dataset"
input_ds = pds.parquet_dataset(input_dataset_path / "_metadata")

# Unit tests expect the Npix=11 data file
input_frag = next(frag for frag in input_ds.get_fragments() if frag.path.endswith("Npix=11.parquet"))
frag_key = Path(input_frag.path).relative_to(input_dataset_path)
input_tbl = input_frag.to_table()
```

--------------------------------

### Cleanup Demo Directory

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Cleans up the temporary directory created for the HATS import demo. This is a final step to remove intermediate files and free up space.

```python
# Cleanup

demo_dir.cleanup()
print(f"\nCleaned up demo directory at: {demo_dir.name}")
```

--------------------------------

### Run Pipeline with Existing Dask Client

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/index_table.md

Execute the indexing pipeline using a pre-configured Dask client. This is useful when managing Dask resources separately, such as within a SLURM worker pool.

```python
from dask.distributed import Client
from hats_import.pipeline import pipeline_with_client

args = IndexArguments(...) 
with Client('scheduler:port') as client:
    pipeline_with_client(args, client)
```

--------------------------------

### Basic HiPSCat to HATS Conversion

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/hipscat_conversion.md

Use this snippet for a minimal conversion pipeline. Ensure you provide valid paths for input HiPSCat data and the desired output directory for the HATS catalog.

```python
import hats_import.pipeline as runner
from hats_import.hipscat_conversion.arguments import ConversionArguments

args = ConversionArguments(
    input_catalog_path="./hipscat_catalogs/my_catalog",
    output_path="./hats_catalogs/",
    output_artifact_name="my_catalog",
)
runner.pipeline(args)
```

--------------------------------

### Import NEOWISE Data with Custom CSV Reader

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/public/neowise.md

Use this script to import NEOWISE data. It requires loading column names and types from a separate CSV file and specifies various import arguments, including input/output paths, file reader configuration, and schema file usage. Ensure the `neowise_types.csv` and `neowise_schema.parquet` files are accessible.

```python
import pandas as pd

import hats_import.pipeline as runner
from hats_import.catalog.arguments import ImportArguments
from hats_import.catalog.file_readers import CsvReader

# Load the column names and types from a side file.
type_frame = pd.read_csv("neowise_types.csv")
type_map = dict(zip(type_frame["name"], type_frame["type"]))

args = ImportArguments(
    output_artifact_name="neowise_1",
    input_path="/path/to/neowiser_year8/",
    file_reader=CsvReader(
        header=None,
        sep="|",
        column_names=type_frame["name"].values.tolist(),
        type_map=type_map,
        chunksize=250_000,
    ).read,
    ra_column="RA",
    dec_column="DEC",
    pixel_threshold=2_000_000,
    highest_healpix_order=9,
    use_schema_file="neowise_schema.parquet",
    sort_columns="SOURCE_ID",
    output_path="/path/to/catalogs/",
)
runner.run(args)
```

--------------------------------

### Hats-import Class Overview

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/_templates/autosummary/class.rst

This section outlines the available attributes and methods for the hats-import class. Use this information to understand and interact with the class's functionalities.

```APIDOC
## Class: hats-import

### Description
Provides functionalities related to importing data within the astronomy-commons project.

### Attributes
This class exposes the following attributes:

* attribute1 (type) - Description of attribute1
* attribute2 (type) - Description of attribute2

### Methods
This class provides the following methods:

* method1(param1, param2) - Description of method1
* method2(param1) - Description of method2
```

--------------------------------

### Display Directory Tree for Mapping Stage

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This snippet visualizes the directory structure relevant to the mapping stage of the pipeline. It's useful for understanding the file organization and context of the mapping process.

```python
show_directory_tree(demo_path, "MAPPING")
```

--------------------------------

### pipeline

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/reference/pipeline.md

Pipeline that creates its own client from the provided runtime arguments.

```APIDOC
## pipeline(args: RuntimeArguments)

### Description
Pipeline that creates its own client from the provided runtime arguments.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **args** (RuntimeArguments) - Required - The runtime arguments for the pipeline.
```

--------------------------------

### Gathering Input Files with Glob

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Use the `glob` module to gather a list of input files based on a pattern. Ensure the list is sorted for consistent processing.

```python
import glob

in_file_paths = glob.glob("/data/object_and_source/object**.csv")
in_file_paths.sort()
```

--------------------------------

### Utility Function: Show Directory Tree

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

A helper function to visualize the directory structure after each pipeline stage. It displays files, their sizes, and subdirectories with item counts.

```python
def show_directory_tree(base_path, stage_name, max_depth=4):
    """Display the directory structure after a pipeline stage.

    Args:
        base_path: Root directory to display
        stage_name: Name of the stage (for display)
        max_depth: Maximum depth to traverse
    """
    print(f"\n{'='*60}")
    print(f"Directory structure after {stage_name}:")
    print(f"{'='*60}")

    def print_tree(path, prefix="", depth=0):
        if depth > max_depth:
            return

        if not path.exists():
            return

        contents = sorted(path.iterdir(), key=lambda p: (p.is_file(), p.name))

        for i, item in enumerate(contents):
            is_last = i == len(contents) - 1
            current_prefix = "└── " if is_last else "├── "
            next_prefix = "    " if is_last else "│   "

            # Add file size for files
            if item.is_file():
                size = item.stat().st_size
                if size < 1024:
                    size_str = f"{size} B"
                elif size < 1024**2:
                    size_str = f"{size/1024:.1f} KB"
                else:
                    size_str = f"{size/1024**2:.1f} MB"
                print(f"{prefix}{current_prefix}{item.name} ({size_str})")
            else:
                # Count items in directory
                try:
                    num_items = len(list(item.iterdir()))
                    print(f"{prefix}{current_prefix}{item.name}/ ({num_items} items)")
                except PermissionError:
                    print(f"{prefix}{current_prefix}{item.name}/")
                print_tree(item, prefix + next_prefix, depth + 1)

    print(f"{base_path.name}/")
    print_tree(base_path)
    print()
```

--------------------------------

### Using a Schema File for Metadata

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Provide the path to a parquet file containing schema metadata using `use_schema_file` to ensure consistent column metadata during file writing.

```python
use_schema_file="path/to/schema.parquet"
```

--------------------------------

### Display Directory Tree

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Displays the directory tree structure for a given path, useful for visualizing the organization of files related to a specific stage, here labeled 'BINNING'.

```python
show_directory_tree(demo_path, "BINNING")
```

--------------------------------

### Configure tqdm Arguments

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Pass custom arguments to `tqdm` via the `tqdm_kwargs` parameter for further configuration of progress bars.

```python
tqdm_kwargs={'ncols': 100}
```

--------------------------------

### Adding Additional HATS Properties

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Use `addl_hats_properties` to pass key-value sets that will be included in the final `hats.properties` file for data provenance.

```python
addl_hats_properties={"hats_cols_default": "id, mjd", "obs_regime": "Optical"}
```

--------------------------------

### Load HATS Catalog with LSDB

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

Loads a HATS catalog using LSDB after the import pipeline is complete. This snippet shows how to read the catalog, compute its contents into a DataFrame, and display partition information.

```python
import lsdb

print("=== LOADING WITH LSDB ===")
print()

# Load the catalog
catalog = lsdb.read_hats(output_path)
computed_df = catalog.compute(progress_bar=False)

print(f"Catalog: {catalog}")
print("")
print(f"Partitions: {catalog.get_healpix_pixels()}")
print("")
print("Data:")
computed_df
```

--------------------------------

### Splitting Stage: Create Shard Files

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This Python script processes input CSV files to create intermediate shard parquet files. It maps objects to pixels, applies alignment to determine destination pixels, and writes filtered data into shard files organized by destination pixel. Use this script to generate the intermediate shard files required for subsequent pipeline stages.

```python
import pyarrow as pa
import pyarrow.parquet as pq
import nested_pandas as npd
from hats.io import file_io
from hats_import.pipeline_resume_plan import get_pixel_cache_directory

# Create cache directory for shards
cache_shard_path = demo_path / "cache_shards"
cache_shard_path.mkdir(exist_ok=True)

print(f"=== SPLITTING STAGE ===")
print(f"Creating intermediate shard files...\n")

shard_files_created = []

# Process each input file
for file_idx, input_file in enumerate(input_files):
    print(f"Processing file {file_idx + 1}/{len(input_files)}: {input_file.name}")
    splitting_key = f"split_{file_idx}"

    # Read data and map to pixels
    data = pd.read_csv(input_file)
    mapped_pixels = hp.radec2pix(
        mapping_healpix_order, data[ra_column].to_numpy(dtype=float), data[dec_column].to_numpy(dtype=float)
    )

    # Apply alignment to get destination pixels
    aligned_pixels = alignment_array[mapped_pixels]

    print("  Object Distribution:")
    for idx, row in data.iterrows():
        dest_info = aligned_pixels[idx]
        print(f"    ID {row['id']}: → Order {dest_info[0]}, Pixel {dest_info[1]}")

    # Find unique destination pixels in this chunk
    unique_pixels, unique_inverse = np.unique(aligned_pixels, return_inverse=True, axis=0)

    print(f"  Writing {len(unique_pixels)} shard(s) for this file...")

    # For each unique destination pixel, write a shard file
    for unique_index, pixel_alignment_count in enumerate(unique_pixels):
        order = pixel_alignment_count[0]
        pixel = pixel_alignment_count[1]

        # Create directory for this pixel
        healpix_pixel = HealpixPixel(order, pixel)
        pixel_dir = get_pixel_cache_directory(cache_shard_path, healpix_pixel)
        pixel_dir.mkdir(parents=True, exist_ok=True)

        # Filter data for this destination pixel
        filtered_data = data.iloc[unique_inverse == unique_index]

        # Write to parquet shard - note the splitting_key in the filename!
        output_file = pixel_dir / f"shard_{splitting_key}_0.parquet"
        table = pa.Table.from_pandas(filtered_data, preserve_index=False).replace_schema_metadata()
        pq.write_table(table, output_file)

        shard_files_created.append((healpix_pixel, output_file, len(filtered_data)))

        print(
            f"    Order {order}, Pixel {pixel}: {len(filtered_data)} object(s) → {output_file.relative_to(demo_path)}"
        )
        print(f"      Objects: {filtered_data['id'].tolist()}")
    print()

print(f"Shard Summary:")
print(f"  Total shards created: {len(shard_files_created)}")
print(f"\nGrouping shards by destination pixel:")
from collections import defaultdict

shards_by_pixel = defaultdict(list)
for hp_pixel, shard_file, count in shard_files_created:
    shards_by_pixel[(hp_pixel.order, hp_pixel.pixel)].append((shard_file.name, count))

for (order, pixel), shards in sorted(shards_by_pixel.items()):
    print(f"  Order {order}, Pixel {pixel}: {len(shards)} shard(s)")
    for shard_name, count in shards:
        print(f"    - {shard_name} ({count} rows)")

```

--------------------------------

### Read CSV and Convert to Parquet

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/estimate_pixel_threshold.ipynb

Reads a CSV input file using `CsvReader` and converts a sample chunk to a parquet file. Adjust the `input_file` path to your specific data. If your data is already in parquet format, skip this step.

```python
from hats_import.catalog.file_readers import CsvReader

### Change this path!!!
input_file = "../../tests/data/small_sky/catalog.csv"

file_reader = CsvReader(chunksize=5_000)

next(file_reader.read(input_file)).to_parquet(sample_parquet_file)
```

--------------------------------

### Generate Small Sky Source Catalog with NPix Suffix

Source: https://github.com/astronomy-commons/hats-import/blob/main/tests/data/generate_data.ipynb

Generates a source catalog similar to the previous one but includes LEAF pixel directories by setting the `npix_suffix` argument. This configuration affects how output files are organized.

```python
remove_directory("./small_sky_source_npix_dir_catalog")
with tempfile.TemporaryDirectory() as pipeline_tmp:
    args = ImportArguments(
        input_path=Path(hats_import_dir) / "small_sky_source",
        output_path=".",
        file_reader="csv",
        ra_column="source_ra",
        dec_column="source_dec",
        catalog_type="source",
        highest_healpix_order=5,
        pixel_threshold=3000,
        drop_empty_siblings=False,
        output_artifact_name="small_sky_source_npix_dir_catalog",
        tmp_dir=pipeline_tmp,
        npix_suffix="/",  # Directory suffix
    )
    pipeline_with_client(args, client)
```

--------------------------------

### Cite HATS-import with Rubin DP1

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/citation.md

If Rubin Data Preview 1 (DP1) was used with HATS in your research, consider citing this additional paper.

```bibtex
@ARTICLE{2025arXiv250623955M,
    author        = {{Malanchev}, Konstantin and {DeLucchi}, Melissa and {Caplar}, Neven and {Malz}, Alex I. and {Beebe}, Wilson and {Branton}, Doug and {Campos}, Sandro and {Connolly}, Andrew and {Dai}, Mi and {Kubica}, Jeremy and {Lynn}, Olivia and {Mandelbaum}, Rachel and {McGuire}, Sean and {Aubourg}, Eric and {Blum}, Robert David and {Carlin}, Jeffrey L. and {Delgado}, Francisco and {Gangler}, Emmanuel and {Jannuzi}, Buell T. and {Jenness}, Tim and {Kang}, Yijung and {Kannawadi}, Arun and {Moniez}, Marc and {Plazas Malag{\'o}n}, Andr\'es A. and {van Reeven}, Wouter and {Sanmartim}, David and {Urbach}, Elana K. and {Wood-Vasey}, W.~M.},
    title         = "{Variability-finding in Rubin Data Preview 1 with LSDB}",
    journal       = {arXiv e-prints},
    keywords      = {Instrumentation and Methods for Astrophysics, Solar and Stellar Astrophysics},
    year          = 2025,
    month         = jun,
    eid           = {arXiv:2506.23955},
    pages         = {arXiv:2506.23955},
    doi           = {10.48550/arXiv.2506.23955},
    archivePrefix = {arXiv},
    eprint        = {2506.23955},
    primaryClass  = {astro-ph.IM},
    adsurl        = {https://ui.adsabs.harvard.edu/abs/2025arXiv250623955M},
    adsnote       = {Provided by the SAO/NASA Astrophysics Data System}
}
```

--------------------------------

### Display Directory Tree After Cleanup

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/notebooks/pipeline_stages.ipynb

This snippet visualizes the directory structure after the reducing stage and cleanup operations. It's useful for verifying that intermediate files have been removed and the final output is organized as expected.

```python
show_directory_tree(demo_path, "REDUCING (after cleanup)")
```

--------------------------------

### View Verification Pipeline Progress

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/guide/verification.md

Observe the progress reporting during pipeline execution. This output details the stages of verification, including dataset loading, specific tests run, and their results.

```default
Loading dataset and schema.

Starting: Test hats.io.validation.is_valid_catalog (hats version 0.4.6).
Validating catalog at path /data/hats/catalogs/ztf_dr22/ztf_lc ...
Found 10839 partitions.
Approximate coverage is 78.13 % of the sky.
Result: PASSED

Starting: Test that files in _metadata match the data files on disk.
Result: PASSED

Starting: Test that number of rows are equal.
    file footers vs catalog properties
    file footers vs _metadata
Result: PASSED

Starting: Test that schemas are equal, excluding metadata.
    _common_metadata vs truth
    _metadata vs truth
    file footers vs truth
Result: PASSED

Verifier results written to results/ztf_dr22_lc/verifier_results.csv
Elapsed time (seconds): 26.33
```

--------------------------------

### Define Catalog Collection Paths

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/pre_executed/backport_collection.ipynb

Set these variables to define the structure and naming of your catalog collection. Use None if a path is not applicable.

```python
## Set these values based on the paths / subdirectory names shown above
## Or SET TO NONE if there's nothing relevant.

collection_path = "/data3/epyc/data3/hats/catalogs/"
catalog_subdir = "main_catalog"
margin_paths = ["margin_1deg", "margin_20deg"]
default_margin = "margin_1deg"
index_paths = {"id": "id_index"}

## This is a human-readable name of the collection, often the survey or data release.
collection_name = "survey_drK"
```

--------------------------------

### Pipeline with Dask Client

Source: https://github.com/astronomy-commons/hats-import/blob/main/docs/catalogs/arguments.md

Integrates a pre-existing Dask client with the import pipeline. Use this when you need to manage your Dask client separately or within a specific environment.

```python
from dask.distributed import Client
from hats_import.pipeline import pipeline_with_client

args = ...  # ImportArguments()
with Client('scheduler:port') as client:
    pipeline_with_client(args, client)
```