New example app featuring astropy

Author: jcheng5

examples/airmass/app.py

@@ -0,0 +1,175 @@
+import datetime
+from typing import Dict, List, Optional, Tuple
+
+import astropy.units as u
+import matplotlib.dates as mpldates
+import matplotlib.pyplot as plt
+import numpy as np
+import pandas as pd
+import pytz
+import suntime
+import timezonefinder
+from astropy.coordinates import AltAz, EarthLocation, SkyCoord
+from shiny import App, Inputs, Outputs, Session, reactive, render, req, ui
+
+from location import location_server, location_ui
+
+app_ui = ui.page_fixed(
+    ui.tags.h3("Air mass calculator", class_="mb-5"),
+    ui.row(
+        ui.column(
+            8,
+            ui.output_ui("timeinfo"),
+            ui.output_plot("plot", height="800px"),
+            # For debugging
+            # ui.output_table("table"),
+            class_="order-2 order-sm-1",
+        ),
+        ui.column(
+            4,
+            ui.panel_well(
+                ui.input_date("date", "Date"),
+                class_="pb-1 mb-3",
+            ),
+            ui.panel_well(
+                ui.input_text_area(
+                    "objects", "Target object(s)", "Sirius, Betelgeuse, NGC35", rows=3
+                ),
+                class_="pb-1 mb-3",
+            ),
+            ui.panel_well(
+                location_ui("location", lat=0, long=0),
+                class_="mb-3",
+            ),
+            class_="order-1 order-sm-2",
+        ),
+    ),
+)
+
+
+def server(input: Inputs, output: Outputs, session: Session):
+    loc = location_server("location")
+    time_padding = datetime.timedelta(hours=1.5)
+
+    @reactive.Calc
+    def obj_names() -> List[str]:
+        """Returns a split and *slightly* cleaned-up list of object names"""
+        req(input.objects())
+        return [x.strip() for x in input.objects().split(",") if x.strip() != ""]
+
+    @reactive.Calc
+    def obj_coords() -> List[SkyCoord]:
+        return [SkyCoord.from_name(name) for name in obj_names()]
+
+    @reactive.Calc
+    def times_utc() -> Tuple[datetime.datetime, datetime.datetime]:
+        req(input.date())
+        lat, long = loc()
+        sun = suntime.Sun(lat, long)
+        return (
+            sun.get_sunset_time(input.date()),
+            sun.get_sunrise_time(input.date() + datetime.timedelta(days=1)),
+        )
+
+    @reactive.Calc
+    def timezone() -> Optional[str]:
+        lat, long = loc()
+        return timezonefinder.TimezoneFinder().timezone_at(lat=lat, lng=long)
+
+    @reactive.Calc
+    def times_at_loc():
+        start, end = times_utc()
+        tz = pytz.timezone(timezone())
+        return (start.astimezone(tz), end.astimezone(tz))
+
+    @reactive.Calc
+    def df() -> Dict[str, pd.DataFrame]:
+        start, end = times_at_loc()
+        times = pd.date_range(
+            start - time_padding,
+            end + time_padding,
+            periods=100,
+        )
+        lat, long = loc()
+        eloc = EarthLocation(lat=lat * u.deg, lon=long * u.deg, height=0)
+        altaz_list = [
+            obj.transform_to(AltAz(obstime=times, location=eloc))
+            for obj in obj_coords()
+        ]
+        return {
+            obj: pd.DataFrame(
+                {
+                    "obj": obj,
+                    "time": times,
+                    "alt": altaz.alt,
+                    # Filter out discontinuity
+                    "secz": np.where(altaz.alt > 0, altaz.secz, np.nan),
+                }
+            )
+            for (altaz, obj) in zip(altaz_list, obj_names())
+        }
+
+    @output
+    @render.plot
+    def plot():
+        fig, [ax1, ax2] = plt.subplots(nrows=2)
+
+        sunset, sunrise = times_at_loc()
+
+        def add_boundary(ax, xval):
+            ax.axvline(x=xval, c="#888888", ls="dashed")
+
+        ax1.set_ylabel("Altitude (deg)")
+        ax1.set_xlabel("Time")
+        ax1.set_ylim(-10, 90)
+        ax1.set_xlim(sunset - time_padding, sunrise + time_padding)
+        ax1.grid()
+        add_boundary(ax1, sunset)
+        add_boundary(ax1, sunrise)
+        for obj_name, data in df().items():
+            ax1.plot(data["time"], data["alt"], label=obj_name)
+        ax1.xaxis.set_major_locator(mpldates.AutoDateLocator())
+        ax1.xaxis.set_major_formatter(
+            mpldates.DateFormatter("%H:%M", tz=pytz.timezone(timezone()))
+        )
+        ax1.legend(loc="upper right")
+
+        ax2.set_ylabel("Air mass")
+        ax2.set_xlabel("Time")
+        ax2.set_ylim(4, 1)
+        ax2.set_xlim(sunset - time_padding, sunrise + time_padding)
+        ax2.grid()
+        add_boundary(ax2, sunset)
+        add_boundary(ax2, sunrise)
+        for data in df().values():
+            ax2.plot(data["time"], data["secz"])
+        ax2.xaxis.set_major_locator(mpldates.AutoDateLocator())
+        ax2.xaxis.set_major_formatter(
+            mpldates.DateFormatter("%H:%M", tz=pytz.timezone(timezone()))
+        )
+
+        return fig
+
+    @output
+    @render.table
+    def table() -> pd.DataFrame:
+        return pd.concat(df())
+
+    @output
+    @render.ui
+    def timeinfo():
+        start_utc, end_utc = times_utc()
+        start_at_loc, end_at_loc = times_at_loc()
+        return ui.TagList(
+            f"Sunset: {start_utc.strftime('%H:%M')}, ",
+            f"Sunrise: {end_utc.strftime('%H:%M')} ",
+            "(UTC)",
+            ui.tags.br(),
+            f"Sunset: {start_at_loc.strftime('%H:%M')}, ",
+            f"Sunrise: {end_at_loc.strftime('%H:%M')} ",
+            f"({timezone()})",
+        )
+
+
+# The debug=True causes it to print messages to the console.
+app = App(app_ui, server, debug=True)

