stats_can.sc

Functionality that extends on what the base StatsCan api returns in some way.

Todo

Function to delete tables

Extend getChangedCubeList with a function that returns all tables updated within a date range

get_tables_for_vectors(vectors: str | list[str]) -> dict[str | int, str | list[str]]

Get a list of dicts mapping vectors to tables.

Parameters:

Name	Type	Description	Default
vectors	str | list[str]	Vectors to find tables for	required

Returns:

Type	Description
dict[str | int, str | list[str]]	keys for each vector number return the table, plus a key for 'all_tables' that has a list of unique tables used by vectors

Source code in src/stats_can/sc.py

def get_tables_for_vectors(
    vectors: str | list[str],
) -> dict[str | int, str | list[str]]:
    """Get a list of dicts mapping vectors to tables.

    Parameters
    ----------
    vectors
        Vectors to find tables for

    Returns
    -------
    :
        keys for each vector number return the table, plus a key for
        'all_tables' that has a list of unique tables used by vectors
    """
    v_json = get_series_info_from_vector(vectors)
    clean_vectors = [j["vectorId"] for j in v_json]
    tables_list: dict[int | str, str | list[str]] = {
        j["vectorId"]: str(j["productId"]) for j in v_json
    }
    all_tables = {tables_list[vector] for vector in clean_vectors}
    tables_list["all_tables"] = list(all_tables)
    return tables_list

table_subsets_from_vectors(vectors: str | list[str]) -> dict[str, list[str]]

Get a list of dicts mapping tables to vectors.

Parameters:

Name	Type	Description	Default
vectors	str | list[str]	Vectors to find tables for	required

Returns:

Type	Description
dict[str, list[str]]	keys for each table used by the vectors, matched to a list of vectors

Source code in src/stats_can/sc.py

def table_subsets_from_vectors(vectors: str | list[str]) -> dict[str, list[str]]:
    """Get a list of dicts mapping tables to vectors.

    Parameters
    ----------
    vectors
        Vectors to find tables for

    Returns
    -------
    :
        keys for each table used by the vectors, matched to a list of vectors
    """
    start_tables_dict = get_tables_for_vectors(vectors)
    tables_dict = {t: [] for t in start_tables_dict["all_tables"]}
    vecs = list(start_tables_dict.keys())[:-1]  # all but the all_tables key
    for vec in vecs:
        tables_dict[start_tables_dict[vec]].append(vec)
    return tables_dict

download_tables(tables: str | list[str], path: pathlib.Path | None = None, csv: bool = True) -> list[int]

Download a json file and zip of data for a list of tables to path.

Parameters:

Name	Type	Description	Default
tables	str | list[str]	tables to be downloaded	required
path	Path | None	Where to download the table and json	None
csv	bool	download in CSV format, if not download SDMX	True

Returns:

Type	Description
list[int]	list of tables that were downloaded

Source code in src/stats_can/sc.py

def download_tables(
    tables: str | list[str], path: pathlib.Path | None = None, csv: bool = True
) -> list[int]:
    """Download a json file and zip of data for a list of tables to path.

    Parameters
    ----------
    tables
        tables to be downloaded
    path
        Where to download the table and json
    csv
        download in CSV format, if not download SDMX

    Returns
    -------
    :
        list of tables that were downloaded
    """
    dl_path = pathlib.Path(path) if path else pathlib.Path()
    metas = get_cube_metadata(tables)
    for meta in metas:
        product_id = meta["productId"]
        zip_url = get_full_table_download(product_id, csv=csv)
        zip_file_name = f"{product_id}{'-eng' if csv else ''}.zip"
        json_file_name = f"{product_id}.json"
        zip_file = dl_path / zip_file_name
        json_file = dl_path / json_file_name

        # Thanks http://evanhahn.com/python-requests-library-useragent/
        response = _session.get(zip_url, stream=True, timeout=120)
        response.raise_for_status()

        progress_bar = tqdm(
            desc=zip_file_name,
            total=int(response.headers.get("content-length", 0)),
            unit="B",
            unit_scale=True,
        )

        # Thanks https://bit.ly/2sPYPYw
        with open(json_file, "w") as outfile:
            json.dump(meta, outfile)
        with open(zip_file, "wb") as handle:
            for chunk in response.iter_content(chunk_size=512):
                if chunk:  # filter out keep-alive new chunks
                    handle.write(chunk)
                    progress_bar.update(len(chunk))
        progress_bar.close()
    return [meta["productId"] for meta in metas]

