utils.py

import logging
from io import BytesIO
from pathlib import Path

import holoviews as hv
import numpy as np
import pandas as pd
import panel as pn
import pyarrow.parquet as pq
from fsspec.parquet import open_parquet_file
from holoviews import opts
from PIL import Image

FORMAT = "%(asctime)s | %(levelname)s | %(name)s | %(message)s"

MAJOR_TOM_LOGO = "assets/major-tom-esa-logo.png"
MAJOR_TOM_PICTURE = (
    "https://upload.wikimedia.org/wikipedia/en/6/6d/Major_tom_space_oddity_video.JPG"
)
MAJOR_TOM_REF_URL = "https://huggingface.co/Major-TOM"
MAJOR_TOM_ARXIV_URL = "https://www.arxiv.org/abs/2402.12095"
PANEL_LOGO = "https://panel.holoviz.org/_static/logo_horizontal_light_theme.png"
PANEL_URL = "https://panel.holoviz.org"
DATASHADER_LOGO = "https://datashader.org/_static/logo_horizontal.svg"
DATASHADER_URL = "https://datashader.org/"
REPOSITORY = "Major-TOM"
DEFAULT_DATASET = "Core-S2L2A"
DATASETS = ["Core-S1RTC", "Core-S2L2A", "Core-S2L1C"]
ESA_EASTING = 250668.73322714816
ESA_NORTHING = 6259216.653115547
_DATASET_COLUMNS_1 = {
    # "Linear Power in the VV Polarization": "vv",
    # "Linear Power in the VH Polarization": "vh",
    "Thumbnail": "thumbnail",
}
_DATASET_COLUMNS_2 = {
    "Coastal aerosol": "B01",
    "Blue": "B02",
    "Green": "B03",
    "Red": "B04",
    "Vegetation Blue": "B05",
    "Vegetation Green": "B06",
    "Vegetation Red": "B07",
    "NIR": "B08",
    "Narrow NIR": "B8A",
    "Water vapour": "B09",
    "SWIR, 1613.7": "B11",
    "SWIR, 2202.4": "B12",
    "Cloud Mask": "cloud_mask",
    "Thumbnail": "thumbnail",
}
DATASET_COLUMNS = {
    "Core-S1RTC": _DATASET_COLUMNS_1,
    "Core-S2L2A": _DATASET_COLUMNS_2,
    "Core-S2L1C": _DATASET_COLUMNS_2,
}
DATA_PATH = Path(__file__).parent / "data"

DESCRIPTION = f"""\
## Dataset Explorer

This app provides a way of exploring samples present in the [MajorTOM-Core]({MAJOR_TOM_REF_URL}) dataset. It contains nearly every piece of Earth captured by ESA [Sentinel-2](https://sentinels.copernicus.eu/web/sentinel/missions/sentinel-2) satellite.

[Website]({MAJOR_TOM_REF_URL}), [arXiv Paper]({MAJOR_TOM_ARXIV_URL})

## Instructions

To find a sample, navigate on the map to a place of interest. Click the map to find a dataset sample at the location you clicked.

## Powered by

"""

MAJOR_TOM_LYRICS = """
Standing there alone, the ship is waiting
All systems are go, are you sure?
Control is not convinced, but the computer
Has the evidence, no need to abort
The countdown starts

Watching in a trance, the crew is certain
Nothing left to chance, all is working
Trying to relax up in the capsule
"Send me up a drink, " jokes **Major Tom**
The count goes on

Four, three, two, one

Earth below us, drifting, falling
Floating weightless, calling, calling home

Second stage is cut, we're now in orbit
Stabilizers up, running perfect
Starting to collect requested data
"What will it affect when all is done?"
Thinks **Major Tom**

Back at ground control, there is a problem
Go to rockets full, not responding
"Hello **Major Tom**, are you receiving?
Turn the thrusters on, we're standing by"
There's no reply

Four, three, two, one

Earth below us, drifting, falling
Floating weightless, calling, calling home

Across the stratosphere a final message
"Give my wife my love, " then nothing more
Far beneath the ship, the world is mourning
They don't realize he's alive
No one understands, but **Major Tom** sees
"Now the light commands, this is my home
I'm coming home"

Earth below us, drifting, falling
Floating weightless, coming home
Earth below us, drifting, falling
Floating weightless, coming home
Earth below us, drifting, falling
Floating weightless, coming, coming home

Home
Home
Home
Home
Home
"""

hv.extension("bokeh")

opts.defaults(
    # opts.Curve(xaxis=None, yaxis=None, show_grid=False, show_frame=False,
    #            color='orangered', framewise=True, width=100),
    opts.HLine(color="gray", line_width=1),
    # opts.Layout(shared_axes=False),
    opts.VLine(color="gray", line_width=1),
)


def _meta_data_url(dataset="Core-S2L2A", repository=REPOSITORY):
    return f"https://huggingface.co/datasets/{repository}/{dataset}/resolve/main/metadata.parquet"


def _meta_data_path(dataset="Core-S2L2A", repository=REPOSITORY):
    DATA_PATH.mkdir(parents=True, exist_ok=True)
    return DATA_PATH / f"{dataset}_metadata.parquet"


def get_meta_data(dataset="Core-S2L2A", repository=REPOSITORY):
    logging.info("Loading %s", dataset)
    path = _meta_data_path(dataset=dataset)
    if not path.exists():
        data = pd.read_parquet(_meta_data_url(dataset=dataset))
        data.to_parquet(path)
    data = pd.read_parquet(path)

    data["centre_easting"], data["centre_northing"] = (
        hv.util.transform.lon_lat_to_easting_northing(
            data["centre_lon"], data["centre_lat"]
        )
    )
    # Optimize Performance
    data["timestamp"] = pd.to_datetime(data["timestamp"])
    numeric_cols = data.select_dtypes(include=["float64"]).columns
    data[numeric_cols] = data[numeric_cols].astype("float32")

    return data


def get_image(row, column="thumbnail"):
    parquet_url = row["parquet_url"]
    parquet_row = row["parquet_row"]
    with open_parquet_file(parquet_url, columns=[column]) as f:
        with pq.ParquetFile(f) as pf:
            first_row_group = pf.read_row_group(parquet_row, columns=[column])

    stream = BytesIO(first_row_group[column][0].as_py())
    image = Image.open(stream)
    return image


def euclidean_distance(x, y, target_x, target_y):
    return np.sqrt((x - target_x) ** 2 + (y - target_y) ** 2)


def get_closest_row(data, target_easting, target_northing):
    distance = euclidean_distance(
        data["centre_easting"], data["centre_northing"], target_easting, target_northing
    )
    closest_row = data.loc[distance.idxmin()]
    return closest_row


def get_closest_rows(data, target_easting, target_northing):
    distance = euclidean_distance(
        data["centre_easting"], data["centre_northing"], target_easting, target_northing
    )
    closest_rows = data[distance == distance.min()]
    return closest_rows


def reconfig_basic_config(format_=FORMAT, level=logging.INFO):
    """(Re-)configure logging"""
    logging.basicConfig(format=format_, level=level, force=True)
    logging.info("Logging.basicConfig completed successfully")


reconfig_basic_config()