solrad

SOLRAD Reader

SOLRADReader

Bases: PointReader

Reader for NOAA SOLRAD network data.

Source code in monetio/readers/solrad.py

@register_reader("solrad")
class SOLRADReader(PointReader):
    """
    Reader for NOAA SOLRAD network data.
    """

    def open_dataset(
        self,
        files: str | list[str] | None = None,
        dates: datetime | list[datetime] | pd.DatetimeIndex | None = None,
        sites: list[str] | None = None,
        as_xarray: bool = True,
        lazy: bool = False,
        **kwargs: dict,
    ) -> Union[xr.Dataset, pd.DataFrame, "dd.DataFrame"]:
        """
        Open SOLRAD dataset.

        Parameters
        ----------
        files : Union[str, List[str]], optional
            File paths or URLs. If None, uses `dates` and `sites` to discover files.
        dates : Union[datetime, List[datetime], pd.DatetimeIndex], optional
            Dates to retrieve if `files` is None.
        sites : List[str], optional
            Site abbreviations (e.g. ['abq', 'bis', 'hnx', 'msn', 'slc', 'sea', 'ste']).
        as_xarray : bool, optional
            If True, returns an xarray.Dataset, by default True.
        lazy : bool, optional
            If True, returns a dask-backed object, by default False.
        **kwargs : dict
            Additional arguments passed to the reader and driver.

        Returns
        -------
        Union[xr.Dataset, pd.DataFrame, dd.DataFrame]
            The loaded dataset.

        Examples
        --------
        >>> from monetio.readers.solrad import SOLRADReader
        >>> reader = SOLRADReader()
        >>> ds = reader.open_dataset(dates="2024-01-01", sites=["abq"])
        """
        if files is None:
            if dates is None or sites is None:
                raise ValueError("Either 'files' or both 'dates' and 'sites' must be provided.")
            files = self.build_urls(dates, sites)

        # Separate driver kwargs from to_xarray/postprocess kwargs
        driver_kwargs = {
            k: v
            for k, v in kwargs.items()
            if k not in ["expand2d", "wide_fmt", "pivot", "as_xarray", "lazy"]
        }

        # We use read_solrad as the custom read_method
        df = self.driver.open(files, read_method=read_solrad, lazy=lazy, **driver_kwargs)

        # Post-processing: Harmonize column names
        df = self._postprocess(df)

        # Consistently force object strings
        df = force_object_strings(df)

        if as_xarray:
            ds = self.to_xarray(df, **kwargs)
            # Update history for provenance
            ds = update_history(ds, "Read SOLRAD dataset.")
            return ds

        return df

    def _postprocess(
        self, df: Union[pd.DataFrame, "dd.DataFrame"]
    ) -> Union[pd.DataFrame, "dd.DataFrame"]:
        """
        Harmonize column names.

        Parameters
        ----------
        df : Union[pd.DataFrame, dd.DataFrame]
            Input dataframe.

        Returns
        -------
        Union[pd.DataFrame, dd.DataFrame]
            Post-processed dataframe.
        """
        # Rename according to VARIABLE_MAP
        df = df.rename(columns=VARIABLE_MAP)

        # Update history for provenance
        df = update_history(df, "Applied variable name mapping for SOLRAD.")

        return df

    def build_urls(
        self,
        dates: datetime | list[datetime] | pd.DatetimeIndex,
        sites: list[str],
    ) -> list[str]:
        """
        Discover available URLs for the given dates and sites.

        Parameters
        ----------
        dates : Union[datetime, List[datetime], pd.DatetimeIndex]
            Dates to retrieve.
        sites : List[str]
            Site abbreviations.

        Returns
        -------
        List[str]
            List of URLs.
        """
        baseurl = "https://gml.noaa.gov/aftp/data/radiation/solrad/"

        urls = []
        dates = pd.DatetimeIndex(np.atleast_1d(dates))

        for date in dates:
            for site in sites:
                year = date.year
                # Format: site/year/siteyyjday.dat
                # e.g. abq/2024/abq24001.dat
                fname = f"{site.lower()}{date.strftime('%y%j')}.dat"
                url = f"{baseurl}{site.lower()}/{year}/{fname}"
                urls.append(url)

        return urls

