readabs.read_abs_cat

  1"""Download *timeseries* data from the Australian Bureau of Statistics.
  2
  3Download timeseries data from the Australian Bureau of Statistics (ABS)
  4for a specified ABS catalogue identifier.
  5"""
  6
  7import calendar
  8from functools import cache
  9from typing import Any, Unpack
 10
 11import pandas as pd
 12from pandas import DataFrame
 13
 14from readabs.abs_meta_data import metacol
 15from readabs.grab_abs_url import grab_abs_url, grab_abs_zip
 16from readabs.read_support import HYPHEN, ReadArgs
 17
 18# Constants
 19MAX_DATETIME_CHARS = 20
 20TABLE_DESC_ROW = 4
 21TABLE_DESC_COL = 1
 22
 23
 24# --- functions ---
 25# - public -
 26@cache  # minimise slowness for any repeat business
 27def read_abs_cat(
 28    cat: str,
 29    **kwargs: Unpack[ReadArgs],
 30) -> tuple[dict[str, DataFrame], DataFrame]:
 31    """For a specific catalogue identifier, return the complete ABS Catalogue information as DataFrames.
 32
 33    This function returns the complete ABS Catalogue information as a
 34    python dictionary of pandas DataFrames, as well as the associated metadata
 35    in a separate DataFrame. The function automates the collection of zip and
 36    excel files from the ABS website. If necessary, these files are downloaded,
 37    and saved into a cache directory. The files are then parsed to extract time
 38    series data, and the associated metadata.
 39
 40    By default, the cache directory is `./.readabs_cache/`. You can change the
 41    default directory name by setting the shell environment variable
 42    `READABS_CACHE_DIR` with the name of the preferred directory.
 43
 44    Parameters
 45    ----------
 46    cat : str
 47        The ABS Catalogue Number for the data to be downloaded and made
 48        available by this function. This argument must be specified in the
 49        function call.
 50
 51    **kwargs : Unpack[ReadArgs]
 52        The following parameters may be passed as optional keyword arguments.
 53
 54    keep_non_ts : bool = False
 55        A flag for whether to keep the non-time-series tables
 56        that might form part of an ABS catalogue item. Normally, the
 57        non-time-series information is ignored, and not made available to
 58        the user.
 59
 60    history : str = ""
 61        Provide a month-year string to extract historical ABS data.
 62        For example, you can set history="dec-2023" to the get the ABS data
 63        for a catalogue identifier that was originally published in respect
 64        of Q4 of 2023. Note: not all ABS data sources are structured so that
 65        this technique works in every case; but most are.
 66
 67    verbose : bool = False
 68        Setting this to true may help diagnose why something
 69        might be going wrong with the data retrieval process.
 70
 71    ignore_errors : bool = False
 72        Normally, this function will cease downloading when
 73        an error in encountered. However, sometimes the ABS website has
 74        malformed links, and changing this setting is necessitated. (Note:
 75        if you drop a message to the ABS, they will usually fix broken
 76        links with a business day).
 77
 78    get_zip : bool = True
 79        Download the excel files in .zip files.
 80
 81    get_excel_if_no_zip : bool = True
 82        Only try to download .xlsx files if there are no zip
 83        files available to be downloaded. Only downloading individual excel
 84        files when there are no zip files to download can speed up the
 85        download process.
 86
 87    get_excel : bool = False
 88        The default value means that excel files are not
 89        automatically download. Note: at least one of `get_zip`,
 90        `get_excel_if_no_zip`, or `get_excel` must be true. For most ABS
 91        catalogue items, it is sufficient to just download the one zip
 92        file. But note, some catalogue items do not have a zip file.
 93        Others have quite a number of zip files.
 94
 95    single_excel_only : str = ""
 96        If this argument is set to a table name (without the
 97        .xlsx extension), only that excel file will be downloaded. If
 98        set, and only a limited subset of available data is needed,
 99        this can speed up download times significantly. Note: overrides
100        `get_zip`, `get_excel_if_no_zip`, `get_excel` and `single_zip_only`.
101
102    single_zip_only : str = ""
103        If this argument is set to a zip file name (without
104        the .zip extension), only that zip file will be downloaded.
105        If set, and only a limited subset of available data is needed,
106        this can speed up download times significantly. Note: overrides
107        `get_zip`, `get_excel_if_no_zip`, and `get_excel`.
108
109    cache_only : bool = False
110        If set to True, this function will only access
111        data that has been previously cached. Normally, the function
112        checks the date of the cache data against the date of the data
113        on the ABS website, before deciding whether the ABS has fresher
114        data that needs to be downloaded to the cache.
115
116    zip_file: str | Path = ""
117        If set to a specific zip file name (with or without the .zip
118        extension), this function will only extract data from that zip file
119        on the local file system. This may be useful for debugging purposes.
120
121    Returns
122    -------
123    tuple[dict[str, DataFrame], DataFrame]
124        The function returns a tuple of two items. The first item is a
125        python dictionary of pandas DataFrames (which is the primary data
126        associated with the ABS catalogue item). The second item is a
127        DataFrame of ABS metadata for the ABS collection.
128
129        Note:
130        You can retrieve non-timeseries data using the grab_abs_url()
131        function. That takes the URL for the ABS landing page for the ABS
132        collection you are interested in. The read_abs_cat function is for
133        ABS catalogue identifiers which are timeseries data, for which the
134        metadata can be extracted.
135
136    Example
137    -------
138
139    ```python
140    import readabs as ra
141    from pandas import DataFrame
142    cat_num = "6202.0"  # The ABS labour force survey
143    data: tuple[dict[str, DataFrame], DataFrame] = ra.read_abs_cat(cat=cat_num)
144    abs_dict, meta = data
145    ```
146
147    """
148    # --- get the time series data ---
149    if "zip_file" in kwargs and kwargs["zip_file"]:
150        raw_abs_dict = grab_abs_zip(kwargs["zip_file"], **kwargs)
151    else:
152        raw_abs_dict = grab_abs_url(cat=cat, **kwargs)
153    response = _get_time_series_data(cat, raw_abs_dict, **kwargs)
154
155    if not response:
156        response = {}, DataFrame()
157
158    return response  # dictionary of DataFrames, and a DataFrame of metadata
159
160
161# - private -
162def _get_time_series_data(
163    cat: str,
164    abs_dict: dict[str, DataFrame],
165    **kwargs: Any,  # keep_non_ts, verbose, ignore_errors
166) -> tuple[dict[str, DataFrame], DataFrame]:
167    """Extract the time series data for a specific ABS catalogue identifier."""
168    # --- set up ---
169    cat = "<catalogue number missing>" if not cat.strip() else cat.strip()
170    new_dict: dict[str, DataFrame] = {}
171    meta_data = DataFrame()
172
173    # --- group the sheets and iterate over these groups
174    long_groups = _group_sheets(abs_dict)
175    for table, sheets in long_groups.items():
176        args = {
177            "cat": cat,
178            "from_dict": abs_dict,
179            "table": table,
180            "long_sheets": sheets,
181        }
182        new_dict, meta_data = _capture(new_dict, meta_data, args, **kwargs)
183    return new_dict, meta_data
184
185
186def _copy_raw_sheets(
187    from_dict: dict[str, DataFrame],
188    long_sheets: list[str],
189    to_dict: dict[str, DataFrame],
190    *,
191    keep_non_ts: bool,
192) -> dict[str, DataFrame]:
193    """Copy the raw sheets across to the final dictionary.
194
195    Used if the data is not in a timeseries format, and keep_non_ts
196    flag is set to True. Returns an updated final dictionary.
197    """
198    if not keep_non_ts:
199        return to_dict
200
201    for sheet in long_sheets:
202        if sheet in from_dict:
203            to_dict[sheet] = from_dict[sheet]
204        else:
205            # should not happen
206            raise ValueError(f"Glitch: Sheet {sheet} not found in the data.")
207    return to_dict
208
209
210def _capture(
211    to_dict: dict[str, DataFrame],
212    meta_data: DataFrame,
213    args: dict[str, Any],
214    **kwargs: Any,  # keep_non_ts, ignore_errors
215) -> tuple[dict[str, DataFrame], DataFrame]:
216    """Capture the time series data and meta data from an Excel file.
217
218    For a specific Excel file, capture *both* the time series data
219    from the ABS data files as well as the meta data. These data are
220    added to the input 'to_dict' and 'meta_data' respectively, and
221    the combined results are returned as a tuple.
222    """
223    # --- step 0: set up ---
224    keep_non_ts: bool = kwargs.get("keep_non_ts", False)
225    ignore_errors: bool = kwargs.get("ignore_errors", False)
226
227    # --- step 1: capture the meta data ---
228    short_names = [x.split(HYPHEN, 1)[1] for x in args["long_sheets"]]
229    if "Index" not in short_names:
230        print(f"Table {args['table']} has no 'Index' sheet.")
231        to_dict = _copy_raw_sheets(args["from_dict"], args["long_sheets"], to_dict, keep_non_ts=keep_non_ts)
232        return to_dict, meta_data
233    index = short_names.index("Index")
234
235    index_sheet = args["long_sheets"][index]
236    this_meta = _capture_meta(args["cat"], args["from_dict"], index_sheet)
237    if this_meta.empty:
238        to_dict = _copy_raw_sheets(args["from_dict"], args["long_sheets"], to_dict, keep_non_ts=keep_non_ts)
239        return to_dict, meta_data
240
241    meta_data = pd.concat([meta_data, this_meta], axis=0)
242
243    # --- step 2: capture the actual time series data ---
244    data = _capture_data(meta_data, args["from_dict"], args["long_sheets"], **kwargs)
245    if len(data):
246        to_dict[args["table"]] = data
247    else:
248        # a glitch: we have the metadata but not the actual data
249        error = f"Unexpected: {args['table']} has no actual data."
250        if not ignore_errors:
251            raise ValueError(error)
252        print(error)
253        to_dict = _copy_raw_sheets(args["from_dict"], args["long_sheets"], to_dict, keep_non_ts=keep_non_ts)
254
255    return to_dict, meta_data
256
257
258def _capture_data(
259    abs_meta: DataFrame,
260    from_dict: dict[str, DataFrame],
261    long_sheets: list[str],
262    **kwargs: Any,  # verbose
263) -> DataFrame:
264    """Take a list of ABS data sheets and stitch them into a DataFrame.
265
266    Find the DataFrames for those sheets in the from_dict, and stitch them
267    into a single DataFrame with an appropriate PeriodIndex.
268    """
269    # --- step 0: set up ---
270    verbose: bool = kwargs.get("verbose", False)
271    merged_data = DataFrame()
272    header_row: int = 8
273
274    # --- step 1: capture the time series data ---
275    # identify the data sheets in the list of all sheets from Excel file
276    data_sheets = [x for x in long_sheets if x.split(HYPHEN, 1)[1].startswith("Data")]
277
278    for sheet_name in data_sheets:
279        if verbose:
280            print(f"About to cature data from {sheet_name=}")
281
282        # --- capture just the data, nothing else
283        sheet_data = from_dict[sheet_name].copy()
284
285        # get the columns
286        header = sheet_data.iloc[header_row]
287        sheet_data.columns = pd.Index(header)
288        sheet_data = sheet_data[(header_row + 1) :]
289
290        # get the row indexes
291        sheet_data = _index_to_period(sheet_data, sheet_name, abs_meta, verbose=verbose)
292
293        # --- merge data into a single dataframe
294        if len(merged_data) == 0:
295            merged_data = sheet_data
296        else:
297            merged_data = merged_data.merge(
298                right=sheet_data,
299                how="outer",
300                left_index=True,
301                right_index=True,
302                suffixes=("", ""),
303            )
304
305    # --- step 2 - final tidy-ups
306    # remove NA rows
307    merged_data = merged_data.dropna(how="all")
308    # check for NA columns - rarely happens
309    # Note: these empty columns are not removed,
310    # but it is useful to know they are there
311    if merged_data.isna().all().any() and verbose:
312        cols = merged_data.columns[merged_data.isna().all()]
313        print("Caution: All columns are NA")
314
315    # check for duplicate columns - should not happen
316    # Note: these duplicate columns are removed
317    duplicates = merged_data.columns.duplicated()
318    if duplicates.any():
319        if verbose:
320            dup_table = abs_meta[metacol.table].iloc[0]
321            print(f"Note: duplicates removed from {dup_table}: " + f"{merged_data.columns[duplicates]}")
322        merged_data = merged_data.loc[:, ~duplicates].copy()
323
324    # make the data all floats.
325    return merged_data.astype(float).sort_index()
326
327
328def _index_to_period(sheet_data: DataFrame, sheet_name: str, abs_meta: DataFrame, *, verbose: bool) -> DataFrame:
329    """Convert the index of a DataFrame to a PeriodIndex."""
330    index_column = sheet_data[sheet_data.columns[0]].astype(str)
331    sheet_data = sheet_data.drop(sheet_data.columns[0], axis=1)
332    long_row_names = index_column.str.len() > MAX_DATETIME_CHARS  # 19 chars in datetime str
333    if verbose and long_row_names.any():
334        print(f"You may need to check index column for {sheet_name}")
335    index_column = index_column.loc[~long_row_names]
336    sheet_data = sheet_data.loc[~long_row_names]
337
338    proposed_index = pd.to_datetime(index_column)
339
340    # get the correct period index
341    short_name = sheet_name.split(HYPHEN, 1)[0]
342    series_id = sheet_data.columns[0]
343    freq_value = abs_meta[abs_meta[metacol.table] == short_name].loc[series_id, metacol.freq]
344    freq = str(freq_value).upper().strip()[0]
345    freq = "Y" if freq == "A" else freq  # pandas prefers yearly
346    freq = "Q" if freq == "B" else freq  # treat Biannual as quarterly
347    if freq not in ("Y", "Q", "M", "D"):
348        print(f"Check the frequency of the data in sheet: {sheet_name}")
349
350    # create an appropriate period index
351    if freq:
352        if freq in ("Q", "Y"):
353            month = str(calendar.month_abbr[proposed_index.dt.month.max()]).upper()
354            freq = f"{freq}-{month}"
355        sheet_data.index = pd.PeriodIndex(proposed_index, freq=freq)
356    else:
357        raise ValueError(f"With sheet {sheet_name} could not determime PeriodIndex")
358
359    return sheet_data
360
361
362def _capture_meta(
363    cat: str,
364    from_dict: dict[str, DataFrame],
365    index_sheet: str,
366) -> DataFrame:
367    """Capture the metadata from the Index sheet of an ABS excel file.
368
369    Returns a DataFrame specific to the current excel file.
370    Returning an empty DataFrame, means that the meta data could not
371    be identified. Meta data for each ABS data item is organised by row.
372    """
373    # --- step 0: set up ---
374    frame = from_dict[index_sheet]
375
376    # --- step 1: check if the metadata is present in the right place ---
377    # Unfortunately, the header for some of the 3401.0
378    #                spreadsheets starts on row 10
379    starting_rows = 8, 9, 10
380    required = metacol.did, metacol.id, metacol.stype, metacol.unit
381    required_set = set(required)
382
383    header_row = None
384    header_columns = None
385    for row in starting_rows:
386        columns = frame.iloc[row]
387        if required_set.issubset(set(columns)):
388            header_row = row
389            header_columns = columns
390            break
391
392    if header_row is None or header_columns is None:
393        print(f"Table has no metadata in sheet {index_sheet}.")
394        return DataFrame()
395
396    # --- step 2: capture the metadata ---
397    file_meta = frame.iloc[header_row + 1 :].copy()
398    file_meta.columns = pd.Index(header_columns)
399
400    # make damn sure there are no rogue white spaces
401    for col in required:
402        file_meta[col] = file_meta[col].str.strip()
403
404    # remove empty columns and rows
405    file_meta = file_meta.dropna(how="all", axis=1).dropna(how="all", axis=0)
406
407    # populate the metadata
408    file_meta[metacol.table] = index_sheet.split(HYPHEN, 1)[0]
409    tab_desc_value = frame.iloc[TABLE_DESC_ROW, TABLE_DESC_COL]
410    tab_desc = str(tab_desc_value).split(".", 1)[-1].strip()
411    file_meta[metacol.tdesc] = tab_desc
412    file_meta[metacol.cat] = cat
413
414    # drop last row - should just be copyright statement
415    file_meta = file_meta.iloc[:-1]
416
417    # set the index to the series_id
418    file_meta.index = pd.Index(file_meta[metacol.id])
419
420    return file_meta
421
422
423def _group_sheets(
424    abs_dict: dict[str, DataFrame],
425) -> dict[str, list[str]]:
426    """Group the sheets from an Excel file."""
427    keys = list(abs_dict.keys())
428    long_pairs = [(x.split(HYPHEN, 1)[0], x) for x in keys]
429
430    def group(p_list: list[tuple[str, str]]) -> dict[str, list[str]]:
431        groups: dict[str, list[str]] = {}
432        for x, y in p_list:
433            if x not in groups:
434                groups[x] = []
435            groups[x].append(y)
436        return groups
437
438    return group(long_pairs)
439
440
441# --- initial testing ---
442if __name__ == "__main__":
443
444    def simple_test() -> None:
445        """Test the read_abs_cat function."""
446        # ABS Catalogue ID 8731.0 has a mix of time
447        # series and non-time series data. Also,
448        # it has unusually structured Excel files. So, a good test.
449
450        print("Starting test.")
451
452        d, _m = read_abs_cat("8731.0", keep_non_ts=False, verbose=False)
453        print(f"--- {len(d)=} ---")
454        print(f"--- {d.keys()=} ---")
455        for table in d:
456            freq_str = getattr(d[table].index, "freqstr", "Unknown")
457            print(f"{table=} {d[table].shape=} {freq_str=}")
458
459        print ("=" * 20)
460
461        d, _m = read_abs_cat("", zip_file=".test-data/Qrtly-CPI-Time-series-spreadsheets-all.zip", verbose=False)
462        print(f"--- {len(d)=} ---")
463        print(f"--- {d.keys()=} ---")
464        for table in d:
465            freq_str = getattr(d[table].index, "freqstr", "Unknown")
466            print(f"{table=} {d[table].shape=} {freq_str=}")
467
468        print("Test complete.")
469
470    simple_test()