Autoencoder as a costom learning block

Help
#1

It sounds like you’re working on a really interesting project with anomaly detection using autoencoders in Edge Impulse! It’s frustrating to hit those errors when you’re so close.

Here’s a question you can post on the Edge Impulse forum to get some help, along with the key points you’ll want to highlight:

Edge Impulse Forum Question: Custom Learning Block for Autoencoder Anomaly Detection

Hello Edge Impulse Community,

I’m currently working on implementing an autoencoder for anomaly detection as a custom learning block in Edge Impulse, and I’m encountering some challenges.

My goal is to train the autoencoder on normal data only, then use the reconstruction error to classify anomalies. Specifically, I want to establish a threshold based on the normal data’s reconstruction errors. During inference, if the reconstruction error of a new data point exceeds this threshold, it should be flagged as anomalous; otherwise, it should be considered normal.

I’ve successfully integrated my autoencoder model into the Edge Impulse interface. However, I’m facing two main issues:

  1. Output Interpretation: The custom learning block currently seems to interpret the output as “classes” rather than a continuous reconstruction error or a single anomaly score. How can I configure the custom learning block to output a numerical reconstruction error (or a similar anomaly score) that I can then use for thresholding, instead of a categorical class prediction?
  2. Thresholding and Classification: What’s the recommended approach within Edge Impulse for applying a post-processing threshold to the autoencoder’s output (reconstruction error) to classify data as “normal” or “anomaly”? Is there a way to integrate this thresholding logic directly into the custom learning block, or should this be handled in a separate processing step (e.g., in the impulse or a custom inference script)?

Any guidance on how to correctly set up a custom learning block for this type of autoencoder-based anomaly detection, particularly regarding output configuration and thresholding, would be greatly appreciated. Are there specific considerations or best practices I should be aware of when implementing this?

Thank you for your time and help!

#2

parameters.json file is contunue like this:

{
“version”: 1,
“type”: “machine-learning”,
“info”: {
“name”: “autoencoder”,
“description”: “Anomaly detection using a custom autoencoder”,
“operatesOn”: “other”,
“learningBlockType”: “anomaly”,
“indRequiresGpu”: false
},
“parameters”: [
{
“name”: “Batch size”,
“value”:“512”,
“type”: “int”,
“help”: "The number of samples processed before the model is updated. Larger values can speed up training but may use more me> “param”:“batch_size”
},

train.py

— ARGUMENT PARSING —

parser = argparse.ArgumentParser(description=“Autoencoder training script.”)
#parser.add_argument(‘–data-directory’, type=str, default=‘data’,help=‘Path to the directory containing normal_train_data.npy and test_data.npy’)
parser.add_argument(‘–data-directory’, type=str, required=True)

#parser.add_argument(‘–epochs’, type=int, default=20, help=‘Number of training epochs’)
#parser.add_argument(‘–learning-rate’, type=float, default=0.001, help=‘Learning rate for the optimizer’)
#parser.add_argument(‘–validation-set-size’, type=float, default=0.2, help=‘Validation split fraction’)
#parser.add_argument(‘–input-shape’, type=str, default=‘(140,)’, help=‘Shape of the input data’)
#parser.add_argument(‘–batch_size’, type=int, required=True)

parser.add_argument(‘–epochs’, type=int, required=True)
parser.add_argument(‘–learning-rate’, type=float, required=True)
parser.add_argument(‘–batch-size’, type=int, required=False, default=512)

Let’s add the model-path argument, which specifies where we will save the model.

#parser.add_argument(‘–model-path’, type=str, default=‘model.keras’,

help=‘Path to save the trained autoencoder model (.keras format).’)

parser.add_argument(‘–out-directory’, type=str, required=True)

#parser.add_argument(‘–out-directory’, type=str, default=‘output’)
args = parser.parse_args()

if not os.path.exists(args.out_directory):
os.mkdir(args.out_directory)

#Loading data
#train_data_path = os.path.join(args.data_directory, ‘normal_train_data.npy’)
#test_data_path = os.path.join(args.data_directory, ‘test_data.npy’)

#print(f"Looking for training data at: {train_data_path}“)
#print(f"Looking for test data at: {test_data_path}”)
train_data_path = os.path.join(args.data_directory, ‘X_split_train.npy’)
test_data_path = os.path.join(args.data_directory, ‘X_split_test.npy’)

try:
#normal_train_data = np.load(train_data_path)
#test_data = np.load(test_data_path)
normal_train_data = np.load(train_data_path, mmap_mode=‘r’)
test_data = np.load(test_data_path, mmap_mode=‘r’)
print(“Data loaded successfully!”)
except FileNotFoundError as e:
print(f"Error: One or more data files not found. Check --data-directory path. {e}")

— MODEL —

class AnomalyDetector(Model):
def init(self):
super(AnomalyDetector, self).init()
self.encoder = tf.keras.Sequential([
layers.Dense(32, activation=“relu”),
layers.Dense(16, activation=“relu”),
layers.Dense(8, activation=“relu”)
])
self.decoder = tf.keras.Sequential([
layers.Dense(16, activation=“relu”),
layers.Dense(32, activation=“relu”),
layers.Dense(140, activation=“sigmoid”)
])

def call(self, x):
    encoded = self.encoder(x)
    decoded = self.decoder(encoded)
    return decoded

autoencoder = AnomalyDetector()
autoencoder.compile(optimizer=‘adam’, loss=‘mean_squared_error’)

— TRAIN —

print(f"[INFO] Using batch size: {args.batch_size}“)
print(f”[INFO] Using epochs: {args.epochs}")

history = autoencoder.fit(normal_train_data, normal_train_data,
epochs=args.epochs,
batch_size=args.batch_size,
validation_data=(test_data, test_data),
shuffle=True
)

reconstructions = autoencoder.predict(normal_train_data)
train_loss = tf.keras.losses.mae(reconstructions, normal_train_data)

threshold = np.mean(train_loss) + np.std(train_loss)
print(f"[INFO] Threshold that we recommend to you is : {threshold}")

metrics = {
‘threshold’: float(threshold) # numpy float’ı standart float’a çevirmek için
}

metrics_path = os.path.join(args.out_directory, ‘metrics.json’)

with open(metrics_path, ‘w’) as f:
json.dump(metrics, f, indent=4)

print(f"Metrics (including threshold) saved to {metrics_path}")

#3

when i use parser.add_argument(‘–data-directory’, type=str, required=True) in train.py file
the studio says: usage: train.py [-h] --data-directory DATA_DIRECTORY [–numLayers NUMLAYERS]
–epochs EPOCHS --learning-rate LEARNING_RATE
[–batch-size BATCH_SIZE] --out-directory OUT_DIRECTORY
train.py: error: the following arguments are required: --data-directory
i can solve it with adding default = data, but i am not sure if it is true solution.