build_urls(dates, sites)

Discover available URLs for the given dates and sites.

Parameters:

Name	Type	Description	Default
dates	Union[datetime, List[datetime], DatetimeIndex]	Dates to retrieve.	required
sites	List[str]	Site abbreviations.	required

Returns:

Type	Description
List[str]	List of URLs.

Source code in monetio/readers/solrad.py

def build_urls(
    self,
    dates: datetime | list[datetime] | pd.DatetimeIndex,
    sites: list[str],
) -> list[str]:
    """
    Discover available URLs for the given dates and sites.

    Parameters
    ----------
    dates : Union[datetime, List[datetime], pd.DatetimeIndex]
        Dates to retrieve.
    sites : List[str]
        Site abbreviations.

    Returns
    -------
    List[str]
        List of URLs.
    """
    baseurl = "https://gml.noaa.gov/aftp/data/radiation/solrad/"

    urls = []
    dates = pd.DatetimeIndex(np.atleast_1d(dates))

    for date in dates:
        for site in sites:
            year = date.year
            # Format: site/year/siteyyjday.dat
            # e.g. abq/2024/abq24001.dat
            fname = f"{site.lower()}{date.strftime('%y%j')}.dat"
            url = f"{baseurl}{site.lower()}/{year}/{fname}"
            urls.append(url)

    return urls

open_dataset(files=None, dates=None, sites=None, as_xarray=True, lazy=False, **kwargs)

Open SOLRAD dataset.

Parameters:

Name	Type	Description	Default
files	Union[str, List[str]]	File paths or URLs. If None, uses dates and sites to discover files.	None
dates	Union[datetime, List[datetime], DatetimeIndex]	Dates to retrieve if files is None.	None
sites	List[str]	Site abbreviations (e.g. ['abq', 'bis', 'hnx', 'msn', 'slc', 'sea', 'ste']).	None
as_xarray	bool	If True, returns an xarray.Dataset, by default True.	True
lazy	bool	If True, returns a dask-backed object, by default False.	False
**kwargs	dict	Additional arguments passed to the reader and driver.	{}

Returns:

Type	Description
Union[Dataset, DataFrame, DataFrame]	The loaded dataset.

Examples:

>>> from monetio.readers.solrad import SOLRADReader
>>> reader = SOLRADReader()
>>> ds = reader.open_dataset(dates="2024-01-01", sites=["abq"])

Source code in monetio/readers/solrad.py

def open_dataset(
    self,
    files: str | list[str] | None = None,
    dates: datetime | list[datetime] | pd.DatetimeIndex | None = None,
    sites: list[str] | None = None,
    as_xarray: bool = True,
    lazy: bool = False,
    **kwargs: dict,
) -> Union[xr.Dataset, pd.DataFrame, "dd.DataFrame"]:
    """
    Open SOLRAD dataset.

    Parameters
    ----------
    files : Union[str, List[str]], optional
        File paths or URLs. If None, uses `dates` and `sites` to discover files.
    dates : Union[datetime, List[datetime], pd.DatetimeIndex], optional
        Dates to retrieve if `files` is None.
    sites : List[str], optional
        Site abbreviations (e.g. ['abq', 'bis', 'hnx', 'msn', 'slc', 'sea', 'ste']).
    as_xarray : bool, optional
        If True, returns an xarray.Dataset, by default True.
    lazy : bool, optional
        If True, returns a dask-backed object, by default False.
    **kwargs : dict
        Additional arguments passed to the reader and driver.

    Returns
    -------
    Union[xr.Dataset, pd.DataFrame, dd.DataFrame]
        The loaded dataset.

    Examples
    --------
    >>> from monetio.readers.solrad import SOLRADReader
    >>> reader = SOLRADReader()
    >>> ds = reader.open_dataset(dates="2024-01-01", sites=["abq"])
    """
    if files is None:
        if dates is None or sites is None:
            raise ValueError("Either 'files' or both 'dates' and 'sites' must be provided.")
        files = self.build_urls(dates, sites)

    # Separate driver kwargs from to_xarray/postprocess kwargs
    driver_kwargs = {
        k: v
        for k, v in kwargs.items()
        if k not in ["expand2d", "wide_fmt", "pivot", "as_xarray", "lazy"]
    }

    # We use read_solrad as the custom read_method
    df = self.driver.open(files, read_method=read_solrad, lazy=lazy, **driver_kwargs)

    # Post-processing: Harmonize column names
    df = self._postprocess(df)

    # Consistently force object strings
    df = force_object_strings(df)

    if as_xarray:
        ds = self.to_xarray(df, **kwargs)
        # Update history for provenance
        ds = update_history(ds, "Read SOLRAD dataset.")
        return ds

    return df

