24912462866

Committed 24 Apr 2026 09:21PM UTC coverage: 76.362% (-3.1%) from 79.464%

Build # 24912462866

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #73 from rafaelpadilla/eduardo/refactor_data_operations

Refactor of data operations, trainers and models.

Coverage Stats

244 of 339 branches covered (71.98%)

Branch coverage included in aggregate %.

1317 of 1706 new or added lines in 50 files covered. (77.2%)

28 existing lines in 5 files now uncovered.

2124 of 2762 relevant lines covered (76.9%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

17.39

/toolkit/ThreeWToolkit/data_visualization/three_w_chart.py

import logging

import pandas as pd
import matplotlib.colors as mcolors
import plotly.graph_objects as go

from pathlib import Path
from matplotlib import pyplot as plt
from matplotlib.figure import Figure
from ..utils.data_utils import get_config_dataset_ini
from ..core.base_visualizer import BaseVisualizer

logger = logging.getLogger(__name__)


class ThreeWChart(BaseVisualizer):
    """A class to generate interactive visualizations for 3W dataset files using Plotly.

    Notes
    -----
    Developed by: Yan Tavares (2025) | Github: https://github.com/yantavares
    Adapted by: Matheus Ferreira (2025) | Github: https://github.com/Mathtzt
    """

    def __init__(
        self,
        file_path: str,
        title: str = "ThreeW Chart",
        y_axis: str = "P-MON-CKP",
        use_dropdown: bool = False,
        dropdown_position: tuple = (0.4, 1.4),
    ):
        """Initializes the ThreeWChart class with the given parameters.

        Args:
            file_path (str): Path to the Parquet file containing the dataset.
            title (str, optional): Title of the chart. Defaults to "ThreeW Chart".
            y_axis (str, optional): olumn name to be plotted on the y-axis. Defaults to "P-MON-CKP".
            use_dropdown (bool, optional):  Whether to show a dropdown for selecting the y-axis (default is False). Defaults to False.
            dropdown_position (tuple, optional): Position of the dropdown button on the chart. Defaults to (0.4, 1.4).
        """
        self.file_path: str = file_path
        self.title: str = title
        self.y_axis: str = y_axis
        self.use_dropdown: bool = use_dropdown
        self.dropdown_position: tuple = dropdown_position

        self.dataset_ini: dict = get_config_dataset_ini()
        self.class_mapping: dict[int, str] = self._generate_class_mapping()
        self.class_colors: dict[int, str] = self._generate_class_colors()

    def _generate_class_mapping(self) -> dict[int, str]:
        """Generate a combined mapping of event labels (including transient states) to their descriptions.

        Returns:
            dict[int, str]: Mapping of event labels to their descriptions.
        """
        return {
            **self.dataset_ini["LABELS_DESCRIPTIONS"],
            **self.dataset_ini["TRANSIENT_LABELS_DESCRIPTIONS"],
        }

    def _generate_class_colors(self) -> dict[int, str]:
        """Automatically generate a color mapping for event labels using a colormap.
        For transient states, the color is the event color with lower opacity.

        Returns:
            dict[int, str]: Mapping of event labels to their colors.
        """
        cmap = plt.get_cmap("tab10")
        colors = {}

        def apply_transparency(color: str, opacity: float) -> str:
            rgb = mcolors.to_rgb(color)
            r, g, b = int(rgb[0] * 255), int(rgb[1] * 255), int(rgb[2] * 255)
            return f"rgba({r}, {g}, {b}, {opacity})"

        for idx, (label, _) in enumerate(
            self.dataset_ini["LABELS_DESCRIPTIONS"].items()
        ):
            if label == 0:
                base_color = "white"
            else:
                base_color = mcolors.rgb2hex(cmap(idx % cmap.N))
            colors[label] = base_color

            transient_label = label + self.dataset_ini["TRANSIENT_OFFSET"]
            colors[transient_label] = (
                "white" if label == 0 else apply_transparency(base_color, opacity=0.4)
            )
        return colors

    def _load_data(self) -> pd.DataFrame:
        """Loads and preprocesses the dataset using the load_instance function.

        Returns:
            pd.DataFrame: Preprocessed DataFrame with sorted timestamps and no missing values.
        """
        instance = (int(Path(self.file_path).parent.name), Path(self.file_path))
        df = self._load_instance(instance)
        df.reset_index(inplace=True)
        df = df.dropna(subset=["timestamp"]).drop_duplicates("timestamp").fillna(0)
        return df.sort_values(by="timestamp")

    def _get_non_zero_columns(self, df: pd.DataFrame) -> list[str]:
        """Returns the list of columns that are not all zeros or NaN.

        Args:
            df (pd.DataFrame): DataFrame to check for non-zero columns.

        Returns:
            list[str]: List of column names that are not all zeros or NaN.
        """
        return [
            col
            for col in df.columns
            if df[col].astype(bool).sum() > 0 and col not in ["timestamp", "class"]
        ]

    def _get_background_shapes(self, df: pd.DataFrame) -> list[dict]:
        """Creates background shapes to highlight class transitions in the chart.

        Args:
            df (pd.DataFrame): DataFrame containing the class data.

        Returns:
            list[dict]: List of shape dictionaries for Plotly.
        """
        shapes = []
        prev_class = None
        start_idx = 0

        for i in range(len(df)):
            current_class = df.iloc[i]["class"]

            if pd.isna(current_class):
                logger.warning(f"Warning: NaN class value at index {i}")
                continue

            if prev_class is not None and current_class != prev_class:
                shapes.append(
                    dict(
                        type="rect",
                        x0=df.iloc[start_idx]["timestamp"],
                        x1=df.iloc[i - 1]["timestamp"],
                        y0=0,
                        y1=1,
                        xref="x",
                        yref="paper",
                        fillcolor=self.class_colors.get(prev_class, "white"),
                        opacity=0.2,
                        line_width=0,
                    )
                )
                start_idx = i

            prev_class = current_class

        if prev_class is not None:
            shapes.append(
                dict(
                    type="rect",
                    x0=df.iloc[start_idx]["timestamp"],
                    x1=df.iloc[len(df) - 1]["timestamp"],
                    y0=0,
                    y1=1,
                    xref="x",
                    yref="paper",
                    fillcolor=self.class_colors.get(prev_class, "white"),
                    opacity=0.2,
                    line_width=0,
                )
            )

        return shapes

    def _add_custom_legend(self, fig: go.Figure, present_classes: list[int]) -> None:
        """Adds a custom legend to the chart for only those classes present in the data.

        Args:
            fig (go.Figure): The Plotly figure to which the legend will be added.
            present_classes (list[int]): The unique class values present in the DataFrame.
        """
        for class_value in present_classes:
            if class_value in self.class_mapping:
                event_name = self.class_mapping[class_value]
                fig.add_trace(
                    go.Scatter(
                        x=[None],
                        y=[None],
                        mode="markers",
                        marker=dict(
                            size=12,
                            color=self.class_colors.get(class_value, "white"),
                            line=dict(width=1, color="black"),
                        ),
                        name=f"{class_value} - {event_name}",
                        showlegend=True,
                    )
                )

    def _load_instance(self, instance):
        """Loads all data and metadata from a specific `instance`.

        Args:
            instance (tuple): This tuple must refer to a specific `instance`
                and contain its label (int) and its full path (Path).

        Raises:
            Exception: Error if the Parquet file passed as arg cannot be
            read.

        Returns:
            pandas.DataFrame: Its index contains the timestamps loaded from
                the Parquet file. Its columns contain data loaded from the
                other columns of the Parquet file and metadata loaded from
                the argument `instance` (label, well, and id).
        """
        # Loads label metadata from the argument `instance`
        label, fp = instance

        try:
            # Loads well and id metadata from the argument `instance`
            well, id = fp.stem.split("_")

            # Loads data from the Parquet file
            df = pd.read_parquet(fp, engine="pyarrow")
            expected = self.dataset_ini["COLUMNS_DATA_FILES"][1:]
            if not all(df.columns == expected):
                raise ValueError(
                    f"Invalid columns in the file {fp}: {df.columns.tolist()}"
                )

        except Exception as e:
            raise Exception(f"error reading file {fp}: {e}")

        # Incorporates the loaded metadata
        df["label"] = label
        df["well"] = well
        df["id"] = id

        # Incorporates the loaded data and ordenates the df's columns
        df = df[["label", "well", "id"] + self.dataset_ini["COLUMNS_DATA_FILES"][1:]]

        return df

    def plot(self, ax=None) -> tuple[Figure, None]:
        """Generate and display an interactive Plotly chart.

        This method creates a Plotly figure based on the available data and
        configuration options. Since Plotly does not use Matplotlib axes,
        the returned Axes object is always None.

        Returns:
            Tuple[Figure, Optional[Axes]]:
                - Figure: The generated Plotly figure.
                - Axes: Always None (not applicable for Plotly).

        Raises:
            ValueError: If no valid columns are available for plotting.
        """
        df = self._load_data()

        present_classes = df["class"].dropna().unique().tolist()

        if self.use_dropdown:
            available_y_axes = self._get_non_zero_columns(df)
            if available_y_axes:
                dropdown_buttons = [
                    dict(
                        args=[{"y": [df[col]]}, {"yaxis.title": col}],
                        label=col,
                        method="update",
                    )
                    for col in available_y_axes
                ]
                fig = go.Figure()
                if self.y_axis not in available_y_axes:
                    logger.warning(
                        f"Warning: Default y-axis '{self.y_axis}' not found in available columns."
                    )
                    logger.info(
                        "Using the first available column as the default y-axis."
                    )
                    self.y_axis = available_y_axes[0]
                fig.add_trace(
                    go.Scatter(
                        x=df["timestamp"],
                        y=df[self.y_axis],
                        mode="lines",
                        name="Selected Variable",
                    )
                )
                active_index = available_y_axes.index(self.y_axis)
                fig.update_layout(
                    updatemenus=[
                        dict(
                            buttons=dropdown_buttons,
                            direction="down",
                            showactive=True,
                            x=self.dropdown_position[0],
                            y=self.dropdown_position[1],
                            active=active_index,
                        )
                    ]
                )
            else:
                raise ValueError("No available columns to plot.")
        else:
            fig = go.Figure()
            fig.add_trace(
                go.Scatter(
                    x=df["timestamp"], y=df[self.y_axis], mode="lines", name=self.y_axis
                )
            )

        fig.update_xaxes(rangeslider_visible=True)
        fig.update_layout(
            shapes=self._get_background_shapes(df),
            xaxis_title="Timestamp",
            yaxis_title=self.y_axis if not self.use_dropdown else df[self.y_axis].name,
            title=self.title,
            legend=dict(
                x=1.05, y=1, title="Legend", itemclick=False, itemdoubleclick=False
            ),
        )

        self._add_custom_legend(fig, present_classes)

        return fig, None

1	import logging	1✔
2
3	import pandas as pd	1✔
4	import matplotlib.colors as mcolors	1✔
5	import plotly.graph_objects as go	1✔
6
7	from pathlib import Path	1✔
8	from matplotlib import pyplot as plt	1✔
9	from matplotlib.figure import Figure	1✔
10	from ..utils.data_utils import get_config_dataset_ini	1✔
11	from ..core.base_visualizer import BaseVisualizer	1✔
12
13	logger = logging.getLogger(__name__)	1✔
14
15
16	class ThreeWChart(BaseVisualizer):	1✔
17	"""A class to generate interactive visualizations for 3W dataset files using Plotly.
18
19	Notes
20	-----
21	Developed by: Yan Tavares (2025) \| Github: https://github.com/yantavares
22	Adapted by: Matheus Ferreira (2025) \| Github: https://github.com/Mathtzt
23	"""
24
25	def __init__(	1✔
26	self,
27	file_path: str,
28	title: str = "ThreeW Chart",
29	y_axis: str = "P-MON-CKP",
30	use_dropdown: bool = False,
31	dropdown_position: tuple = (0.4, 1.4),
32	):
33	"""Initializes the ThreeWChart class with the given parameters.
34
35	Args:
36	file_path (str): Path to the Parquet file containing the dataset.
37	title (str, optional): Title of the chart. Defaults to "ThreeW Chart".
38	y_axis (str, optional): olumn name to be plotted on the y-axis. Defaults to "P-MON-CKP".
39	use_dropdown (bool, optional): Whether to show a dropdown for selecting the y-axis (default is False). Defaults to False.
40	dropdown_position (tuple, optional): Position of the dropdown button on the chart. Defaults to (0.4, 1.4).
41	"""
42	self.file_path: str = file_path	×
43	self.title: str = title	×
44	self.y_axis: str = y_axis	×
45	self.use_dropdown: bool = use_dropdown	×
46	self.dropdown_position: tuple = dropdown_position	×
47
48	self.dataset_ini: dict = get_config_dataset_ini()	×
49	self.class_mapping: dict[int, str] = self._generate_class_mapping()	×
50	self.class_colors: dict[int, str] = self._generate_class_colors()	×
51
52	def _generate_class_mapping(self) -> dict[int, str]:	1✔
53	"""Generate a combined mapping of event labels (including transient states) to their descriptions.
54
55	Returns:
56	dict[int, str]: Mapping of event labels to their descriptions.
57	"""
58	return {	×
59	**self.dataset_ini["LABELS_DESCRIPTIONS"],
60	**self.dataset_ini["TRANSIENT_LABELS_DESCRIPTIONS"],
61	}
62
63	def _generate_class_colors(self) -> dict[int, str]:	1✔
64	"""Automatically generate a color mapping for event labels using a colormap.
65	For transient states, the color is the event color with lower opacity.
66
67	Returns:
68	dict[int, str]: Mapping of event labels to their colors.
69	"""
70	cmap = plt.get_cmap("tab10")	×
71	colors = {}	×
72
73	def apply_transparency(color: str, opacity: float) -> str:	×
74	rgb = mcolors.to_rgb(color)	×
75	r, g, b = int(rgb[0] * 255), int(rgb[1] * 255), int(rgb[2] * 255)	×
76	return f"rgba({r}, {g}, {b}, {opacity})"	×
77
78	for idx, (label, _) in enumerate(	×
79	self.dataset_ini["LABELS_DESCRIPTIONS"].items()
80	):
81	if label == 0:	×
82	base_color = "white"	×
83	else:
84	base_color = mcolors.rgb2hex(cmap(idx % cmap.N))	×
85	colors[label] = base_color	×
86
87	transient_label = label + self.dataset_ini["TRANSIENT_OFFSET"]	×
88	colors[transient_label] = (	×
89	"white" if label == 0 else apply_transparency(base_color, opacity=0.4)
90	)
91	return colors	×
92
93	def _load_data(self) -> pd.DataFrame:	1✔
94	"""Loads and preprocesses the dataset using the load_instance function.
95
96	Returns:
97	pd.DataFrame: Preprocessed DataFrame with sorted timestamps and no missing values.
98	"""
99	instance = (int(Path(self.file_path).parent.name), Path(self.file_path))	×
100	df = self._load_instance(instance)	×
101	df.reset_index(inplace=True)	×
102	df = df.dropna(subset=["timestamp"]).drop_duplicates("timestamp").fillna(0)	×
103	return df.sort_values(by="timestamp")	×
104
105	def _get_non_zero_columns(self, df: pd.DataFrame) -> list[str]:	1✔
106	"""Returns the list of columns that are not all zeros or NaN.
107
108	Args:
109	df (pd.DataFrame): DataFrame to check for non-zero columns.
110
111	Returns:
112	list[str]: List of column names that are not all zeros or NaN.
113	"""
114	return [	×
115	col
116	for col in df.columns
117	if df[col].astype(bool).sum() > 0 and col not in ["timestamp", "class"]
118	]
119
120	def _get_background_shapes(self, df: pd.DataFrame) -> list[dict]:	1✔
121	"""Creates background shapes to highlight class transitions in the chart.
122
123	Args:
124	df (pd.DataFrame): DataFrame containing the class data.
125
126	Returns:
127	list[dict]: List of shape dictionaries for Plotly.
128	"""
129	shapes = []	×
130	prev_class = None	×
131	start_idx = 0	×
132
133	for i in range(len(df)):	×
134	current_class = df.iloc[i]["class"]	×
135
136	if pd.isna(current_class):	×
NEW 137	logger.warning(f"Warning: NaN class value at index {i}")	×
138	continue	×
139
140	if prev_class is not None and current_class != prev_class:	×
141	shapes.append(	×
142	dict(
143	type="rect",
144	x0=df.iloc[start_idx]["timestamp"],
145	x1=df.iloc[i - 1]["timestamp"],
146	y0=0,
147	y1=1,
148	xref="x",
149	yref="paper",
150	fillcolor=self.class_colors.get(prev_class, "white"),
151	opacity=0.2,
152	line_width=0,
153	)
154	)
155	start_idx = i	×
156
157	prev_class = current_class	×
158
159	if prev_class is not None:	×
160	shapes.append(	×
161	dict(
162	type="rect",
163	x0=df.iloc[start_idx]["timestamp"],
164	x1=df.iloc[len(df) - 1]["timestamp"],
165	y0=0,
166	y1=1,
167	xref="x",
168	yref="paper",
169	fillcolor=self.class_colors.get(prev_class, "white"),
170	opacity=0.2,
171	line_width=0,
172	)
173	)
174
175	return shapes	×
176
177	def _add_custom_legend(self, fig: go.Figure, present_classes: list[int]) -> None:	1✔
178	"""Adds a custom legend to the chart for only those classes present in the data.
179
180	Args:
181	fig (go.Figure): The Plotly figure to which the legend will be added.
182	present_classes (list[int]): The unique class values present in the DataFrame.
183	"""
184	for class_value in present_classes:	×
185	if class_value in self.class_mapping:	×
186	event_name = self.class_mapping[class_value]	×
187	fig.add_trace(	×
188	go.Scatter(
189	x=[None],
190	y=[None],
191	mode="markers",
192	marker=dict(
193	size=12,
194	color=self.class_colors.get(class_value, "white"),
195	line=dict(width=1, color="black"),
196	),
197	name=f"{class_value} - {event_name}",
198	showlegend=True,
199	)
200	)
201
202	def _load_instance(self, instance):	1✔
203	"""Loads all data and metadata from a specific `instance`.
204
205	Args:
206	instance (tuple): This tuple must refer to a specific `instance`
207	and contain its label (int) and its full path (Path).
208
209	Raises:
210	Exception: Error if the Parquet file passed as arg cannot be
211	read.
212
213	Returns:
214	pandas.DataFrame: Its index contains the timestamps loaded from
215	the Parquet file. Its columns contain data loaded from the
216	other columns of the Parquet file and metadata loaded from
217	the argument `instance` (label, well, and id).
218	"""
219	# Loads label metadata from the argument `instance`
220	label, fp = instance	×
221
222	try:	×
223	# Loads well and id metadata from the argument `instance`
224	well, id = fp.stem.split("_")	×
225
226	# Loads data from the Parquet file
227	df = pd.read_parquet(fp, engine="pyarrow")	×
228	expected = self.dataset_ini["COLUMNS_DATA_FILES"][1:]	×
229	if not all(df.columns == expected):	×
230	raise ValueError(	×
231	f"Invalid columns in the file {fp}: {df.columns.tolist()}"
232	)
233
234	except Exception as e:	×
235	raise Exception(f"error reading file {fp}: {e}")	×
236
237	# Incorporates the loaded metadata
238	df["label"] = label	×
239	df["well"] = well	×
240	df["id"] = id	×
241
242	# Incorporates the loaded data and ordenates the df's columns
243	df = df[["label", "well", "id"] + self.dataset_ini["COLUMNS_DATA_FILES"][1:]]	×
244
245	return df	×
246
247	def plot(self, ax=None) -> tuple[Figure, None]:	1✔
248	"""Generate and display an interactive Plotly chart.
249
250	This method creates a Plotly figure based on the available data and
251	configuration options. Since Plotly does not use Matplotlib axes,
252	the returned Axes object is always None.
253
254	Returns:
255	Tuple[Figure, Optional[Axes]]:
256	- Figure: The generated Plotly figure.
257	- Axes: Always None (not applicable for Plotly).
258
259	Raises:
260	ValueError: If no valid columns are available for plotting.
261	"""
262	df = self._load_data()	×
263
264	present_classes = df["class"].dropna().unique().tolist()	×
265
266	if self.use_dropdown:	×
267	available_y_axes = self._get_non_zero_columns(df)	×
268	if available_y_axes:	×
269	dropdown_buttons = [	×
270	dict(
271	args=[{"y": [df[col]]}, {"yaxis.title": col}],
272	label=col,
273	method="update",
274	)
275	for col in available_y_axes
276	]
277	fig = go.Figure()	×
278	if self.y_axis not in available_y_axes:	×
NEW 279	logger.warning(	×
280	f"Warning: Default y-axis '{self.y_axis}' not found in available columns."
281	)
NEW 282	logger.info(	×
283	"Using the first available column as the default y-axis."
284	)
285	self.y_axis = available_y_axes[0]	×
286	fig.add_trace(	×
287	go.Scatter(
288	x=df["timestamp"],
289	y=df[self.y_axis],
290	mode="lines",
291	name="Selected Variable",
292	)
293	)
294	active_index = available_y_axes.index(self.y_axis)	×
295	fig.update_layout(	×
296	updatemenus=[
297	dict(
298	buttons=dropdown_buttons,
299	direction="down",
300	showactive=True,
301	x=self.dropdown_position[0],
302	y=self.dropdown_position[1],
303	active=active_index,
304	)
305	]
306	)
307	else:
308	raise ValueError("No available columns to plot.")	×
309	else:
310	fig = go.Figure()	×
311	fig.add_trace(	×
312	go.Scatter(
313	x=df["timestamp"], y=df[self.y_axis], mode="lines", name=self.y_axis
314	)
315	)
316
317	fig.update_xaxes(rangeslider_visible=True)	×
318	fig.update_layout(	×
319	shapes=self._get_background_shapes(df),
320	xaxis_title="Timestamp",
321	yaxis_title=self.y_axis if not self.use_dropdown else df[self.y_axis].name,
322	title=self.title,
323	legend=dict(
324	x=1.05, y=1, title="Legend", itemclick=False, itemdoubleclick=False
325	),
326	)
327
328	self._add_custom_legend(fig, present_classes)	×
329
330	return fig, None	×

rafaelpadilla / 3W / 24912462866

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous