Skip to content

improve

IMPROVE Reader

IMPROVEReader

Bases: PointReader

Reader for IMPROVE (Interagency Monitoring of Protected Visual Environments) data.

Source code in monetio/readers/improve.py 
 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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@register_reader("improve")
class IMPROVEReader(PointReader):
    """
    Reader for IMPROVE (Interagency Monitoring of Protected Visual Environments) data.
    """

    def open_dataset(
        self,
        files: str | list[str],
        add_meta: bool = False,
        delimiter: str = "\t",
        as_xarray: bool = True,
        lazy: bool = False,
        pivot: bool = True,
        **kwargs: Any,
    ) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
        """
        Retrieve and load IMPROVE data.

        Parameters
        ----------
        files : Union[str, List[str]]
            File path, list of paths, or glob pattern.
        add_meta : bool, optional
            Whether to add site metadata, by default False.
        delimiter : str, optional
            Delimiter used in the IMPROVE file, by default "\\t".
        as_xarray : bool, optional
            Whether to return an xarray.Dataset, by default True.
        lazy : bool, optional
            Whether to return a dask-backed object, by default False.
        pivot : bool, optional
            Whether to pivot the data to wide format, by default True.
        **kwargs : Any
            Additional arguments passed to the reader and driver.

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

        Examples
        --------
        >>> reader = IMPROVEReader()
        >>> ds = reader.open_dataset("improve_data.txt")
        """
        # Use PandasDriver via base class
        read_func = partial(read_improve_file, delimiter=delimiter)

        driver_kwargs = kwargs.copy()
        for k in ["expand2d", "pivot", "add_meta", "as_xarray", "lazy"]:
            driver_kwargs.pop(k, None)

        df = super().open_dataset(
            files,
            read_method=read_func,
            as_xarray=False,
            lazy=lazy,
            **driver_kwargs,
        )

        # Check for empty (Backend-agnostic)
        if dd is not None and isinstance(df, dd.DataFrame):
            is_empty = df.npartitions == 0
        else:
            is_empty = df.empty

        if is_empty:
            if as_xarray:
                return xr.Dataset()
            return df

        if add_meta:
            df = self.add_metadata(df)

        # Re-harmonize to drop sites without metadata (NaN lat/lon)
        df = self.harmonize(df)

        if as_xarray:
            ds = self.to_xarray(df, pivot=pivot, **kwargs)
            # Update history
            ds = update_history(ds, "Read IMPROVE data.")
            return ds

        return df

    def add_metadata(
        self, df: Union[pd.DataFrame, "dd.DataFrame"]
    ) -> Union[pd.DataFrame, "dd.DataFrame"]:
        """
        Add site metadata from the IMPROVE monitor file.

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

        Returns
        -------
        Union[pd.DataFrame, dd.DataFrame]
            Dataframe with metadata merged.

        Examples
        --------
        >>> df = reader.add_metadata(df)
        """
        df = add_monitor_metadata(
            df,
            network="IMPROVE",
            left_on="epaid",
            history_msg="Merged with IMPROVE station metadata.",
        )

        # Handle IMPROVE-specific column name cleanup
        if "state_name_y" in df.columns:
            df = df.drop(columns=["state_name_y"], errors="ignore").rename(
                columns={"state_name_x": "state_name"}
            )

        return df

add_metadata(df)

Add site metadata from the IMPROVE monitor file.

Parameters:

Name Type Description Default
df Union[DataFrame, DataFrame]

Input dataframe.

 required

Returns:

Type Description
Union[DataFrame, DataFrame]

Dataframe with metadata merged.

Examples:

>>> df = reader.add_metadata(df)
Source code in monetio/readers/improve.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def add_metadata(
    self, df: Union[pd.DataFrame, "dd.DataFrame"]
) -> Union[pd.DataFrame, "dd.DataFrame"]:
    """
    Add site metadata from the IMPROVE monitor file.

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

    Returns
    -------
    Union[pd.DataFrame, dd.DataFrame]
        Dataframe with metadata merged.

    Examples
    --------
    >>> df = reader.add_metadata(df)
    """
    df = add_monitor_metadata(
        df,
        network="IMPROVE",
        left_on="epaid",
        history_msg="Merged with IMPROVE station metadata.",
    )

    # Handle IMPROVE-specific column name cleanup
    if "state_name_y" in df.columns:
        df = df.drop(columns=["state_name_y"], errors="ignore").rename(
            columns={"state_name_x": "state_name"}
        )

    return df

open_dataset(files, add_meta=False, delimiter='\t', as_xarray=True, lazy=False, pivot=True, **kwargs)

Retrieve and load IMPROVE data.

Parameters:

Name Type Description Default
files Union[str, List[str]]

File path, list of paths, or glob pattern.

 required
add_meta bool

Whether to add site metadata, by default False.

 False
delimiter str

Delimiter used in the IMPROVE file, by default "\t".

 '\t'
as_xarray bool

Whether to return an xarray.Dataset, by default True.

 True
lazy bool

Whether to return a dask-backed object, by default False.

 False
pivot bool

Whether to pivot the data to wide format, by default True.

 True
**kwargs Any

Additional arguments passed to the reader and driver.

 {}

Returns:

Type Description
Union[DataFrame, Dataset, DataFrame]

The loaded IMPROVE data.

Examples:

>>> reader = IMPROVEReader()
>>> ds = reader.open_dataset("improve_data.txt")
Source code in monetio/readers/improve.py 
 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
def open_dataset(
    self,
    files: str | list[str],
    add_meta: bool = False,
    delimiter: str = "\t",
    as_xarray: bool = True,
    lazy: bool = False,
    pivot: bool = True,
    **kwargs: Any,
) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
    """
    Retrieve and load IMPROVE data.

    Parameters
    ----------
    files : Union[str, List[str]]
        File path, list of paths, or glob pattern.
    add_meta : bool, optional
        Whether to add site metadata, by default False.
    delimiter : str, optional
        Delimiter used in the IMPROVE file, by default "\\t".
    as_xarray : bool, optional
        Whether to return an xarray.Dataset, by default True.
    lazy : bool, optional
        Whether to return a dask-backed object, by default False.
    pivot : bool, optional
        Whether to pivot the data to wide format, by default True.
    **kwargs : Any
        Additional arguments passed to the reader and driver.

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

    Examples
    --------
    >>> reader = IMPROVEReader()
    >>> ds = reader.open_dataset("improve_data.txt")
    """
    # Use PandasDriver via base class
    read_func = partial(read_improve_file, delimiter=delimiter)

    driver_kwargs = kwargs.copy()
    for k in ["expand2d", "pivot", "add_meta", "as_xarray", "lazy"]:
        driver_kwargs.pop(k, None)

    df = super().open_dataset(
        files,
        read_method=read_func,
        as_xarray=False,
        lazy=lazy,
        **driver_kwargs,
    )

    # Check for empty (Backend-agnostic)
    if dd is not None and isinstance(df, dd.DataFrame):
        is_empty = df.npartitions == 0
    else:
        is_empty = df.empty

    if is_empty:
        if as_xarray:
            return xr.Dataset()
        return df

    if add_meta:
        df = self.add_metadata(df)

    # Re-harmonize to drop sites without metadata (NaN lat/lon)
    df = self.harmonize(df)

    if as_xarray:
        ds = self.to_xarray(df, pivot=pivot, **kwargs)
        # Update history
        ds = update_history(ds, "Read IMPROVE data.")
        return ds

    return df

read_improve_file(fname, delimiter='\t', **kwargs)

Read a single IMPROVE data file.

Parameters:

Name Type Description Default
fname str

File path or URL.

 required
delimiter str

Delimiter used in the file, by default "\t".

 '\t'
**kwargs Any

Additional arguments passed to pd.read_csv.

 {}

Returns:

Type Description
DataFrame

The loaded data.

Examples:

>>> df = read_improve_file("site_data.txt")
Source code in monetio/readers/improve.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def read_improve_file(fname: str, delimiter: str = "\t", **kwargs: Any) -> pd.DataFrame:
    """
    Read a single IMPROVE data file.

    Parameters
    ----------
    fname : str
        File path or URL.
    delimiter : str, optional
        Delimiter used in the file, by default "\\t".
    **kwargs : Any
        Additional arguments passed to pd.read_csv.

    Returns
    -------
    pd.DataFrame
        The loaded data.

    Examples
    --------
    >>> df = read_improve_file("site_data.txt")
    """
    # Determine storage options if S3
    storage_options = kwargs.get("storage_options")
    if fname.startswith("s3://") and storage_options is None:
        storage_options = {"anon": True}

    # Find the data section
    skiprows = 0
    try:
        import fsspec

        with fsspec.open(fname, "r", **(storage_options or {})) as f:
            for i, line in enumerate(f):
                if line.strip() == "Data":
                    skiprows = i + 1
                    break
    except Exception:
        pass

    # Read the CSV
    read_kwargs = kwargs.copy()
    for k in ["pivot", "add_meta", "as_xarray", "lazy", "expand2d", "storage_options"]:
        read_kwargs.pop(k, None)

    df = pd.read_csv(
        fname,
        delimiter=delimiter,
        parse_dates=["Date"],
        dtype={"EPACode": str},
        skiprows=skiprows,
        storage_options=storage_options,
        **read_kwargs,
    )

    # Standardize columns
    df = df.rename(
        columns={
            "EPACode": "epaid",
            "Val": "obs",
            "State": "state_name",
            "ParamCode": "variable",
            "SiteCode": "siteid",
            "Unit": "units",
            "Date": "time",
        }
    )

    if "Dataset" in df.columns:
        df = df.drop(columns="Dataset")

    df.columns = [i.lower() for i in df.columns]

    if "epaid" in df.columns:
        df["epaid"] = df["epaid"].astype(str).str.zfill(9)

    return force_object_strings(df)