zip_update_tables(path: pathlib.Path | None = None, csv: bool = True) -> list[str]

Check local json, update zips of outdated tables.

Grabs the json files in path, checks them against the metadata on StatsCan and grabs updated tables where there have been changes There isn't actually a "last modified date" part to the metadata What I'm doing is comparing the latest reference period. Almost all data changes will at least include incremental releases, so this should capture what I want

Parameters:

Name	Type	Description	Default
path	Path | None	where to look for tables to update	None
csv	bool	Downloads updates in CSV form by default, SDMX if false	True

Returns:

Type	Description
list[str]	list of the tables that were updated

Source code in src/stats_can/sc.py

def zip_update_tables(path: pathlib.Path | None = None, csv: bool = True) -> list[str]:
    """Check local json, update zips of outdated tables.

    Grabs the json files in path, checks them against the metadata on
    StatsCan and grabs updated tables where there have been changes
    There isn't actually a "last modified date" part to the metadata
    What I'm doing is comparing the latest reference period. Almost all
    data changes will at least include incremental releases, so this should
    capture what I want

    Parameters
    ----------
    path
        where to look for tables to update
    csv
        Downloads updates in CSV form by default, SDMX if false

    Returns
    -------
    :
        list of the tables that were updated

    """
    local_jsons = list_zipped_tables(path=path)
    tables = [j["productId"] for j in local_jsons]
    remote_jsons = get_cube_metadata(tables)
    update_table_list = [
        local["productId"]
        for local, remote in zip(local_jsons, remote_jsons)
        if local["cubeEndDate"] != remote["cubeEndDate"]
    ]

    download_tables(update_table_list, path, csv=csv)
    return update_table_list

zip_table_to_dataframe(table: str, path: pathlib.Path | None = None) -> pd.DataFrame

Read a StatsCan table into a pandas DataFrame.

If a zip file of the table does not exist in path, downloads it

Parameters:

Name	Type	Description	Default
table	str	the table to load to dataframe from zipped csv	required
path	Path | None	where to download the tables or load them, default will go to current working directory	None

Returns:

Type	Description
DataFrame	the table as a dataframe

Source code in src/stats_can/sc.py

def zip_table_to_dataframe(
    table: str, path: pathlib.Path | None = None
) -> pd.DataFrame:
    """Read a StatsCan table into a pandas DataFrame.

    If a zip file of the table does not exist in path, downloads it

    Parameters
    ----------
    table
        the table to load to dataframe from zipped csv
    path
        where to download the tables or load them, default will go to current working directory

    Returns
    -------
    :
        the table as a dataframe
    """
    path = pathlib.Path(path) if path else pathlib.Path()
    # Parse tables returns a list, can only do one table at a time here though
    table = parse_tables(table)[0]
    table_zip = table + "-eng.zip"
    table_zip = path / table_zip
    if not table_zip.is_file():
        download_tables([table], path)
    csv_file = table + ".csv"
    with zipfile.ZipFile(table_zip) as myzip:
        with myzip.open(csv_file) as myfile:
            col_names = pd.read_csv(myfile, nrows=0).columns
        # reopen the file or it misses the first row
        with myzip.open(csv_file) as myfile:
            types_dict = {"VALUE": float}
            types_dict.update({col: str for col in col_names if col not in types_dict})
            df = pd.read_csv(myfile, dtype=types_dict)

    possible_cats = [
        "GEO",
        "DGUID",
        "STATUS",
        "SYMBOL",
        "TERMINATED",
        "DECIMALS",
        "UOM",
        "UOM_ID",
        "SCALAR_FACTOR",
        "SCALAR_ID",
        "VECTOR",
        "COORDINATE",
        "Wages",
        "National Occupational Classification for Statistics (NOC-S)",
        "Supplementary unemployment rates",
        "Sex",
        "Age group",
        "Labour force characteristics",
        "Statistics",
        "Data type",
        "Job permanency",
        "Union coverage",
        "Educational attainment",
    ]
    actual_cats = [col for col in possible_cats if col in col_names]
    df[actual_cats] = df[actual_cats].astype("category")
    df["REF_DATE"] = pd.to_datetime(df["REF_DATE"], format="%Y-%m-%d", errors="coerce")
    return df

