""" Clean a DataFrame column containing duplicate values. """ from typing import Tuple, List, Dict, Any, Union from ipywidgets.widgets import Label, Dropdown, Checkbox, Button, HBox, VBox, Box, Layout, Text import pandas as pd import dask.dataframe as dd from varname import argname from .clean_duplication_utils import Clusterer DEFAULT_NGRAM = "2" DEFAULT_RADIUS = "2" DEFAULT_BLOCK_SIZE = "6" [docs]def clean_duplication( df: Union[pd.DataFrame, dd.DataFrame], column: str, df_var_name: str = "default", page_size: int = 5, ) -> Box: """ Cleans and standardizes duplicate values in a DataFrame. Read more in the :ref:`User Guide <duplication_userguide>`. Parameters ---------- df A pandas or Dask DataFrame containing the data to be cleaned. column The name of the column containing duplicate values. df_var_name Optional parameter containing the variable name of the DataFrame being cleaned. This is only needed for legacy compatibility with the original veraion of this function, which needed it to produce correct exported code. (default: 'default') page_size The number of clusters to display on each page. (default: 5) Examples -------- After running clean_duplication(df, 'city') below in a notebook, a GUI will appear. Select the merge checkbox, press merge and re-cluster, then press finish. >>> df = pd.DataFrame({'city': ['New York', 'new york']}) >>> clean_duplication(df, 'city') city 0 New York 1 New York """ if df_var_name == "default": df_var_name = argname("df", func=clean_duplication) return UserInterface(df, column, df_var_name, page_size).display() [docs]class UserInterface: """ A user interface used by the clean_duplication function. """ # pylint: disable=too-many-instance-attributes _clusterer: Clusterer _page_size: int _clustering_method_label: Label _clustering_method_drop: Dropdown _export_code: Checkbox _sel_all: Checkbox _next_button: Button _prev_button: Button _page_pos: int _ngram_text: Text _radius_text: Text _block_chars_text: Text _dropds: HBox _reprs: List[Text] _checks: List[Checkbox] _cluster_vbox: VBox _box: Box _loading_label: Label _invalid_param_label: Label def __init__(self, df: pd.DataFrame, col_name: str, df_name: str, page_size: int): self._clusterer = Clusterer(df, col_name, df_name) self._clusterer.cluster("fingerprint") self._page_size = page_size # clustering dropdown and export code checkbox, used in the top row self._clustering_method_label = Label( " Clustering Method: ", layout=Layout(margin="2px 0 0 20px") ) self._clustering_method_drop = Dropdown( options=["fingerprint", "ngram-fingerprint", "phonetic-fingerprint", "levenshtein"], layout=Layout(width="150px", margin="0 0 0 10px"), ) self._clustering_method_drop.observe(self._cluster_method_change, names="value") self._export_code = Checkbox( value=True, description="export code", layout=Layout(width="165px", margin="0 0 0 482px"), style={"description_width": "initial"}, ) self._dropds = HBox( [ self._clustering_method_label, self._clustering_method_drop, self._export_code, ], layout=Layout(height="35px", margin="10px 0 0 0"), ) # text boxes for clustering parameters used in the top row self._ngram_text = Text( value=DEFAULT_NGRAM, description="n-gram", layout=Layout(width="130px"), continuous_update=False, ) self._radius_text = Text( value=DEFAULT_RADIUS, description="Radius", layout=Layout(width="130px"), continuous_update=False, ) self._block_chars_text = Text( value=DEFAULT_BLOCK_SIZE, description="Block Chars", layout=Layout(width="130px"), continuous_update=False, ) self._ngram_text.observe(self._param_recluster, names="value") self._radius_text.observe(self._param_recluster, names="value") self._block_chars_text.observe(self._param_recluster, names="value") # create header labels, second row headers = HBox( [ Label("Distinct values", layout=Layout(margin="0 0 0 10px")), Label("Total values", layout=Layout(margin="0 0 0 35px")), Label("Cluster values", layout=Layout(margin="0 0 0 95px")), Label("Merge?", layout=Layout(margin="0 0 0 295px")), Label("Representative value", layout=Layout(margin="0 0 0 50px")), ], layout=Layout(margin="10px"), ) # create buttons for bottom row self._sel_all = Checkbox(description="Select all", layout=Layout(width="165px")) self._sel_all.observe(self._select_all, names="value") merge_and_recluster = Button( description="Merge and Re-Cluster", layout=Layout(margin="0 0 0 466px", width="150px") ) merge_and_recluster.on_click(self._execute_merge) finish = Button(description="Finish", layout=Layout(margin="0 0 0 10px")) finish.on_click(self._close) # next and previous page buttons self._next_button = Button(description="Next") self._next_button.on_click(self._next_page) self._prev_button = Button(description="Previous", layout=Layout(margin="0 0 0 20px")) self._prev_button.on_click(self._prev_page) # an index in the clusters Series indicating the start of the current page self._page_pos = 0 # loading label, displayed when re-clustering or next page load self._loading_label = Label("Loading...", layout=Layout(margin="170px 0 0 440px")) # displayed when the user enters a non integer value into a clustering parameter text box self._invalid_param_label = Label( "Invalid clustering parameter, please enter an integer", layout=Layout(margin="170px 0 0 350px"), ) self._reprs = [ Text(layout=Layout(width="200px", margin="0 10px 0 40px")) for _ in range(self._page_size) ] self._checks = [ Checkbox(indent=False, layout=Layout(width="auto", margin="0 0 0 20px")) for _ in range(self._page_size) ] # VBox containing a VBox with all the clusters in the first row and an optional # second row containing next and previous page buttons self._cluster_and_next_prev = VBox() self._cluster_vbox = VBox(layout=Layout(height="450px", flex_flow="row wrap")) footer = HBox([self._sel_all, merge_and_recluster, finish]) box_children = [self._dropds, headers, self._cluster_and_next_prev, footer] box_layout = Layout( display="flex", flex_flow="column", align_items="stretch", border="solid" ) self._box = Box(children=box_children, layout=box_layout) self._update_clusters() def _update_clusters(self) -> None: """ Updates the clusters currently being displayed. """ line = HBox(children=[Label("-" * 186, layout=Layout(margin="0 0 0 18px"))]) self._sel_all.value = False cluster_page = self._clusterer.get_page(self._page_pos, self._page_pos + self._page_size) label_layout = Layout(height="22px", width="360px") box_children = [line] for idx, cluster in enumerate(cluster_page): labels = [] for cluster_val, cnt in cluster: if cnt > 1: cluster_val += f" ({cnt} rows)" labels.append(Label(cluster_val, layout=label_layout)) totals_vals = sum(cnt for _, cnt in cluster) distinct_vals = len(cluster) self._reprs[idx].value = cluster[0][0] self._checks[idx].value = False box_children.append( HBox( [ Label(str(distinct_vals), layout=Layout(width="60px", margin="0 0 0 60px")), Label(str(totals_vals), layout=Layout(width="60px", margin="0 0 0 50px")), VBox(children=labels, layout=Layout(margin="0 0 0 80px")), self._checks[idx], self._reprs[idx], ] ) ) box_children.append(line) # no clusters to display if len(cluster_page) == 0: box_children = [ Label( "No clusters, try a different clustering method", layout=Layout(margin="170px 0 0 360px"), ) ] self._cluster_vbox.children = box_children cluster_and_next_prev = [self._cluster_vbox] self._add_next_prev_button_row(cluster_and_next_prev) self._cluster_and_next_prev.children = cluster_and_next_prev def _update_dropds(self, clustering_method: str) -> None: """ Update the dropdowns row of the UI to display the required text boxes for passing parameters needed for the given clustering method. """ if clustering_method in ("fingerprint", "phonetic-fingerprint"): self._export_code.layout.margin = "0 0 0 482px" self._dropds.children = [ self._clustering_method_label, self._clustering_method_drop, self._export_code, ] if clustering_method == "ngram-fingerprint": self._export_code.layout.margin = "0 0 0 348px" self._dropds.children = [ self._clustering_method_label, self._clustering_method_drop, self._ngram_text, self._export_code, ] if clustering_method == "levenshtein": self._export_code.layout.margin = "0 0 0 214px" self._dropds.children = [ self._clustering_method_label, self._clustering_method_drop, self._radius_text, self._block_chars_text, self._export_code, ] def _param_recluster(self, _: Dict[str, Any]) -> None: """ Re-cluster the dataframe with the new clustering parameters. Triggered when the value in a clustering parameter textbox is changed. """ self._display_message(self._loading_label) try: self._clusterer.set_cluster_params(*self._cluster_params()) cluster_method = self._clustering_method_drop.value self._clusterer.cluster(cluster_method) self._page_pos = 0 self._update_clusters() except ValueError: self._display_message(self._invalid_param_label) def _cluster_params(self) -> Tuple[int, int, int]: """ Retrieve clustering parameters from their respective text boxes. """ ngram = self._ngram_text.value if self._ngram_text.value else DEFAULT_NGRAM radius = self._radius_text.value if self._radius_text.value else DEFAULT_RADIUS block_size = self._block_chars_text.value if self._block_chars_text else DEFAULT_BLOCK_SIZE return int(ngram), int(radius), int(block_size) def _select_all(self, change: Dict[str, Any]) -> None: """ Triggered when the select all checkbox is selected or unselected. Changes the value of the cluster checkboxes to match the state of the select all checkbox. """ for check in self._checks: check.value = change["new"] def _cluster_method_change(self, change: Dict[str, Any]) -> None: """ Triggered when the cluster method dropdown state is changed. Re-clusters the DataFrame with the new clustering method. """ self._update_dropds(change["new"]) self._display_message(self._loading_label) cluster_method = self._clustering_method_drop.value self._clusterer.cluster(cluster_method) self._page_pos = 0 self._update_clusters() def _add_next_prev_button_row(self, box_children: List[Union[HBox, VBox]]) -> None: """ Adds a next page or previous page button, if the operation is valid. """ next_prev = [] prev_is_valid = self._page_pos - self._page_size >= 0 next_is_valid = self._page_pos + self._page_size < len(self._clusterer.clusters) if prev_is_valid and next_is_valid: self._next_button.layout.margin = "0 0 0 628px" next_prev.append(self._prev_button) next_prev.append(self._next_button) elif prev_is_valid: next_prev.append(self._prev_button) elif next_is_valid: self._next_button.layout.margin = "0 0 0 795px" next_prev.append(self._next_button) if next_is_valid or prev_is_valid: box_children.append(HBox(next_prev, layout={"height": "50px"})) def _next_page(self, _: Dict[str, Any]) -> None: """ Display the next page of clusters by increasing the page position. """ self._display_message(self._loading_label) self._page_pos += self._page_size self._update_clusters() self._sel_all_on_page() def _prev_page(self, _: Dict[str, Any]) -> None: """ Display the previous page of clusters by decreasing the page position. """ self._display_message(self._loading_label) self._page_pos -= self._page_size self._update_clusters() self._sel_all_on_page() def _display_message(self, message_label: Label) -> None: """ Display a message to the user, used for the loading screen and invalid clustering parameter screen """ self._cluster_vbox.children = [message_label] # don't display next and prev buttons self._cluster_and_next_prev.children = [self._cluster_and_next_prev.children[0]] def _sel_all_on_page(self) -> None: """ Select all checkboxes on the current page of clusters if the select all checkbox is selected. """ if self._sel_all.value: for check in self._checks: check.value = True def _close(self, _: Dict[str, Any]) -> None: """ Close the UI and display the final dataframe in the next jupyter notebook cell. """ self._clusterer.final_df() self._box.close() def _execute_merge(self, _: Dict[str, Any]) -> None: """ Merge the selected clusters and re-cluster the dataframe. If the export code checkbox is selected the required replace function calls and add them to the jupyter notebook cell. """ self._display_message(self._loading_label) do_merge = [check.value for check in self._checks] new_values = [text.value for text in self._reprs] cluster_page = self._clusterer.get_page(self._page_pos, self._page_pos + self._page_size) if self._export_code.value: self._clusterer.live_export_code(cluster_page, do_merge, new_values) cluster_method = self._clustering_method_drop.value self._clusterer.execute_merge_code(cluster_page, do_merge, new_values) self._clusterer.cluster(cluster_method) self._update_page_if_empty() self._update_clusters() def _update_page_if_empty(self) -> None: """ Decrease the page if the last page is empty. Needed for when all clusters on the last page are merged. """ if self._page_pos >= len(self._clusterer.clusters): self._page_pos -= self._page_size [docs] def display(self) -> Box: """Display the UI.""" return self._box