ssd-logging-daemon.py

Ordinateur industriel monocarte BPI-5202 Loongson 2K1000LAi · Fichier : ssd-logging-daemon.py · Langage : Python · Usage : Enregistrer les données de capteurs sur le stockage SSD avec rotation automatique pour éviter la saturation.

Script Python gérant l'écriture de logs JSON sur le SSD avec une politique de rotation de fichiers.

Assurez-vous que le SSD est monté sur /mnt/ssd. Le BPI-5202 supporte les SSD pour un stockage fiable. Ce script évite la corruption de données en cas de coupure de courant grâce à l'écriture séquentielle, mais nécessite un système de fichiers journalisé (ext4) recommandé.

ssd-logging-daemon.py

import os
import json
import time
from datetime import datetime
import shutil

LOG_DIR = '/mnt/ssd/logs'
MAX_SIZE = 10 * 1024 * 1024 # 10 Mo par fichier
MAX_FILES = 10

# Vérifier et créer le répertoire sur le SSD monté
if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)
    print(f"Répertoire {LOG_DIR} créé.")

def log_data(sensor_id, value):
    timestamp = datetime.now().isoformat()
    entry = {"timestamp": timestamp, "sensor": sensor_id, "value": value}
    
    # Nom de fichier basé sur la date pour faciliter la rotation
    filename = f"data_{datetime.now().strftime('%Y%m%d')}.json"
    filepath = os.path.join(LOG_DIR, filename)
    
    # Vérifier la taille du fichier
    if os.path.exists(filepath) and os.path.getsize(filepath) > MAX_SIZE:
        # Rotation simple : renommer avec un suffixe
        backup_name = f"{filename}.bak.{int(time.time())}"
        os.rename(filepath, os.path.join(LOG_DIR, backup_name))
        print(f"Rotation effectuée: {backup_name}")
        
    # Écrire les données
    with open(filepath, 'a') as f:
        f.write(json.dumps(entry) + '\n')
        
    # Nettoyage des anciens fichiers de sauvegarde si trop nombreux
    files = [f for f in os.listdir(LOG_DIR) if f.endswith('.bak')]
    if len(files) > MAX_FILES:
        files.sort(key=lambda x: os.path.getctime(os.path.join(LOG_DIR, x)))
        for f in files[:len(files)-MAX_FILES]:
            os.remove(os.path.join(LOG_DIR, f))

if __name__ == "__main__":
    print("Démarrage du logger SSD...")
    counter = 0
    while True:
        # Simulation de données de capteur
        log_data("TempSensor_01", 25.5 + (counter % 10) * 0.1)
        counter += 1
        time.sleep(5)

← Retour à la fiche