Код: Выделить всё
interactive_plot.pyКод: Выделить всё
import matplotlib.pyplot as plt
import numpy as np
from sklearn.preprocessing import MinMaxScaler
# this finds the closest datum to the click point
def find_closest_point(data, click_coords, mass, scaler):
scaled_data = scaler.transform(data[mass][['time_sec', mass]])
scaled_click = scaler.transform([click_coords])
x_diff = np.abs(scaled_data[:,0] - scaled_click[0, 0])
y_diff = np.abs(scaled_data[:,1] - scaled_click[0, 1])
distances = np.sqrt(x_diff**2 + y_diff**2)
closest_index = distances.argmin()
column_name = distances[closest_index]
return closest_index, column_name
# what to do when the user clicks on the plot
def on_click(event, data_entry, fig):
click_coords = (event.xdata, event.ydata)
active_subplot = event.inaxes
mass = active_subplot.get_title()
# produce a scale for the data so that the least squares fit works
scaler = MinMaxScaler()
scaler.fit(data_entry.raw_data[mass].values)
closest_index, column_name = find_closest_point(data_entry.raw_data, click_coords, mass, scaler)
data_entry.data_status[mass].iloc[closest_index] = (data_entry.data_status[mass].iloc[closest_index] + 1) % 2
# if any datum is excluded from 3 amu or 4 amu, also exclude it from the 4/3 Ratio
if column_name in ['3 amu', '4 amu']:
ratio_index = data_entry.raw_data[mass][data_entry.raw_data[mass]['time_sec'] == data_entry.raw_data[mass].loc[closest_index, 'time_sec']].index(0)
data_entry.data_status[mass][ratio_index, '4/3 Ratio'] = 0
update_plot_and_trendlines(data_entry, fig)
# updates the plot
def update_plot_and_trendlines(data_entry, fig):
fig.clf()
axes = fig.subplots(2,3)
# Store plot styling for each mass
plot_kwargs = {
"3 amu": {"color": "blue"},
"4 amu": {"color": "red"},
"4/3 Ratio": {"color": "purple"},
"2 amu": {"edgecolor": "black", "facecolor": "white", "linewidth": 1},
"40 amu": {"color": "magenta"},
"5 amu": {"color": "gray"}
}
# loop to plot masses on the top row
for ax, mass in zip(axes[0], ['3 amu', '4 amu', '4/3 Ratio']):
# define x and y
x = data_entry.raw_data[mass]['time_sec']
y = data_entry.raw_data[mass][mass]
# get active and inactive indices
active_indices = data_entry.data_status[mass].to_numpy().flatten()
# draw active and inactive points
ax.scatter(x[active_indices==1], y[active_indices==1], **plot_kwargs[mass])
ax.scatter(x[active_indices==0], y[active_indices==0], marker='x', color='gray')
# dress up the plot
ax.set_xlabel('Time (s)')
ax.set_ylabel('Intensity (A)')
ax.set_title(mass)
# draw trendline/average from active indices
if mass == '4/3 Ratio':
average_y = np.mean(y[active_indices==1]) if np.any(active_indices) else np.nan
trend_x = [x.min(), x.max()]
trend_y = [average_y, average_y]
ax.plot(trend_x, trend_y, color="black")
else:
trend = np.polyfit(x[active_indices==1], y[active_indices==1], 1)
trend_x = [x.min(), x.max()]
trend_y = np.polyval(trend, trend_x)
ax.plot(trend_x, trend_y, color="black")
# loop to plot masses in the bottom row
for ax, mass in zip(axes[1], ['2 amu', '40 amu', '5 amu']):
# define x and y
x = data_entry.raw_data[mass]['time_sec']
y = data_entry.raw_data[mass][mass]
# get active and inactive indices
active_indices = data_entry.data_status[mass].to_numpy().flatten()
# draw active and inactive plots
ax.scatter(x[active_indices==1], y[active_indices==1], **plot_kwargs[mass])
ax.scatter(x[active_indices==0], y[active_indices==0], marker='x', color='gray')
# dress up the plot
ax.set_xlabel('Time (s)')
ax.set_ylabel('Intensity (A)')
ax.set_title(mass)
# draw trendline/average from active indices
if mass == '5 amu':
average_y = np.mean(y[active_indices==1]) if np.any(active_indices) else np.nan
trend_x = [x.min(), x.max()]
trend_y = [average_y, average_y]
ax.plot(trend_x, trend_y, color="black")
else:
trend = np.polyfit(x[active_indices==1], y[active_indices==1], 1)
trend_x = [x.min(), x.max()]
trend_y = np.polyval(trend, trend_x)
ax.plot(trend_x, trend_y, color="black")
def plot_raw_data(data_entry):
fig, axes = plt.subplots(2, 3, figsize=(15, 8)) # Create 2x3 subplot grid
update_plot_and_trendlines(data_entry, fig) # draw
plt.tight_layout() # Adjust spacing to prevent labels overlapping
fig.canvas.mpl_connect('button_press_event', lambda event: on_click(event, data_entry, fig)) # interactivity
return fig
- Когда active_indices обращается в ноль для правильных данных при нажатии я не вижу обновления линии тренда и не вижу, чтобы данные перерисовывались в виде серого «x», как ожидалось. Код построения графика в этом отношении довольно прост, поэтому я не понимаю, почему он не работает.
- Я использую MinMaxScaler(), потому что мои данные y находятся на уровне как минимум на девять порядков меньше, чем мои данные x, поэтому в противном случае метод наименьших квадратов не работает. Однако я получаю следующее предупреждение:
Код: Выделить всё
UserWarning: X has feature names, but MinMaxScaler was fitted without feature names
warnings.warn(
Если это актуально, вот код, который читается в необработанные данные и определяет поля raw_data и класс data_entry:
Код: Выделить всё
import pandas as pd
import os
import numpy as np
class DataEntry:
def __init__(self, helium_number, analysis_label, timestamp, PAB_data, raw_data):
self.helium_number = helium_number
self.analysis_label = analysis_label
self.timestamp = timestamp
self.PAB_data = PAB_data
self.raw_data = raw_data
self.analysis_type = self.determine_analysis_type(analysis_label)
self.sequence_number = None
self.data_status = {}
for mass in raw_data:
column_name = raw_data[mass].columns[1]
self.data_status[mass] = pd.DataFrame(np.ones(raw_data[mass].shape[0], dtype=int), columns=[column_name])
def determine_analysis_type(self, analysis_label):
if analysis_label.startswith('Q'):
return 'Q'
elif analysis_label.startswith('DT'):
return 'D'
elif analysis_label.startswith('PCT'):
return 'PCT'
elif analysis_label.startswith('LB'):
return 'lineblank'
elif analysis_label.startswith('CB'):
return 'coldblank'
elif analysis_label.startswith('HB'):
return 'hotblank'
else:
return 'unknown'
def identify_sequences(all_data):
batch_num = 1 # Start with the first batch
previous_timestamp = None
sequence_data = []
for analysis in all_data:
if not previous_timestamp:
analysis.sequence_number = batch_num
previous_timestamp = analysis.timestamp
sequence_data.append({'sequence_number': batch_num, 'start_time': analysis.timestamp})
continue
if (analysis.timestamp - previous_timestamp) > pd.Timedelta(hours=2):
batch_num += 1
sequence_data[-1]['end_time'] = previous_timestamp
sequence_data.append({'sequence_number': batch_num, 'start_time': analysis.timestamp})
analysis.sequence_number = batch_num
previous_timestamp = analysis.timestamp
sequence_data[-1]['end_time'] = previous_timestamp
return all_data, sequence_data
def parse_raw(file_path):
# import the data file as df
df = pd.read_csv(file_path, sep='\t', header=1)
# check if the data file is empty
if df.shape[0]
Подробнее здесь: [url]https://stackoverflow.com/questions/78422718/python-plot-not-updating-on-interactive-click[/url]