diff --git a/src/isp_workbook_parser/config_model.py b/src/isp_workbook_parser/config_model.py index 625ad68..31456d8 100644 --- a/src/isp_workbook_parser/config_model.py +++ b/src/isp_workbook_parser/config_model.py @@ -28,7 +28,6 @@ class TableConfig(BaseModel): The `Pydantic` class verifies the type of each element of the configuration. Examples: - A TableConfig instance can be manually defined: >>> table_config = TableConfig( @@ -70,6 +69,7 @@ class TableConfig(BaseModel): 'missed_column_on_right_hand_side', 'missed_column_on_left_hand_side', 'last_column_isnt_empty', 'over_run_into_another_table', 'over_run_into_notes'. + """ name: str @@ -84,13 +84,12 @@ class TableConfig(BaseModel): def load_yaml(path: Path) -> dict[str, TableConfig]: - """Loads the YAML file specified by the path returning a dict of `TableConfig`s. + """Load the YAML file specified by the path returning a dict of `TableConfig`s. Each table config defined in a YAML file is converted to a `TableConfig` and stored in the dictionary using its name as the key value. Examples: - >>> path_to_yaml = Path("src/isp_table_configs/6.0/capacity_factors.yaml") The contents of the YAML file should look like: @@ -144,7 +143,7 @@ def load_yaml(path: Path) -> dict[str, TableConfig]: path: pathlib Path instance specifying the location of the YAML file. """ - with open(path, "r") as f: + with Path.open(path) as f: config = yaml.safe_load(f) f.close() if config is not None: diff --git a/src/isp_workbook_parser/parser.py b/src/isp_workbook_parser/parser.py index f0047f5..9abbdcb 100644 --- a/src/isp_workbook_parser/parser.py +++ b/src/isp_workbook_parser/parser.py @@ -5,8 +5,6 @@ # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. -import glob -import os import warnings from pathlib import Path from typing import Any @@ -42,7 +40,6 @@ class Parser: 3. Extract tables using a user-specified config with `Parser.get_table_from_config`. Examples: - Create a Parser instance for a particular workbook. Will also check config is available for workbook version. >>> workbook = Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") # doctest: +SKIP @@ -50,10 +47,13 @@ class Parser: Save all the tables with available config to the directory example_output as csv files. >>> workbook.save_tables('example_output') # doctest: +SKIP + """ def __init__( - self, file_path: str | Path, user_config_directory_path: str | Path = None + self, + file_path: str | Path, + user_config_directory_path: str | Path | None = None, ) -> None: self.file_path = self._make_path_object(file_path) @@ -96,7 +96,7 @@ def _get_version(self) -> str: def _determine_config_path( self, - user_config_directory_path: str | Path = None, + user_config_directory_path: str | Path | None = None, ) -> Path: """Determine the path to where the directory containing config YAML files are stored. @@ -112,25 +112,23 @@ def _determine_config_path( config_path = config_path / Path(f"{self.workbook_version}/") return config_path - def _check_version_is_supported(self, config_path) -> None: + def _check_version_is_supported(self, config_path: Path) -> None: """Check the default config directory contains a subdirectory that matches the workbook version number.""" - versions = os.listdir(config_path) + versions = [dir.name for dir in config_path.iterdir()] if self.workbook_version not in versions: - raise ValueError( - f"The workbook version {self.workbook_version} is not supported." - ) + msg = f"The workbook version {self.workbook_version} is not supported." + raise ValueError(msg) def _load_config(self) -> dict[str, dict[str, Any]]: """Load all the YAML files stored in the config directory into a nested dictionary with sheet names as keys and table names as second level keys. For robustness across workbook versions, the config sheet name is matched with a workbook sheet name in case-agnostic manner. """ - pattern = os.path.join(self.config_path, "*.yaml") - config_files = glob.glob(pattern) + config_files = Path(self.config_path).glob("*.yaml") configs = {} for file in config_files: config_dict = load_yaml(Path(file)) - for config_name in config_dict.keys(): + for config_name in config_dict: config = config_dict[config_name] config_sheet_name_lowercase = config.sheet_name.lower() sheet_names = [ @@ -139,20 +137,19 @@ def _load_config(self) -> dict[str, dict[str, Any]]: if sheet_name.lower() == config_sheet_name_lowercase ] if len(sheet_names) > 1: - raise TableConfigError( - f"Workbook sheet '{config.sheet_name}' is not unique" - ) - elif len(sheet_names) < 1: - raise TableConfigError( + msg = f"Workbook sheet '{config.sheet_name}' is not unique" + raise TableConfigError(msg) + if len(sheet_names) < 1: + msg = ( f" Sheet '{config.sheet_name}' cannot be found in the workbook" ) - else: - config.sheet_name = sheet_names.pop() + raise TableConfigError(msg) + config.sheet_name = sheet_names.pop() config_dict[config_name] = config configs.update(config_dict) return configs - def _get_table_names_by_sheet(self): + def _get_table_names_by_sheet(self) -> dict: table_names_by_sheet = {} for table_name, config in self.table_configs.items(): if config.sheet_name not in table_names_by_sheet: @@ -191,7 +188,7 @@ def _check_data_ends_where_expected( raise TableConfigError(error_message) def _check_no_data_above_first_header_row( - self, tab: str, header_rows: int, range: str, name: str + self, tab: str, header_rows: int, cellrange: str, name: str ) -> None: """Check that the cell before the first header row of the table in the second column is blank. @@ -199,7 +196,7 @@ def _check_no_data_above_first_header_row( second column appears to be always blank. Therefore, checking that this cell is blank can be used to verify that the config has not specified a table header row that is after the first header row of the table. """ - first_column = range.split(":")[0] + first_column = cellrange.split(":", maxsplit=1)[0] first_col_index = openpyxl.utils.column_index_from_string(first_column) second_col_index = first_col_index + 1 @@ -268,7 +265,7 @@ def _check_last_column_isnt_empty(data: pd.DataFrame, name: str) -> None: @staticmethod def _check_columns_unique(data: pd.DataFrame, name: str) -> None: - """Check that columns in the data are unique + """Check that columns in the data are unique. Unique columns names are required for sanitisation to work without error (i.e. in `isp_workbook_parser.sanitisers._values_casting_and_sanitisation`). If an error @@ -280,15 +277,15 @@ def _check_columns_unique(data: pd.DataFrame, name: str) -> None: raise TableConfigError(error_message) def _check_for_missed_column_on_right_hand_side_of_table( - self, sheet_name: str, start_row: int, end_row: int, range: str, name: str + self, sheet_name: str, start_row: int, end_row: int, cellrange: str, name: str ) -> None: - """Checks if there is data in the column adjacent to last column specified in the config. + """Checkg if there is data in the column adjacent to last column specified in the config. It appears that the column adjacent to the last column in a table is always blank. Therefore, checking if there is data in the adjacent column can help detect when the column range in the config has been incorrectly specified. """ - last_column = range.split(":")[1] + last_column = cellrange.split(":")[1] last_col_index = openpyxl.utils.column_index_from_string(last_column) column_next_to_last_column = openpyxl.utils.get_column_letter( last_col_index + 1 @@ -320,15 +317,15 @@ def _check_for_missed_column_on_right_hand_side_of_table( raise TableConfigError(error_message) def _check_for_missed_column_on_left_hand_side_of_table( - self, sheet_name: str, start_row: int, end_row: int, range: str, name: str + self, sheet_name: str, start_row: int, end_row: int, cellrange: str, name: str ) -> None: - """Checks if there is data in the column adjacent to first column specified in the config. + """Check if there is data in the column adjacent to first column specified in the config. It appears that the column adjacent to the first column in a table is always blank. Therefore, checking if there is data in the adjacent column can help detect when the column range in the config has been incorrectly specified. """ - first_column = range.split(":")[0] + first_column = cellrange.split(":", maxsplit=1)[0] first_col_index = openpyxl.utils.column_index_from_string(first_column) column_next_to_first_column = openpyxl.utils.get_column_letter( first_col_index - 1 @@ -345,10 +342,9 @@ def _check_for_missed_column_on_left_hand_side_of_table( usecols=column_next_to_first_column, nrows=(end_row - start_row), ) - if data[data.columns[0]].isna().all(): - range_error = False - elif ( - "DO NOT DELETE THIS COLUMN" in str(data.columns[0]) + if ( + data[data.columns[0]].isna().all() + or "DO NOT DELETE THIS COLUMN" in str(data.columns[0]) or first_column == "B" ): range_error = False @@ -361,8 +357,10 @@ def _check_for_missed_column_on_left_hand_side_of_table( error_message = f"There is data in the column adjacent to the first column in the table {name}." raise TableConfigError(error_message) - def _check_if_header_row_and_end_row_are_on_sheet(self, table_config) -> None: - """Checks if first row of header and end_row are within the sheet.""" + def _check_if_header_row_and_end_row_are_on_sheet( + self, table_config: TableConfig + ) -> None: + """Check if first row of header and end_row are within the sheet.""" if isinstance(table_config.header_rows, int): first_header_row = table_config.header_rows else: @@ -375,8 +373,10 @@ def _check_if_header_row_and_end_row_are_on_sheet(self, table_config) -> None: error_message = f"The end_row for table {table_config.name} is not within the excel sheet." raise TableConfigError(error_message) - def _check_if_start_and_end_column_are_on_sheet(self, table_config) -> None: - """Checks if first column and last column in config are within the sheet.""" + def _check_if_start_and_end_column_are_on_sheet( + self, table_config: TableConfig + ) -> None: + """Check if first column and last column in config are within the sheet.""" first_column = table_config.column_range.split(":")[0] first_col_index = openpyxl.utils.column_index_from_string(first_column) if first_col_index > self.openpyxl_file[table_config.sheet_name].max_column: @@ -389,8 +389,8 @@ def _check_if_start_and_end_column_are_on_sheet(self, table_config) -> None: error_message = f"The last column for table {table_config.name} is not within the excel sheet." raise TableConfigError(error_message) - def _build_checks(self, data, table_config): - """Builds a dict mapping each skippable check name to its check method and arguments. + def _build_checks(self, data: pd.DataFrame, table_config: TableConfig) -> dict: + """Build a dict mapping each skippable check name to its check method and arguments. The keys must stay in sync with `CheckName` in `config_model` (enforced by `test_skippable_check_names_match_config_literal`). @@ -455,7 +455,7 @@ def _build_checks(self, data, table_config): ), } - def _check_table(self, data, table_config) -> None: + def _check_table(self, data: pd.DataFrame, table_config: TableConfig) -> None: checks = self._build_checks(data, table_config) skips = table_config.skip_checks or [] for check_name, (check, args) in checks.items(): @@ -481,6 +481,7 @@ def _postprocess_percentage_columns_between_0_and_100( Returns: `pandas.DataFrame` with percentage columns multiplied by 100 (i.e. values should be between 0 and 100) + """ percentage_columns = [] sheet = self.openpyxl_file[table_config.sheet_name] @@ -505,7 +506,7 @@ def _postprocess_percentage_columns_between_0_and_100( if isinstance(sr, list) and cell.row in sr: skipped_rows += 1 continue - elif isinstance(sr, int) and cell.row == sr: + if isinstance(sr, int) and cell.row == sr: skipped_rows += 1 continue if isinstance(cell.value, (int, float)) and "%" in cell.number_format: @@ -521,7 +522,7 @@ def _postprocess_percentage_columns_between_0_and_100( # add the data column index if the entire column consists of percentage values # else, add the individual cells as a list of tuples if len(percentage_cells) == (table_config.end_row - min_row + 1): - percentage_columns.append(set(x[1] for x in percentage_cells).pop()) + percentage_columns.append({x[1] for x in percentage_cells}.pop()) else: percentage_columns.append(percentage_cells) @@ -534,7 +535,7 @@ def _postprocess_percentage_columns_between_0_and_100( return data def get_table_names(self) -> list[str]: - """Returns a dict of table names by sheet name that there is config for. + """Return a dict of table names by sheet name that there is config for. Examples: >>> workbook = Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") @@ -547,16 +548,16 @@ def get_table_names(self) -> list[str]: Returns: List of the tables that there is configuration information for extracting from the workbook. + """ return self.table_names_by_sheet def get_table_from_config( self, table_config: TableConfig, config_checks: bool = True ) -> pd.DataFrame: - """Retrieves a table from the assumptions workbook using the config provided and returns as pd.DataFrame. + """Retrieve a table from the assumptions workbook using the config provided and returns as pd.DataFrame. Examples: - >>> import pandas as pd >>> from isp_workbook_parser import TableConfig @@ -603,9 +604,9 @@ def get_table_from_config( return data def get_table(self, table_name: str, config_checks: bool = True) -> pd.DataFrame: - """Retrieves a table from the assumptions workbook and returns as `pd.DataFrame`. + """Retrieve a table from the assumptions workbook and returns as `pd.DataFrame`. - Examples + Examples: >>> workbook = Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") >>> workbook.get_table('wind_high_capacity_factors').head() @@ -622,19 +623,21 @@ def get_table(self, table_name: str, config_checks: bool = True) -> pd.DataFrame table_name: Specified the table to retrieve. config_checks: Specifies whether to check the tabe config by checking if the data starts and ends where expected and the workbook header matches the config header. + """ if not isinstance(table_name, str): - raise ValueError("The parameter table_name must be provided as a string.") - if table_name not in self.table_configs.keys(): + msg = "The parameter table_name must be provided as a string." + raise TypeError(msg) + if table_name not in self.table_configs: closest = process.extractOne(table_name, self.table_configs.keys())[0] - raise ValueError( + msg = ( f"The table_name ({table_name}) provided is not in the config for this workbook version." - + f" Did you mean '{closest}'?" + f" Did you mean '{closest}'?" ) + raise ValueError(msg) table_config = self.table_configs[table_name] - data = self.get_table_from_config(table_config, config_checks=config_checks) - return data + return self.get_table_from_config(table_config, config_checks=config_checks) def save_tables( self, @@ -642,7 +645,7 @@ def save_tables( tables: list[str] | str = "all", config_checks: bool = True, ) -> None: - """Saves tables from the provided workbook to the specified directory as CSV files. + """Save tables from the provided workbook to the specified directory as CSV files. Examples: >>> workbook = Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") # doctest: +SKIP @@ -659,24 +662,26 @@ def save_tables( Returns: None + """ directory = self._make_path_object(directory) if not directory.exists(): directory.mkdir(parents=True) if not directory.is_dir(): - raise ValueError("The path provided is not a directory.") + msg = "The path provided is not a directory." + raise ValueError(msg) - if not (isinstance(tables, str) or isinstance(tables, list)): - raise ValueError( - "The parameter tables must be provided as str or list[str]." - ) + if not (isinstance(tables, (str, list))): + msg = "The parameter tables must be provided as str or list[str]." + raise TypeError(msg) if isinstance(tables, str) and tables != "all": - raise ValueError( + msg = ( "If the parameter tables is provided as a str it must \n", f"have the value 'all' but '{tables}' was provided.", ) + raise ValueError(msg) if tables == "all": tables = self.table_configs.keys() diff --git a/src/isp_workbook_parser/read_table.py b/src/isp_workbook_parser/read_table.py index 88b628b..6815e36 100644 --- a/src/isp_workbook_parser/read_table.py +++ b/src/isp_workbook_parser/read_table.py @@ -16,7 +16,7 @@ def read_table(workbook_file: pd.ExcelFile, table: TableConfig) -> pd.DataFrame: - """Parses a table given a YAML config for the table + """Parse a table given a YAML config for the table. If `table.header_rows` is an integer, the table is parsed directly. @@ -34,7 +34,6 @@ def read_table(workbook_file: pd.ExcelFile, table: TableConfig) -> pd.DataFrame: header rows in the table are dropped Examples: - The example below reads the "Existing Generators Summary" table from the 2024 version 6 workbook. @@ -68,6 +67,7 @@ def read_table(workbook_file: pd.ExcelFile, table: TableConfig) -> pd.DataFrame: Returns: Table as a pandas DataFrame + """ if isinstance(table.header_rows, int): df = pd.read_excel( @@ -85,70 +85,73 @@ def read_table(workbook_file: pd.ExcelFile, table: TableConfig) -> pd.DataFrame: df, table.columns_with_merged_rows, table.column_range ) return df + + df_initial = pd.read_excel( + workbook_file, + sheet_name=table.sheet_name, + header=(table.header_rows[0] - 1), + usecols=table.column_range, + nrows=(table.end_row - table.header_rows[0]), + # do not parse dtypes + dtype="object", + ) + df_initial.columns = _column_name_sanitiser(df_initial.columns) + # check that header_rows list is sorted + if sorted(table.header_rows) != table.header_rows: + msg = "table header rows are not sorted" + raise ValueError(msg) + # check that the header_rows are adjacent + if set(np.diff(table.header_rows)) != {1}: + msg = "header rows are not adjacent" + raise ValueError(msg) + # start processing multiple header rows + header_rows_in_table = table.header_rows[-1] - table.header_rows[0] + initial_header = pd.Series(df_initial.columns) + ffilled_initial_header = _ffill_highest_header(initial_header) + filled_headers = [] + # ffill intermediate header rows + for i in range(header_rows_in_table - 1): + if i == 0: + preceding_header = initial_header + filled_headers.append( + _ffill_intermediate_header_row(df_initial.iloc[i, :], preceding_header) + ) + preceding_header = df_initial.iloc[i, :] + # process last header row + if not filled_headers: + processed_last_header = _process_last_header_row( + df_initial.iloc[header_rows_in_table - 1, :], ffilled_initial_header + ) else: - df_initial = pd.read_excel( - workbook_file, - sheet_name=table.sheet_name, - header=(table.header_rows[0] - 1), - usecols=table.column_range, - nrows=(table.end_row - table.header_rows[0]), - # do not parse dtypes - dtype="object", + processed_last_header = _process_last_header_row( + df_initial.iloc[header_rows_in_table - 1, :], filled_headers[-1] ) - df_initial.columns = _column_name_sanitiser(df_initial.columns) - # check that header_rows list is sorted - assert sorted(table.header_rows) == table.header_rows - # check that the header_rows are adjacent - assert set(np.diff(table.header_rows)) == set([1]) - # start processing multiple header rows - header_rows_in_table = table.header_rows[-1] - table.header_rows[0] - initial_header = pd.Series(df_initial.columns) - ffilled_initial_header = _ffill_highest_header(initial_header) - filled_headers = [] - # ffill intermediate header rows - for i in range(0, header_rows_in_table - 1): - if i == 0: - preceding_header = initial_header - filled_headers.append( - _ffill_intermediate_header_row(df_initial.iloc[i, :], preceding_header) - ) - preceding_header = df_initial.iloc[i, :] - # process last header row - if not filled_headers: - processed_last_header = _process_last_header_row( - df_initial.iloc[header_rows_in_table - 1, :], ffilled_initial_header - ) - else: - processed_last_header = _process_last_header_row( - df_initial.iloc[header_rows_in_table - 1, :], filled_headers[-1] - ) - filled_headers.append(processed_last_header) - # add separators manually - ignore any "" entries - for series in filled_headers: - series[series != ""] = "_" + series[series != ""] - merged_headers = ffilled_initial_header.str.cat(filled_headers) - df_cleaned = _build_cleaned_dataframe( - df_initial, header_rows_in_table, merged_headers, table.forward_fill_values + filled_headers.append(processed_last_header) + # add separators manually - ignore any "" entries + for series in filled_headers: + series[series != ""] = "_" + series[series != ""] + merged_headers = ffilled_initial_header.str.cat(filled_headers) + df_cleaned = _build_cleaned_dataframe( + df_initial, header_rows_in_table, merged_headers, table.forward_fill_values + ) + if table.skip_rows: + df_cleaned = _skip_rows_in_dataframe( + df_cleaned, table.skip_rows, table.header_rows[-1] ) - if table.skip_rows: - df_cleaned = _skip_rows_in_dataframe( - df_cleaned, table.skip_rows, table.header_rows[-1] - ) - if table.columns_with_merged_rows: - df_cleaned = _handle_merged_rows( - df_cleaned, table.columns_with_merged_rows, table.column_range - ) - return df_cleaned + if table.columns_with_merged_rows: + df_cleaned = _handle_merged_rows( + df_cleaned, table.columns_with_merged_rows, table.column_range + ) + return df_cleaned def _ffill_highest_header(initial_header: pd.Series) -> pd.Series: """ Forward fills the highest header row (parsed as DataFrame columns) for processing - a multi-header table + a multi-header table. """ initial_header[initial_header.str.contains("Unnamed")] = pd.NA - ffill_initial_header = initial_header.ffill().reset_index(drop=True).fillna("") - return ffill_initial_header + return initial_header.ffill().reset_index(drop=True).fillna("") def _ffill_intermediate_header_row( @@ -156,7 +159,7 @@ def _ffill_intermediate_header_row( ) -> pd.Series: """ Forward fills intermediate header row (parsed as a DataFrame row), with the - following strategy: + following strategy. 1. If the nth element value of the intermediate header is NaN, make the nth element equal to the (n-1)th value in the intermediate header row if the @@ -168,7 +171,7 @@ def _ffill_intermediate_header_row( N.B. "Unnamed" columns in pandas are actually NaNs """ int_header = intermediate_header.copy(deep=True) - for n, value in zip(range(1, len(int_header)), int_header.iloc[1:]): + for n, value in zip(range(1, len(int_header)), int_header.iloc[1:], strict=True): preceding_value = preceding_header.iloc[n] if pd.isna(value): if pd.isna(preceding_value): @@ -177,23 +180,22 @@ def _ffill_intermediate_header_row( int_header.iloc[n] = pd.NA _ffill_intermediate_header = int_header.reset_index(drop=True).fillna("") - _ffill_intermediate_header = _column_name_sanitiser(_ffill_intermediate_header) - return _ffill_intermediate_header + return _column_name_sanitiser(_ffill_intermediate_header) def _process_last_header_row( last_header: pd.Series, preceding_header: pd.Series ) -> pd.Series: """ - Processes last header row by removing duplicated table names if the nth element - value is equal to the nth value of the preceding header, + Process last header row by removing duplicated table names if the nth element + value is equal to the nth value of the preceding header. + (e.g. "Name" in row 1 and "Name" in row 2). This is done by making the nth element value an empty string """ last_header = last_header.reset_index(drop=True).fillna("") last_header = _column_name_sanitiser(last_header) - last_header = last_header.where(last_header != preceding_header, "") - return last_header + return last_header.where(last_header != preceding_header, "") def _build_cleaned_dataframe( @@ -203,7 +205,8 @@ def _build_cleaned_dataframe( forward_fill_values: bool, ) -> pd.DataFrame: """ - Builds a cleaned DataFrame with the merged headers by: + Build a cleaned DataFrame with the merged headers using the following steps. + 1. Dropping the header rows in the table 2. Applying the merged headers as the columns of the DataFrame 3. Forward fill values across columns if `forward_fill_values` is True @@ -213,8 +216,7 @@ def _build_cleaned_dataframe( df_cleaned.columns = new_headers if forward_fill_values: df_cleaned = df_cleaned.ffill(axis=1) - df_cleaned = df_cleaned.reset_index(drop=True) - return df_cleaned + return df_cleaned.reset_index(drop=True) def _skip_rows_in_dataframe( @@ -222,7 +224,7 @@ def _skip_rows_in_dataframe( ) -> pd.DataFrame: """ Drop rows specified by `skip_rows` by applying an offset from the header and - dropping based on index values + dropping based on index values. """ df_reset_index = df.reset_index(drop=True) if isinstance(config_skip_rows, int): @@ -232,8 +234,7 @@ def _skip_rows_in_dataframe( skip_rows = np.subtract(skip_rows, last_header_row + 1) else: skip_rows = np.subtract(config_skip_rows, last_header_row + 1) - dropped = df_reset_index.drop(index=skip_rows).reset_index(drop=True) - return dropped + return df_reset_index.drop(index=skip_rows).reset_index(drop=True) def _handle_merged_rows( @@ -242,15 +243,13 @@ def _handle_merged_rows( column_range: str, ) -> pd.DataFrame: """ - Forward fill down columns in `columns_with_merged_rows` + Forward fill down columns in `columns_with_merged_rows`. """ if isinstance(config_cols_with_merged_rows, str): cols = [config_cols_with_merged_rows] else: cols = config_cols_with_merged_rows - actual_col_indices = list( - map(lambda col: _find_data_column_index(col, column_range), cols) - ) + actual_col_indices = [_find_data_column_index(col, column_range) for col in cols] for index in actual_col_indices: df.iloc[:, index] = df.iloc[:, index].ffill() return df @@ -259,7 +258,7 @@ def _handle_merged_rows( def _find_data_column_index( column_alphabetical: str, column_range_from_table_config: str ) -> int: - """Returns the zero-index (integer) index of a column within a table defined by + """Return the zero-index (integer) index of a column within a table defined by a TableConfig column range. Args: @@ -270,9 +269,10 @@ def _find_data_column_index( Returns: Integer index of the column that `column_alphabetical` refers to in the data (zero-indexed) + """ first_col_index = openpyxl.utils.column_index_from_string( - column_range_from_table_config.split(":")[0] + column_range_from_table_config.split(":", maxsplit=1)[0] ) data_col_index = openpyxl.utils.column_index_from_string(column_alphabetical) return data_col_index - first_col_index diff --git a/src/isp_workbook_parser/sanitisers.py b/src/isp_workbook_parser/sanitisers.py index e9b5db7..0d34606 100644 --- a/src/isp_workbook_parser/sanitisers.py +++ b/src/isp_workbook_parser/sanitisers.py @@ -5,6 +5,7 @@ # the Free Software Foundation; either version 3 of the License, or # (at your option) any later version. +import contextlib import re import numpy as np @@ -15,7 +16,8 @@ def _column_name_sanitiser(columns: pd.Index | pd.Series) -> pd.Index | pd.Series: """ - Sanitises column names by: + Sanitises column names using the following steps. + 1. Removing 'versioning' from column names introduced by `mangle_dupe_cols` in pandas parser, e.g. 'Generator.1' is sanitised to 'Generator' 2. Stripping leading and trailing whitespaces @@ -29,14 +31,13 @@ def _column_name_sanitiser(columns: pd.Index | pd.Series) -> pd.Index | pd.Serie columns = columns.str.strip() columns = _replace_series_newlines_with_whitespace(columns) columns = _remove_series_double_whitespaces(columns) - columns = _remove_column_name_trailing_footnotes(columns) - return columns + return _remove_column_name_trailing_footnotes(columns) def _custom_string_replacements( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """If a known typo or unwanted note exits replace it with a known correction""" + """If a known typo or unwanted note exits replace it with a known correction.""" for known_bad_string, correction in typos_and_notes.items(): series = series.str.replace(known_bad_string, correction, regex=True) return series @@ -45,7 +46,9 @@ def _custom_string_replacements( def _remove_column_name_trailing_footnotes( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Removes footnotes by replacing a single trailing digit not preceded by + """Remove footnotes. + + It does this by replacing a single trailing digit not preceded by a hat (e.g. power to in loss equations), whitespace (e.g. name of a unit), another digit (i.e. footnotes are assumed to be single digit) or a capital letter preceded by an underscore (e.g. REZ names) with an empty string""" @@ -53,7 +56,7 @@ def _remove_column_name_trailing_footnotes( def _values_casting_and_sanitisation(df: pd.DataFrame) -> pd.DataFrame: - """Attempts to convert `pd.DataFrame` values to numeric types. If this fails, + """Attempt to convert `pd.DataFrame` values to numeric types. If this fails, sanitises strings in the same column and then re-attempts casting to a numeric type. String sanitisation is only applied to string values in columns that cannot be @@ -81,29 +84,27 @@ def _values_casting_and_sanitisation(df: pd.DataFrame) -> pd.DataFrame: ): df.loc[where_str_values, object_col] = series_func(df[object_col]) # re-attempt conversion following sanitisation - try: + with contextlib.suppress(ValueError, TypeError): df[object_col] = pd.to_numeric(df[object_col]) - except (ValueError, TypeError): - pass return df def _replace_dataframe_hyphens_with_na(df: pd.DataFrame) -> pd.DataFrame: - """Replaces any hyphen values with a `pandas.NA`""" + """Replace any hyphen values with a `pandas.NA`.""" return df.replace("-", np.nan, regex=False) def _replace_series_newlines_with_whitespace( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Replaces newlines in a `pandas.Series` or `pandas.Index` with a whitespace""" + """Replace newlines in a `pandas.Series` or `pandas.Index` with a whitespace.""" return series.str.replace(r"\n", " ", regex=True) def _remove_series_double_whitespaces( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Removes any duplicated whitespaces in a `pandas.Series` or `pandas.Index`""" + """Remove any duplicated whitespaces in a `pandas.Series` or `pandas.Index`.""" return series.str.replace(r"\s\s", " ", regex=True) @@ -111,8 +112,8 @@ def _remove_series_trailing_asterisks( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: """ - Replaces trailing asterisks with an empty string in a `pandas.Series` - or `pandas.Index` + Replace trailing asterisks with an empty string in a `pandas.Series` + or `pandas.Index`. """ return series.str.replace(r"\*$", "", regex=True) @@ -120,7 +121,7 @@ def _remove_series_trailing_asterisks( def _remove_series_trailing_footnotes( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Removes footnotes in a `pandas.Series` or `pandas.Index` + """Remove footnotes in a `pandas.Series` or `pandas.Index`. This is done by replacing a single trailing digit NOT preceded by a whitespace (e.g. name of a unit), another digit (i.e. footnotes are assumed @@ -132,15 +133,15 @@ def _remove_series_trailing_footnotes( def _strip_series_whitespaces(series: pd.Index | pd.Series) -> pd.Index | pd.Series: - """Strips trailing and leading whitespaces in a `pandas.Series` or `pandas.Index`""" + """Strip trailing and leading whitespaces in a `pandas.Series` or `pandas.Index`.""" return series.str.strip(" ") def _remove_series_thousands_commas( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Removes thousands commas (i.e. commas preceded by and following digits) - in a `pandas.Series` or `pandas.Index`""" + """Remove thousands commas (i.e. commas preceded by and following digits) + in a `pandas.Series` or `pandas.Index`.""" return series.str.replace(r"(?<=[0-9]),(?=[0-9]{1,3})", "", regex=True) @@ -180,7 +181,7 @@ def _where_multiple_values_with_notes( def _remove_series_notes_after_values( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: - """Removes notes after numeric values in a `pandas.Series` or `pandas.Index` + """Remove notes after numeric values in a `pandas.Series` or `pandas.Index`. Cells that hold more than one value, each with its own note (see `_where_multiple_values_with_notes`), are left as text rather than run through any @@ -206,7 +207,7 @@ def _remove_series_notes_after_values( regex=True, ) cleaned = cleaned.str.replace(r"^\-\s?(?:(\([\w\s\.\<\=\-\(\)]+)+)", "", regex=True) - series = series.where(keep_full_text, cleaned) + return series.where(keep_full_text, cleaned) return series @@ -214,11 +215,12 @@ def _extract_numeric_value_millions( series: pd.Index | pd.Series, ) -> pd.Index | pd.Series: """ - Extracts numeric value from strings like "$ 2849 M" and multiplies by 1 million. + Extract numeric value from strings like "$ 2849 M" and multiplies by 1 million. + If no 'M' is present, returns the value as is. """ - def extract(val): + def extract(val: any) -> any: # Return value unchanged if it's not a string if not isinstance(val, str): return val @@ -231,9 +233,8 @@ def extract(val): if num_str.replace(".", "", 1).isdigit(): # Convert to float and multiply by 1,000,000 return float(num_str) * 1_000_000 - else: - # If not a valid number, return the original value - return val + # If not a valid number, return the original value + return val # Return value unchanged if pattern does not match return val diff --git a/tests/conftest.py b/tests/conftest.py index fe28d21..49028b4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,8 +13,7 @@ @pytest.fixture(scope="session") def workbook_v6() -> Parser: - workbook = Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") - return workbook + return Parser("workbooks/6.0/2024-isp-inputs-and-assumptions-workbook.xlsx") @pytest.fixture(scope="module") diff --git a/tests/test_read_table_functionality.py b/tests/test_read_table_functionality.py index a046034..580d103 100644 --- a/tests/test_read_table_functionality.py +++ b/tests/test_read_table_functionality.py @@ -45,7 +45,7 @@ def test_skip_multiple_rows_in_multiple_header_row_table(workbook_v6): end_row=48, column_range="B:R", # Victoria - skip_rows=(list(range(29, 35)) + [48]), + skip_rows=[*list(range(29, 35)), 48], ) df = workbook_v6.get_table_from_config(table_config) assert len(df) == (table_config.end_row - table_config.header_rows[-1] - 7)