# /// script
# requires-python = ">=3.13"
# dependencies = [
# "keras==3.8.0",
# "marimo",
# "numpy==2.2.2",
# ]
# ///
import marimo
__generated_with = "0.10.17"
app = marimo.App(width="medium")
@app.cell(hide_code=True)
def title():
import marimo as mo
notebook_name = 'train_perfspec.py'
from lib_perfspec import perfspec_vars
(_,_defs) = perfspec_vars.run()
perfspec = _defs['perfspec']
from lib_perfspec import perfspec_header
(_,_defs) = perfspec_header.run()
lib_header = _defs['header']
lib_intro = _defs['intro']
mo.md(
f"""
{lib_header(notebook_name)}
## Train **{perfspec['app']['train_mode']}** model
"""
)
return (
lib_header,
lib_intro,
mo,
notebook_name,
perfspec,
perfspec_header,
perfspec_vars,
)
@app.cell(hide_code=True)
def imports():
import tensorflow as tf
import numpy as np
from pathlib import Path
import keras
return Path, keras, np, tf
@app.cell(hide_code=True)
def intro_load(Path, lib_intro, mo, notebook_name, perfspec):
verbose = perfspec['settings']['verbose']
perfspec['vars'] = {}
from lib_perfspec import perfspec_args
(_,_defs) = perfspec_args.run()
if not Path(perfspec['defaults']['models_dirpath']).exists():
exit(f"Trained models dir path not found: {perfspec['defaults']['models_dirpath']}")
if not Path(perfspec['defaults']['checkpoints_dirpath']).exists():
exit(f"Trained checkpoints models dir path not found: {perfspec['defaults']['checkpoints_dirpath']}")
if not Path(perfspec['defaults']['data_dirpath']).exists():
exit(f"data dir path not found: {perfspec['defaults']['data_dirpath']}")
from lib_perfspec import perfspec_load_actions
(_,_defs) = perfspec_load_actions.run()
lib_load_actions = _defs['load_actions']
from lib_perfspec import perfspec_input_sequence
(_,_defs) = perfspec_input_sequence.run()
lib_get_input_sequence = _defs['get_input_sequence']
from lib_perfspec import perfspec_predict
_, _defs = perfspec_predict.run()
lib_predict_action = _defs['predict_action']
verbose=perfspec['settings'].get('verbose')
perfspec['vars']['model'] = None
perfspec['vars']['history'] = None
(perfspec['vars']['actions'],
perfspec['vars']['unique_actions'],
perfspec['vars']['label_encoder'],
perfspec['vars']['encoded_actions']
) = lib_load_actions(
actions_path=perfspec['settings'].get('actions_filepath'),
verbose=verbose
)
perfspec['vars']['input_sequence'] = lib_get_input_sequence(
input_str=perfspec['settings']['input_str'],
unique_actions=perfspec['vars']['unique_actions']
)
mo.md(
f"""
{lib_intro(notebook_name)}
"""
)
return (
lib_get_input_sequence,
lib_load_actions,
lib_predict_action,
perfspec_args,
perfspec_input_sequence,
perfspec_load_actions,
perfspec_predict,
verbose,
)
@app.cell(hide_code=True)
def setting(mo, notebook_name, perfspec):
from lib_perfspec import perfspec_out_settings
(_,_defs) = perfspec_out_settings.run()
out_settings = _defs['out_settings']
mo.md(
f"""
Settings are defined in: [lib_perfspec.py]({mo.notebook_dir()} / lib_perfspec.py)
{out_settings(notebook_name)}
Total of values from data: **{len(perfspec['vars']['actions'])}**
with total of unique values: **{len(perfspec['vars']['unique_actions'])}**
> Train size {perfspec['defaults']['train_size']} to train model: **{
int(perfspec['defaults']['train_size'] * len(perfspec['vars']['actions']))
}** with sequence of **{perfspec['defaults']['sequence_length']}**
Values can be overwritten by using **command-line** (see options below).
"""
)
return out_settings, perfspec_out_settings
@app.cell(hide_code=True)
def command_line(mo, notebook_name):
from lib_perfspec import perfspec_cli_ops
(_,_defs) = perfspec_cli_ops.run()
out_cli_ops = _defs['out_cli_ops']
mo.accordion({
"Mostrar command Line options ": out_cli_ops(notebook_name)
})
return out_cli_ops, perfspec_cli_ops
@app.cell(hide_code=True)
def perfspec_define_confusion_matrix(
load_model_from_path,
mo,
np,
prepare_train,
):
def make_confusion_matrix(perfspec):
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.preprocessing.sequence import pad_sequences
from sklearn.preprocessing import LabelEncoder
from sklearn.metrics import confusion_matrix
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
import matplotlib.pyplot as plt
import seaborn as sns
if perfspec['vars']['model'] == None:
model = load_model_from_path(perfspec['settings']['verbose'])
if perfspec['vars'].get('model') != None:
print("No model found")
return
else:
model = perfspec['vars']['model']
(perfspec['vars']['X'],perfspec['vars']['y'],perfspec['vars']['_y']) = prepare_train(perfspec)
#X = []
#y = []
#sequence_length = perfspec['settings']['sequence_length']
# Generate input-output pairs
#for j in range(len(perfspec['vars']['encoded_actions']) - sequence_length):
# X.append(perfspec['vars']['encoded_actions'][j:j + sequence_length]) # Input sequence
# y.append(perfspec['vars']['encoded_actions'][j + sequence_length]) # Target (next action)
#X = np.array(X)
#y = np.array(y)
## Ensure _X has the correct shape for LSTM
#X = pad_sequences(X, maxlen=perfspec['settings']['sequence_length'], padding='pre')
#X = np.expand_dims(X, axis=-1) # Shape: (num_samples, sequence_length, 1)
X = perfspec['vars']['X']
y = perfspec['vars']['_y']
encoder = LabelEncoder()
y_encoded = encoder.fit_transform(y)
#y_encoded = encoder.fit_transform(perfspec['vars']['unique_actions'])
#y_encoded = encoded_input
# Split the dataset into training and testing sets
X_train, X_test, y_train, y_test = train_test_split(X, y_encoded, test_size=0.2, random_state=42)
# Predict with your model (replace this with actual predictions)
y_pred = model.predict(X_test) # This returns the predicted probabilities
#y_pred = predicted_probabilities
y_pred_classes = np.argmax(y_pred, axis=1) # Get class predictions from probabilities
# Plot confusion matrix
cm = confusion_matrix(y_test, y_pred_classes)
ax = sns.heatmap(cm, annot=True, fmt="d", cmap="Blues", xticklabels='auto', yticklabels='auto')
ax.set_xticklabels(ax.get_xticklabels(), fontsize=7)
ax.set_yticklabels(ax.get_yticklabels(), fontsize=7)
plt.title("Confusion Matrix")
plt.xlabel("Predicted Labels")
plt.ylabel("True Labels")
plt.show()
mo.md(
r"""
### Define Confusion Matrix
"""
)
return (make_confusion_matrix,)
@app.cell(hide_code=True)
def consfusion_matrix_callback(
ConfusionMatrixDisplay,
confusion_matrix,
keras,
mo,
np,
plt,
tf,
):
@keras.saving.register_keras_serializable(package="Custom")
class ConfusionMatrixCallback(tf.keras.callbacks.Callback):
def __init__(self, val_data, label_encoder=None, display=False):
"""
Custom callback to compute and display a confusion matrix at the end of each epoch.
Parameters:
- val_data: Tuple (X_val, y_val) for validation data.
- label_encoder: Optional, if labels are encoded, to map back to original labels.
- display: Whether to display the confusion matrix as a plot.
"""
super().__init__()
self.val_data = val_data
self.label_encoder = label_encoder
self.display = display
def on_epoch_end(self, epoch, logs=None):
# Get the validation data
val_X, val_y = self.val_data
# Predict the classes for validation data
y_pred = self.model.predict(val_X, verbose=0)
y_pred_classes = np.argmax(y_pred, axis=1)
# If label encoding is provided, decode the labels
if self.label_encoder:
val_y = self.label_encoder.inverse_transform(val_y)
y_pred_classes = self.label_encoder.inverse_transform(y_pred_classes)
# Compute the confusion matrix
cm = confusion_matrix(val_y, y_pred_classes)
# Optionally display or save the confusion matrix
if self.display:
self.plot_confusion_matrix(cm, epoch)
# Print the confusion matrix to console
print(f"Epoch {epoch + 1}: Confusion Matrix\n{cm}")
def plot_confusion_matrix(self, cm, epoch):
"""
Plot and display the confusion matrix using matplotlib.
"""
plt.figure(figsize=(8, 8))
disp = ConfusionMatrixDisplay(confusion_matrix=cm)
disp.plot(cmap='viridis', values_format='d')
plt.title(f'Confusion Matrix - Epoch {epoch + 1}')
plt.show()
def get_config(self):
# Return the configuration of the metric as a dictionary
config = super().get_config()
return config
mo.md(
r"""
### Confusion Matrix Callback
"""
)
return (ConfusionMatrixCallback,)
@app.cell(hide_code=True)
def precsion_metric(F1Score, keras, mo, tf):
@keras.saving.register_keras_serializable(package="Custom")
class PrecisionMetric(tf.keras.metrics.Metric):
def __init__(self, name="precision", **kwargs):
super(PrecisionMetric, self).__init__(name=name, **kwargs)
self.true_positives = self.add_weight(name="tp", initializer="zeros")
self.false_positives = self.add_weight(name="fp", initializer="zeros")
def update_state(self, y_true, y_pred, sample_weight=None):
y_pred = tf.argmax(y_pred, axis=1)
y_true = tf.cast(y_true, tf.int32)
y_pred = tf.cast(y_pred, tf.int32)
true_positives = tf.reduce_sum(tf.cast(tf.equal(y_true, y_pred) & tf.equal(y_true, 1), tf.float32))
false_positives = tf.reduce_sum(tf.cast(tf.not_equal(y_true, y_pred) & tf.equal(y_pred, 1), tf.float32))
self.true_positives.assign_add(true_positives)
self.false_positives.assign_add(false_positives)
def result(self):
return self.true_positives / (self.true_positives + self.false_positives + tf.keras.backend.epsilon())
def reset_states(self):
self.true_positives.assign(0.0)
self.false_positives.assign(0.0)
def get_config(self):
# Return the configuration of the metric as a dictionary
config = super().get_config()
return config
def get_config(self):
base_config = super(F1Score, self).get_config()
return base_config
@classmethod
def from_config(cls, config):
return cls(**config)
mo.md(
r"""
### Precision Metric
**Precision** measures how many of the predicted positive labels are actually correct. It is calculated as:
\[
\text{Precision} = \frac{\text{True Positives}}{\text{True Positives} + \text{False Positives}}
\]
- **True Positives (TP)**: Cases where the model predicted `1` (positive) and the actual label is also `1`.
- **False Positives (FP)**: Cases where the model predicted `1` (positive), but the actual label is not `1`.
- The logical conditions check if predictions match and filter for the positive class (`1`).
- These counts are accumulated over batches.
"""
)
return (PrecisionMetric,)
@app.cell(hide_code=True)
def recall_metric(keras, mo, tf):
@keras.saving.register_keras_serializable(package="Custom")
class RecallMetric(tf.keras.metrics.Metric):
def __init__(self, name="recall", **kwargs):
super(RecallMetric, self).__init__(name=name, **kwargs)
self.true_positives = self.add_weight(name="tp", initializer="zeros")
self.false_negatives = self.add_weight(name="fn", initializer="zeros")
def update_state(self, y_true, y_pred, sample_weight=None):
y_pred = tf.argmax(y_pred, axis=1)
y_true = tf.cast(y_true, tf.int32)
y_pred = tf.cast(y_pred, tf.int32)
true_positives = tf.reduce_sum(tf.cast(tf.equal(y_true, y_pred) & tf.equal(y_true, 1), tf.float32))
false_negatives = tf.reduce_sum(tf.cast(tf.not_equal(y_true, y_pred) & tf.equal(y_true, 1), tf.float32))
self.true_positives.assign_add(true_positives)
self.false_negatives.assign_add(false_negatives)
def result(self):
return self.true_positives / (self.true_positives + self.false_negatives + tf.keras.backend.epsilon())
def reset_states(self):
self.true_positives.assign(0.0)
self.false_negatives.assign(0.0)
def get_config(self):
# Return the configuration of the metric as a dictionary
config = super().get_config()
return config
@classmethod
def from_config(cls, config):
return cls(**config)
mo.md(
r"""
### Define Recall Mectric
**Recall** measures how many actual positive labels were correctly predicted. It is calculated as:
\[
\text{Recall} = \frac{\text{True Positives}}{\text{True Positives} + \text{False Negatives}}
\]
- **False Negatives (FN)**: Cases where the model predicted `0` (negative), but the actual label is `1`.
- This is similar to precision but focuses on capturing all actual positives.
"""
)
return (RecallMetric,)
@app.cell(hide_code=True)
def f1_score_metric(PrecisionMetric, RecallMetric, keras, mo, tf):
@keras.saving.register_keras_serializable(package="Custom")
class F1ScoreMetric(tf.keras.metrics.Metric):
def __init__(self, name="f1_score", **kwargs):
super(F1ScoreMetric, self).__init__(name=name, **kwargs)
self.precision = PrecisionMetric()
self.recall = RecallMetric()
def update_state(self, y_true, y_pred, sample_weight=None):
self.precision.update_state(y_true, y_pred, sample_weight)
self.recall.update_state(y_true, y_pred, sample_weight)
def result(self):
precision = self.precision.result()
recall = self.recall.result()
return 2 * (precision * recall) / (precision + recall + tf.keras.backend.epsilon())
def reset_states(self):
self.precision.reset_states()
self.recall.reset_states()
def get_config(self):
# Return the configuration of the metric as a dictionary
config = super().get_config()
# config = super(F1Score, self).get_config()
return config
@classmethod
def from_config(cls, config):
return cls(**config)
mo.md(
r"""
### Define F1 Score Metric
F1-score balances precision and recall, especially useful when the dataset is imbalanced. It is the harmonic mean of precision and recall:
\[
\text{F1-Score} = 2 \cdot \frac{\text{Precision} \cdot \text{Recall}}{\text{Precision} + \text{Recall}}
\]
- `epsilon` is added to avoid division by zero.
"""
)
return (F1ScoreMetric,)
@app.cell(hide_code=True)
def custom_validation_metrics(mo, tf):
#Custom callback to compute metrics on validation data
class CustomValidationMetrics(tf.keras.callbacks.Callback):
def __init__(self, X_val, y_val):
super().__init__() # Initialize the parent class
self.X_val = X_val
self.y_val = y_val
def on_epoch_end(self, epoch, logs=None):
val_predictions = self.model.predict(self.X_val, verbose=0)
val_predictions = (val_predictions > 0.5).astype(int) # Binarize predictions
# Compute precision, recall, and f1-score
precision = tf.keras.metrics.Precision()(self.y_val, val_predictions)
recall = tf.keras.metrics.Recall()(self.y_val, val_predictions)
f1_score = 2 * (precision * recall) / (precision + recall + 1e-7)
print(f"\nEpoch {epoch + 1} Validation Metrics - Precision: {precision:.4f}, Recall: {recall:.4f}, F1 Score: {f1_score:.4f}")
mo.md(
r"""
### Custom Validation Metrics
To apply metrics only to the validation set in Keras, via custom callback that computes metrics on the validation data at the end of each epoch.
Keras does not seems to have a natively support specifying metrics exclusively for validation data in the compile() method
"""
)
return (CustomValidationMetrics,)
@app.cell(hide_code=True)
def perfspec_prepare_model_train(mo, np):
def prepare_train(perfspec):
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.preprocessing.sequence import pad_sequences
#from sklearn.preprocessing import LabelEncoder
# Parameters
X = []
y = []
sequence_length = perfspec['settings']['sequence_length']
# Generate input-output pairs
for j in range(len(perfspec['vars']['encoded_actions']) - sequence_length):
X.append(perfspec['vars']['encoded_actions'][j:j + sequence_length]) # Input sequence
y.append(perfspec['vars']['encoded_actions'][j + sequence_length]) # Target (next action)
X = np.array(X)
y = np.array(y)
# Ensure _X has the correct shape for LSTM
X = pad_sequences(X, maxlen=sequence_length, padding='pre')
X = np.expand_dims(X, axis=-1) # Shape: (num_samples, sequence_length, 1)
_y = y
vocab_size =len(perfspec['vars']['label_encoder'].classes_) # Total number of unique actions
# One-hot encode _y for classification
y = to_categorical(y, num_classes=vocab_size)
return (X,y,_y)
mo.md(
r"""
## Prepare train model input
Load input and **shape X,y** for model
"""
)
return (prepare_train,)
@app.cell(hide_code=True)
def show_train_model_shape(mo, perfspec, prepare_train, verbose):
(perfspec['vars']['X'], perfspec['vars']['y'],perfspec['vars']['_y']) = prepare_train(perfspec)
if verbose != None or mo.cli_args().get("verbose") != None or mo.running_in_notebook():
print(f"X shape : {perfspec['vars']['X'].shape}")
print(f"y shape : {perfspec['vars']['y'].shape}")
mo.md(
r"""
### Show train model shape
"""
)
return
@app.cell(hide_code=True)
def make_model(CustomValidationMetrics, mo, np, perfspec):
# Define the LSTM model
def make_model(X=[],y=[],label_encoder=[], encoded_actions=[]):
if len(X) == 0 or len(y) == 0:
print ("make_model: No values fond for X y i")
return
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Input, LSTM, Dense, Embedding, Dropout
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import (
EarlyStopping,
ReduceLROnPlateau,
Callback,
ModelCheckpoint,
)
vocab_size = len(
label_encoder.classes_
) # Get the number of unique action labels
# Model architecture
embedding_dim = 50
#input_shape = (perfspec['settings']['input_length'],perfspec['settings']['input_length'],vocab_size),
perfspec['vars']['model'] = Sequential(
[
#Embedding(input_dim=vocab_size, output_dim=embedding_dim),
Input(shape=(perfspec['settings']['sequence_length'], 1)),
LSTM(
perfspec['settings']['lstm_units_1'],
return_sequences=True,
recurrent_dropout=perfspec['settings']['dropout_rate'],
input_shape=(perfspec['settings']['sequence_length'], 1),
),
LSTM(
perfspec['settings']['lstm_units_2'],
return_sequences=False,
recurrent_dropout=perfspec['settings']['dropout_rate'],
),
Dropout(0.5),
Dense(
vocab_size, activation="softmax"
), # Softmax for multi-class classification
]
)
# Split the data for training and validation
train_size = int(perfspec['defaults']['train_size'] * len(X))
train_X, val_X = X[:train_size], X[train_size:]
train_y, val_y = y[:train_size], y[train_size:]
if train_y.ndim > 1 and train_y.shape[-1] == vocab_size:
loss_function = "categorical_crossentropy"
else:
# Convert targets to integers if needed
train_y = np.argmax(train_y, axis=-1) if train_y.ndim > 1 else train_y
val_y = np.argmax(val_y, axis=-1) if val_y.ndim > 1 else val_y
loss_function = "sparse_categorical_crossentropy"
# Compile the model
perfspec['vars']['model'].compile(
optimizer=Adam(),
loss=loss_function,
metrics=[
"accuracy"
# PrecisionMetric(),
# RecallMetric(),
# F1ScoreMetric()
],
)
# Create confusion matrix callback
#confusion_callback = ConfusionMatrixCallback(
# val_data=(val_X, val_y),
# label_encoder=label_encoder,
# display=True
#)
# Callbacks
early_stopping = EarlyStopping(
monitor="val_loss", patience=10, restore_best_weights=True
)
lr_reduction = ReduceLROnPlateau(
monitor="val_loss", patience=8, factor=0.8, min_lr=0.0001
)
custom_metrics_callback = CustomValidationMetrics(X, y)
if perfspec['settings']['checkpoint_mode'] == "weights":
# Save only the weights of the model instead of the full model.
checkpoint = ModelCheckpoint(
filepath=perfspec['settings']['checkpoint_filepath'].as_posix(),
save_best_only=True, # True to save only when validation loss improves
monitor='val_loss', # Metric to monitor
mode='min', # Minimize the validation loss
verbose=1 # Print messages when saving
)
elif perfspec['settings']['checkpoint_mode'] == "epochs":
# Define to collect save checkpoint at the end of every epoch via callback
#_checkpoint_filepath = checkpoint_dirpath + '/' + 'model_at_epoch_{epoch:02d}.h5'
checkpoint = ModelCheckpoint(
filepath=perfspec['settings']['checkpoint_filepath'].as_posix(),
save_best_only=False, # True to save only when validation loss improves
verbose=1 # Print messages when saving
)
callbacks=[early_stopping,lr_reduction]
callbacks=[early_stopping,lr_reduction]
callbacks.append(custom_metrics_callback)
if checkpoint != None:
callbacks.append(checkpoint)
# Print the model summary
perfspec['vars']['model'].summary()
print (" train size: {}".format(train_size))
print (" train_X size: {}".format(len(train_X)))
print (" train_y size: {}".format(len(train_y)))
print (" val_X size: {}".format(len(train_X)))
print (" val_y size: {}".format(len(train_y)))
# Train the model
perfspec['vars']['history'] = perfspec['vars']['model'].fit(
train_X,
train_y,
batch_size=perfspec['settings']['batch_size'],
epochs=perfspec['defaults']['epochs'],
validation_data=(val_X, val_y),
verbose=2,
callbacks=callbacks
)
return
mo.md(
r"""
## Make Model train
This is where **model** is creates and **fit**
Saved in `perfspec['vars']` as `model` and `history`
"""
)
return (make_model,)
@app.cell(hide_code=True)
def main(mo):
mo.md(r"""""")
return
@app.cell(hide_code=True)
def use_or_make_model(
Path,
load_model_from_path,
make_model,
mo,
perfspec,
verbose,
):
if Path(perfspec['settings']['model_filepath']).exists():
if mo.running_in_notebook() or (verbose != None or mo.cli_args().get("verbose") != None):
print (f"Trained model path already exist, to train model DELETE existing path: {perfspec['settings']['model_filepath']}")
perfspec['vars']['model']=load_model_from_path(perfspec,perfspec['settings']['verbose'])
else:
if perfspec['vars'].get('X').any() and perfspec['vars'].get('y').any():
make_model(
X=perfspec['vars']['X'],
y=perfspec['vars']['y'],
label_encoder=perfspec['vars']['label_encoder'],
encoded_actions=perfspec['vars']['encoded_actions']
)
#evaluate_model(the_model,X,y)
mo.md(
f"""
### Use or Train Model
**IMPORTANT** if model_filepathmodel_filepath not exists it will be created otherwise it will be loaded form model_filepathmodel_filepath
Path: "🍃" {perfspec['settings']['model_filepath']}
"""
)
return
@app.cell(hide_code=True)
def perfspec_load_model_from_path(Path, mo, perfspec):
def load_model_from_path(perfspec,verbose):
from keras.models import load_model
custom_objects = {
# "PrecisionMetric": PrecisionMetric,
# "RecallMetric": RecallMetric,
# "F1ScoreMetric": F1ScoreMetric,
}
if Path(perfspec['settings']['model_filepath']).exists():
perfspec['vars']['model'] = load_model(
perfspec['settings']['model_filepath'],
#custom_objects=custom_objects
)
if verbose != None and not mo.running_in_notebook():
print (f"Trained model loaded from: {perfspec['settings']['model_filepath']}")
return perfspec['vars']['model']
else:
return None
mo.md(
f"""
### Load model from path
Load trained model from file **model_filepath** 🍃
Path: {perfspec['settings']['model_filepath']}
"""
)
return (load_model_from_path,)
@app.cell(hide_code=True)
def perfspec_save_model(Path, mo, perfspec):
def save_model():
if not Path(perfspec['settings']['model_filepath']).exists() and perfspec['vars']['model'] != None:
# Save the model
print (f"Save trained model to: {perfspec['settings']['model_filepath']}")
perfspec['vars']['model'].save(perfspec['settings']['model_filepath'])
if perfspec['vars']['history'] != None:
import json
try:
with open(perfspec['settings']['model_history_filepath'], 'w') as output_file:
json.dump(perfspec['vars']['history'].history,output_file)
print (f"Save trained model history to: {perfspec['settings']['model_history_filepath']}")
except IOError as e:
print(f"Error writing to file: {e}")
except json.JSONDecodeError as e:
print(f"Error encoding JSON: {e}")
save_model()
mo.md(
f"""
## Save Model
to `model_filepath` "🍃" {perfspec['settings']['model_filepath']}
Path can be changed via **command-line** with **--model** `model-filepath`
"""
)
return (save_model,)
@app.cell(hide_code=True)
def perfspec_plot_history(Path, mo):
def plot_history(perfspec):
import json
from keras.src.callbacks import History
if 'vars' not in perfspec:
return None
if perfspec['vars']['history'] != None:
if isinstance(perfspec['vars']['history'], History):
_model_history = perfspec['vars']['history'].history
else:
_model_history = perfspec['vars']['history']
else:
if not Path(perfspec['settings']['model_history_filepath']).exists():
print(f"Not found: {perfspec['settings']['model_history_filepath']}")
_model_history = {}
else:
with open(perfspec['settings']['model_history_filepath']) as history_file:
perfspec['vars']['history'] = json.load(history_file)
_model_history = perfspec['vars']['history']
return _model_history
mo.md(
f"""
## Plot Model history
"""
)
#if 'plt_accuracy' in perfspec['vars']: mo.as_html(perfspec['vars']['plt_accuracy'].gcf())
return (plot_history,)
@app.cell(hide_code=True)
def perfspec_plot_defs(mo, model_history, plot_history):
def plot_accuracy(perfspec):
_model_history = plot_history(perfspec)
if _model_history != None and mo.running_in_notebook():
import matplotlib.pyplot as plt
import seaborn as sns
if _model_history.get('accuracy'):
plt.style.use('dark_background')
# Plot training & validation accuracy values
(_, acc) = plt.subplots()
acc.plot(_model_history['accuracy'],color='white')
acc.plot(_model_history['val_accuracy'], color='orange')
acc.set_title('Model Accuracy')
acc.set_xlabel('Epochs')
acc.set_ylabel('Accuracy')
acc.legend(['Train', 'Test'], loc='upper left')
return plt
elif perfspec['settings']['verbose'] == 2:
if 'accuracy' in _model_history:
print ('model val_accuracy')
for idx,it in enumerate(_model_history.get('val_accuracy')):
print(f"epoch: {idx} value: {it}")
return None
def plot_loss(perfspec):
_model_history = plot_history(perfspec)
if _model_history != None and mo.running_in_notebook():
import matplotlib.pyplot as plt
import seaborn as sns
if _model_history.get('loss'):
plt.style.use('dark_background')
(_, loss) = plt.subplots()
loss.plot(_model_history['loss'])
loss.plot(_model_history['val_loss'])
loss.set_title('Model Loss')
loss.set_label('Epochs')
loss.set_ylabel('Loss')
loss.legend(['Train', 'Test'], loc='upper left')
return plt
elif perfspec['settings']['verbose'] == 2:
if 'val_loss' in _model_history:
print ('model val_loss')
for idx,it in enumerate(_model_history.get('val_loss')):
print(f"epoch: {idx} value: {it}")
return None
def plot_precision(perfspec):
_model_history = plot_history(perfspec)
if _model_history != None and mo.running_in_notebook():
import matplotlib.pyplot as plt
import seaborn as sns
if _model_history.get('precision'):
plt.style.use('dark_background')
# Plot training & validation accuracy values
plt.plot(model_history['learning_rate'])
plt.plot(model_history['precision'])
plt.plot(model_history['recall'])
plt.plot(model_history['f1_score'])
plt.title('Model learning')
plt.xlabel('Epochs')
plt.ylabel('Learning')
plt.legend(['Rate','Precision','Recall', 'f1_score'], loc='upper left')
return plt
else:
return None
return plot_accuracy, plot_loss, plot_precision
@app.cell(hide_code=True)
def pefspec_ploc_accuracy(mo, perfspec, plot_accuracy):
_plot_acc=plot_accuracy(perfspec)
if perfspec['vars']['history'] != None and mo.running_in_notebook():
_output = mo.as_html(_plot_acc.gcf())
else:
_output = None
mo.md(
f"""
## Model Accuracy history
From model train plot accuracy and epochs
{_output}
"""
)
return
@app.cell(hide_code=True)
def pefspec_plot_loss(mo, perfspec, plot_loss):
_plot_loss = plot_loss(perfspec)
if perfspec['vars']['history'] != None and mo.running_in_notebook():
_output = mo.as_html(_plot_loss.gcf())
else:
_output = None
mo.md(
f"""
## Model loss history
From model train loss
{_output}
"""
)
return
@app.cell(hide_code=True)
def perfspec_plot_precision(mo, perfspec, plot_precision):
_plot_pre = plot_precision(perfspec)
if perfspec['vars']['history'] != None and mo.running_in_notebook():
mo.md(
f"""
From model train plot Precision
{mo.as_html(_plot_pre.gcf())}
"""
)
return
@app.cell(hide_code=True)
def perfspec_confusion_matric(
load_model_from_path,
make_confusion_matrix,
mo,
perfspec,
):
if mo.running_in_notebook():
if perfspec['vars'].get('model') == None:
load_model_from_path(perfspec['settings']['verbose'])
if perfspec['vars'].get('model') != None:
make_confusion_matrix(perfspec)
mo.md("### Confusion Matrix")
return
@app.cell(hide_code=True)
def perfspec_evaluate_model(Path, mo, np, prepare_train):
def evaluate_model(test_input=[], test_labels=[],model=None):
if model == None:
return
if len(test_input) > 0:
#Evaluate the model on test data
#(test_loss, test_precision, test_recall, test_f1_score) = perfspec['vars']['model'].evaluate(test_input, test_labels)
(test_loss,test_accuracy) = model.evaluate(test_input, test_labels)
print(f"Test Accuracy: {"{:2.4f}".format(test_accuracy)}")
print(f"Test Loss: {"{:2.4f}".format(test_loss)}")
#print(f"Precision: {test_precision}")
#print(f"Recall: {test_recall}")
#print(f"F1 Score: {test_f1_score}")
def run_evaluate(perfspec):
# load_trained_model()
if perfspec['vars']['model'] == None:
return
(perfspec['vars']['X'], perfspec['vars']['y'], perfspec['vars']['_y']) = prepare_train(perfspec)
print ("\nEvaluate Model")
X=perfspec['vars']['X'],
y=perfspec['vars']['y'],
train_size = int(perfspec['defaults']['train_size'] * len(X))
train_X, val_X = X[:train_size], X[train_size:]
train_y, val_y = y[:train_size], y[train_size:]
evaluate_model(val_X,val_y,perfspec['vars']['model'])
def history_info(perfspec):
import json
from keras.src.callbacks import History
if perfspec['vars']['history'] != None:
if isinstance(perfspec['vars']['history'], History):
model_history = perfspec['vars']['history'].history
else:
model_history = perfspec['vars']['history']
else:
if not Path(perfspec['settings']['model_history_filepath']).exists():
print(f"Not found: {perfspec['settings']['model_history_filepath']}")
model_history = {}
else:
with open(perfspec['settings']['model_history_filepath']) as history_file:
model_history = json.load(history_file)
if model_history != None:
from prettytable import PrettyTable
train_loss = model_history['loss']
val_loss = model_history['val_loss']
train_acc = model_history['accuracy']
val_acc = model_history['val_accuracy']
table_find = PrettyTable()
# Find the epoch with the minimum validation loss
best_epoch_val_loss = np.argmin(val_loss) # The index of the minimum validation loss
#Find the epoch with the maximum validation accuracy
best_epoch_val_acc = np.argmax(val_acc) # The index of the maximum validation accuracy
#print ("\nFrom Model History")
table_find.field_names = ["description", "value"]
table_find.align["description"] = "l"
table_find.align["value"] = "r"
table_find.border = True
table_find.add_rows([
["Best Epoch (Validation Loss)",(best_epoch_val_loss + 1)],
["Validation Loss at Best Epoch","{:2.4f}".format(val_loss[best_epoch_val_loss])],
["Best Epoch (Validation Accuracy)",(best_epoch_val_acc + 1)],
["Validation Accuracy at Best Epoch","{:2.4f}".format(val_acc[best_epoch_val_acc])]
])
if mo.running_in_notebook() :
return table_find.get_formatted_string('html')
else:
return table_find.get_formatted_string('text')
else:
return ""
return evaluate_model, history_info, run_evaluate
@app.cell(hide_code=True)
def perfspec_run_evaluate_model_run(
history_info,
mo,
perfspec,
run_evaluate,
):
#with mo.capture_stdout() as buffer:
# run_evaluate(perfspec)
#_output = buffer.getvalue()
_output=run_evaluate(perfspec)
#_history_info = history_info(perfspec)
mo.md(
f"""
## Evaluate Model
{mo.md(history_info((perfspec)))}
"""
)
return
@app.cell(hide_code=True)
def perfspec_test_model_prediction(
lib_get_input_sequence,
lib_predict_action,
load_model_from_path,
mo,
perfspec,
verbose,
):
_verbose = "1" if mo.running_in_notebook() else perfspec['settings']['verbose']
_input_sequence = lib_get_input_sequence(perfspec['settings']['input_str'],perfspec['vars']['unique_actions'])
if len(_input_sequence) > 0:
if perfspec['vars']['model'] != None:
_model = perfspec['vars']['model']
else:
_model = load_model_from_path(perfspec,verbose)
if _model != None:
(encoded_input,predicted_probabilities) = lib_predict_action(
_model,
perfspec['settings']['sequence_length'],
_input_sequence,
perfspec['vars']['label_encoder'],
_verbose
)
mo.md(
r"""
## Test Model Prediction
Use **{input_sequence}** with trained model created or loaded
from {model_filepath}
input value can be changed in **command-line** with **--input** `value` argument
with **--verbose** option more info is show in **command-line** mode
"""
)
return encoded_input, predicted_probabilities
if __name__ == "__main__":
app.run()