starplot-hyg/build.py at main · steveberardi/starplot-hyg · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
import csv
import time
from pathlib import Path

from shapely import Point

from starplot import Star
from starplot.data import Catalog


__version__ = "0.1.0"

HERE = Path(__file__).resolve().parent
DATA_PATH = HERE / "data"
BUILD_PATH = HERE / "build"


def parse_int(value):
    return int(value) if value else None


def parse_float(value):
    return float(value) if value else None


def stars():
    with open(DATA_PATH / "hygdata_v42.csv", "r") as hygfile:
        reader = csv.DictReader(hygfile)

        for row in reader:
            ra = parse_float(row.get("ra")) * 15
            dec = parse_float(row.get("dec"))
            geometry = Point(ra, dec)
            name = row.get("proper")

            if not geometry.is_valid or geometry.is_empty or name == "Sol":
                continue

            yield Star(
                pk=parse_int(row.get("id")),
                hip=parse_int(row.get("hip")),
                ra=ra,
                dec=parse_float(row.get("dec")),
                constellation_id=row.get("con").lower(),
                magnitude=parse_float(row.get("mag")),
                ra_mas_per_year=parse_float(row.get("pmra")) or 0,
                dec_mas_per_year=parse_float(row.get("pmdec")) or 0,
                bv=parse_float(row.get("ci")),
                geometry=Point(ra, dec),
                epoch_year=2000,
                bayer=row.get("bayer"),
                flamsteed=row.get("flamsteed"),
                name=name,
            )


def build():
    print("Building HYG Star Catalog...")
    time_start = time.time()

    catalog = Catalog(path=BUILD_PATH / "hyg-v4.2.parquet")
    catalog.build(
        objects=stars(),
        chunk_size=200_000,
        columns=[
            "pk",
            "hip",
            "ra",
            "dec",
            "magnitude",
            "bv",
            "ra_mas_per_year",
            "dec_mas_per_year",
            "constellation_id",
            "geometry",
            "ccdm",
            "epoch_year",
            "name",
            "bayer",
            "flamsteed",
        ],
        partition_columns=[],
        sorting_columns=["magnitude"],
        compression="snappy",
        row_group_size=100_000,
    )

    all_stars = [s for s in Star.all(catalog=catalog)]
    assert len(all_stars) == 119_625

    sirius = Star.get(name="Sirius", catalog=catalog)
    assert sirius.magnitude == -1.44
    assert sirius.hip == 32349
    assert sirius.constellation_id == "cma"

    duration = time.time() - time_start
    print(f"Done. {duration:.0f}s")


if __name__ == "__main__":
    build()