-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutils.py
117 lines (94 loc) · 3.72 KB
/
utils.py
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
# -*- coding: UTF-8 -*-
"""
@File ：utils.py
@Author ：heaven
@Date ：2023/3/1 20:16
"""
# Licensed under the MIT license - see LICENSE.txt
import astropy.units as u
import pandas as pd
from astroplan import FixedTarget
from astroplan import Schedule
from astroplan.observer import Observer
from astropy.coordinates import SkyCoord, EarthLocation
from astropy.time import Time
from rich.progress import Progress, TimeElapsedColumn
def df2Targets(df: pd.DataFrame):
coord = SkyCoord(ra=df['_RAJ2000'] * u.deg,
dec=df['_DEJ2000'] * u.deg)
star = FixedTarget(coord=coord, name=df['tyc2-id'])
return star
location = EarthLocation(lat=43.82416667 * u.deg, lon=126.331111 * u.deg, height=313 * u.m)
observer = Observer(location)
observing_date = Time('2023-06-08')
obs_start = observer.twilight_evening_astronomical(time=observing_date, which='next')
obs_end = observer.twilight_morning_astronomical(time=obs_start, which='next')
def schedule2df_ex(schedule: Schedule):
columns = ['name',
'start',
'end',
'RA_ICRS_',
'DE_ICRS_',
'priority',
'configuration']
schedule_df = pd.DataFrame(
columns=columns)
for slot in schedule.slots:
if hasattr(slot.block, 'target'):
start_times = slot.start.iso
end_times = slot.end.iso
target_names = slot.block.target.name
ra = slot.block.target.ra.value
dec = slot.block.target.dec.value
config = str(slot.block.configuration)
priority = slot.block.priority
else:
continue
tmp = pd.Series([target_names, start_times, end_times, ra, dec, priority,config],
index=columns
)
schedule_df = pd.concat([schedule_df, tmp.to_frame().T], ignore_index=True)
schedule_df = schedule_df.infer_objects()
schedule_df['name'] = schedule_df['name'].astype(str)
schedule_df['start'] = schedule_df['start'].astype('datetime64[ns]')
schedule_df['end'] = schedule_df['end'].astype('datetime64[ns]')
schedule_df = schedule_df.sort_values('start')
return schedule_df
def schedule2df(schedule: Schedule):
progress = Progress(
*Progress.get_default_columns(),
TimeElapsedColumn()
)
columns = ['name',
'start',
'end',
'_RAJ2000',
'_DEJ2000',
'priority',
'configuration']
with progress:
task = progress.add_task('to file', total=len(schedule.slots))
schedule_df = pd.DataFrame(
columns=columns)
for slot in schedule.slots:
progress.update(task, advance=1)
if hasattr(slot.block, 'target'):
start_times = slot.start.iso
end_times = slot.end.iso
target_names = slot.block.target.name
ra = slot.block.target.ra.value
dec = slot.block.target.dec.value
config = str(slot.block.configuration)
priority = slot.block.priority
else:
continue
tmp = pd.Series([target_names, start_times, end_times, ra, dec, priority, config],
index=columns
)
schedule_df = pd.concat([schedule_df, tmp.to_frame().T], ignore_index=True)
schedule_df = schedule_df.infer_objects()
schedule_df['name'] = schedule_df['name'].astype(str)
schedule_df['start'] = schedule_df['start'].astype('datetime64[ns]')
schedule_df['end'] = schedule_df['end'].astype('datetime64[ns]')
schedule_df = schedule_df.sort_values('start')
return schedule_df