list_zipped_tables(path: pathlib.Path | None = None) -> list[str]

List StatsCan tables available.

defaults to looking in the current working directory and for zipped CSVs

Parameters:

Name	Type	Description	Default
path	Path | None	Where to look for zipped tables, defaults to current working directory	None

Returns:

Type	Description
list[str]	list of available tables json data

Source code in src/stats_can/sc.py

def list_zipped_tables(path: pathlib.Path | None = None) -> list[str]:
    """List StatsCan tables available.

    defaults to looking in the current working directory and for zipped CSVs

    Parameters
    ----------
    path
        Where to look for zipped tables, defaults to current working directory

    Returns
    -------
    :
        list of available tables json data
    """
    # Find json files
    path = pathlib.Path(path) if path else pathlib.Path.cwd()
    jsons = path.glob("*.json")
    tables = []
    for j in jsons:
        try:
            with open(j) as json_file:
                result = json.load(json_file)
                if "productId" in result:
                    tables.append(result)
        except ValueError:
            logger.warning("failed to read json file %s", j)
    return tables

vectors_to_df(vectors: str | list[str], periods: int = 1, start_release_date: dt.date | None = None, end_release_date: dt.date | None = None) -> pd.DataFrame

Get DataFrame of vectors with n periods data or over range of release dates.

Wrapper on get_bulk_vector_data_by_range and get_data_from_vectors_and_latest_n_periods function to turn the resulting list of JSONs into a DataFrame

Parameters:

Name	Type	Description	Default
vectors	str | list[str]	vector numbers to get info for	required
periods	int	number of periods to retrieve data for	1
start_release_date	date | None	start release date for the data	None
end_release_date	date | None	end release date for the data	None

Returns:

Type	Description
DataFrame	vectors as columns and ref_date as the index (not release date)

Source code in src/stats_can/sc.py

def vectors_to_df(
    vectors: str | list[str],
    periods: int = 1,
    start_release_date: dt.date | None = None,
    end_release_date: dt.date | None = None,
) -> pd.DataFrame:
    """Get DataFrame of vectors with n periods data or over range of release dates.

    Wrapper on get_bulk_vector_data_by_range and
    get_data_from_vectors_and_latest_n_periods function to turn the resulting
    list of JSONs into a DataFrame

    Parameters
    ----------
    vectors
        vector numbers to get info for
    periods
        number of periods to retrieve data for
    start_release_date
        start release date for the data
    end_release_date
        end release date for the data

    Returns
    -------
    :
        vectors as columns and ref_date as the index (not release date)
    """
    df = pd.DataFrame()
    if (end_release_date is None) | (start_release_date is None):
        start_list = get_data_from_vectors_and_latest_n_periods(vectors, periods)
    else:
        start_list = get_bulk_vector_data_by_range(
            vectors, start_release_date, end_release_date
        )
    for vec in start_list:
        name = "v" + str(vec["vectorId"])
        # If there's no data for the series just skip it
        if not vec["vectorDataPoint"]:
            continue
        ser = (
            pd.DataFrame(vec["vectorDataPoint"])
            .assign(
                refPer=lambda x: pd.to_datetime(
                    x["refPer"], format="%Y-%m-%d", errors="coerce"
                )
            )
            .set_index("refPer")
            .rename(columns={"value": name})
            .filter([name])
        )
        df = pd.concat([df, ser], axis=1, sort=True)
    return df

code_sets_to_df_dict() -> dict[str, pd.DataFrame]

Get all code sets.

Code sets provide additional metadata to describe information. Code sets are grouped into scales, frequencies, symbols etc. and returned as dictionary of dataframes.

Returns:

Type	Description
dict[str, DataFrame]	dictionary of dataframes

Source code in src/stats_can/sc.py

def code_sets_to_df_dict() -> dict[str, pd.DataFrame]:
    """Get all code sets.

    Code sets provide additional metadata to describe
    information. Code sets are grouped into scales, frequencies, symbols etc.
    and returned as dictionary of dataframes.

    Returns
    -------
    :
        dictionary of dataframes
    """
    codes = get_code_sets()
    # Packs each code group in a dataframe for better lookup via dictionary
    codes_df_lookup = {key: pd.DataFrame(codes[key]) for key in codes.keys()}
    return codes_df_lookup