get_colspecs(widths)

Generate colspecs for pd.read_fwf from widths.

Parameters:

Name	Type	Description	Default
widths	List[int]	List of column widths.	required

Returns:

Type	Description
List[tuple]	List of (start, end) tuples.

Source code in monetio/readers/solrad.py

def get_colspecs(widths: list[int]) -> list[tuple]:
    """
    Generate colspecs for pd.read_fwf from widths.

    Parameters
    ----------
    widths : List[int]
        List of column widths.

    Returns
    -------
    List[tuple]
        List of (start, end) tuples.
    """
    colspecs = []
    start = 0
    for w in widths:
        colspecs.append((start, start + w))
        start += w + 1
    return colspecs

read_solrad(filename, **kwargs)

Read a single SOLRAD file.

Parameters:

Name	Type	Description	Default
filename	str	The path or URL to the SOLRAD file.	required
**kwargs	dict	Additional arguments passed to pd.read_fwf.	{}

Returns:

Type	Description
DataFrame	The loaded data in long format.

Examples:

>>> df = read_solrad("abq24001.dat")

Source code in monetio/readers/solrad.py

def read_solrad(filename: str, **kwargs: dict) -> pd.DataFrame:
    """
    Read a single SOLRAD file.

    Parameters
    ----------
    filename : str
        The path or URL to the SOLRAD file.
    **kwargs : dict
        Additional arguments passed to pd.read_fwf.

    Returns
    -------
    pd.DataFrame
        The loaded data in long format.

    Examples
    --------
    >>> df = read_solrad("abq24001.dat")
    """
    if "msn" in filename.lower():
        names = MADISON_HEADERS
        colspecs = MADISON_COLSPECS
    else:
        names = HEADERS
        colspecs = COLSPECS

    # Use FileUtility to handle remote files
    fs = FileUtility.get_fs(filename)
    storage_options = kwargs.get("storage_options", {})

    with fs.open(filename, "r", **storage_options) as f:
        # Read header for metadata
        header_lines = []
        for _ in range(2):
            line = f.readline()
            if not line:
                break
            header_lines.append(line.strip())

        if len(header_lines) < 2:
            return pd.DataFrame()

        station_name = header_lines[0]
        metadata_line = header_lines[1].split()

        try:
            latitude = float(metadata_line[0])
            longitude = float(metadata_line[1])
            elevation = float(metadata_line[2])
        except (ValueError, IndexError):
            latitude = np.nan
            longitude = np.nan
            elevation = np.nan

        data_content = f.read()

    df = pd.read_fwf(
        io.StringIO(data_content),
        colspecs=colspecs,
        header=None,
        names=names,
        na_values=-9999.9,
        **kwargs,
    )

    if df.empty:
        return df

    # Add metadata
    df["latitude"] = latitude
    df["longitude"] = longitude
    df["elevation"] = elevation
    df["siteid"] = station_name

    # Vectorized time construction
    # year(4), month(2), day(2), hour(2), minute(2)
    def to_str(series, n):
        # Handle cases where column might be floating point because of NaNs
        s = pd.to_numeric(series, errors="coerce").fillna(0).astype(int).astype(str)
        return s.str.zfill(n)

    df["time"] = pd.to_datetime(
        to_str(df["year"], 4)
        + to_str(df["month"], 2)
        + to_str(df["day"], 2)
        + to_str(df["hour"], 2)
        + to_str(df["minute"], 2),
        format="%Y%m%d%H%M",
        errors="coerce",
    )

    return df