examples/airmass/location.py

@@ -0,0 +1,128 @@
+from typing import Optional
+
+import ipyleaflet as L
+from shiny import Inputs, Outputs, Session, module, reactive, req, ui
+from shinywidgets import output_widget, reactive_read, register_widget
+
+# ============================================================
+# Module: location
+# ============================================================
+
+
+@module.ui
+def location_ui(
+    label: str = "Location",
+    *,
+    lat: Optional[float],
+    long: Optional[float],
+) -> ui.TagChildArg:
+    return ui.div(
+        ui.input_numeric("lat", "Latitude", value=None),
+        ui.input_numeric("long", "Longitude", value=None),
+        ui.help_text("Click to select location"),
+        output_widget("map", height="200px"),
+        ui.tags.style(
+            """
+            .jupyter-widgets.leaflet-widgets {
+                height: 100% !important;
+            }
+            """
+        ),
+    )
+
+
+@module.server
+def location_server(
+    input: Inputs, output: Outputs, session: Session, *, wrap_long: bool = True
+):
+    map = L.Map(center=(0, 0), zoom=1, scoll_wheel_zoom=True)
+    marker = L.Marker(location=(52, 100))
+
+    @reactive.Effect
+    @reactive.isolate()
+    def _():
+        if not input.lat() and not input.long():
+            ui.insert_ui(
+                ui.tags.script(
+                    """
+                    navigator.geolocation.getCurrentPosition(
+                        ({coords}) => {
+                            const {latitude, longitude, altitude} = coords;
+                            Shiny.setInputValue("#HERE#", {latitude, longitude});
+                        },
+                        (err) => {
+                            Shiny.setInputValue("#HERE#", null);
+                        },
+                        {maximumAge: Infinity, timeout: Infinity}
+                    )
+                    """.replace(
+                        "#HERE#", module.resolve_id("here")
+                    )
+                ),
+                selector="body",
+                where="beforeEnd",
+                immediate=True,
+            )
+
+    def move(lat: float, long: float) -> None:
+        lat = round(lat, 8)
+        long = round(long, 8)
+        marker.location = (lat, long)
+        if marker not in map.layers:
+            map.add_layer(marker)
+        ui.update_text("lat", value=lat)
+        ui.update_text("long", value=long)
+
+    def on_map_interaction(**kwargs):
+        if kwargs.get("type") == "click":
+            coords = kwargs.get("coordinates")
+            move(coords[0], coords[1])
+
+    map.on_interaction(on_map_interaction)
+
+    def on_marker_move():
+        move(marker.location[0], marker.location[1])
+
+    marker.on_move(on_marker_move)
+
+    register_widget("map", map)
+
+    @reactive.Effect
+    def detect_location():
+        coords = input.here()
+        if coords and not input.lat() and not input.long():
+            ui.update_numeric("lat", value=coords["latitude"])
+            ui.update_numeric("long", value=coords["longitude"])
+
+    @reactive.Effect
+    def sync_map_lat():
+        req(input.lat() is not None)
+        lat = float(input.lat())
+        if marker.location[0] != lat:
+            marker.location = (lat, marker.location[1])
+        if marker not in map.layers:
+            map.add_layer(marker)
+        map.center = marker.location
+
+    @reactive.Effect
+    def sync_map_long():
+        req(input.long() is not None)
+        long = float(input.long())
+        if marker.location[1] != long:
+            marker.location = (marker.location[0], long)
+        if marker not in map.layers:
+            map.add_layer(marker)
+        map.center = marker.location
+
+    @reactive.Calc
+    def location():
+        req(input.lat() is not None, input.long() is not None)
+        long = float(input.long())
+        if wrap_long:
+            long = (long + 180) % 360 - 180
+        try:
+            return (float(input.lat()), long)
+        except ValueError:
+            raise ValueError("Invalid latitude/longitude specification")
+
+    return location

examples/airmass/requirements.in

@@ -0,0 +1,10 @@
+astropy
+ipyleaflet
+matplotlib
+numpy
+pandas
+pytz
+shiny
+shinywidgets
+suntime
+timezonefinder