Eduardo_Scripniciuc/st-ten-1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto back up upon save

Browse Source
This commit is contained in:
edo-neo 2025-09-15 10:34:15 +02:00
parent ac5eae8241
commit e537721aca
3 changed files with 77 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
src/lib/helpers/recipe_manager.py
View File

@ -3,7 +3,7 @@ import csv
import locale
from datetime import datetime
import shutil
import socket
import re
from PyQt5.QtCore import pyqtSignal, QObject
from PyQt5.QtWidgets import QFileDialog
@ -131,7 +131,7 @@ def read_steps(row, config, defaults=None, unsupported_steps=None):
"chan_sel": safe_parse(row.get("canale_di_prova_2", defaults["canale_di_prova_2"])),
"ext_flush_time": safe_parse(row.get("tempo_svuotamento_esterno_2", defaults["tempo_svuotamento_esterno"])),
"ext_blow_time": safe_parse(row.get("tempo_soffiaggio_esterno_2", defaults["tempo_soffiaggio_esterno"])),
"pid_pressure_correction": safe_parse(row.get("pid_pressure_correction_2", defaults["pid_pressure_correction_2"])),
"pid_pressure_correction": safe_parse(row.get("pid_pressure_correction", defaults["pid_pressure_correction_2"])),
"pid_mod_config": safe_parse(row.get("pid_mod_config", defaults["pid_mod_config"])),
},
"vision": {
@ -288,66 +288,7 @@ def import_recipes(config, csv_path=None, defaults=None, unsupported_steps=None,
logger.error(f"Error importing recipes: {e}")
raise
FIELDNAMES = [
"codice_ricetta",
"cliente",
"part_number",
"dimensione_lotto_abilitata",
"dimensione_lotto",
"verifica_connettore_abilitata",
"connettore",
"verifica_codice_a_barre_abilitata",
"codice_a_barre",
"verifica_resistenza_connettore_abilitata",
"scala_resistenza",
"r nominale",
"tolleranza_resistenza_pos",
"tolleranza_resistenza_neg",
"avvitatura_abilitata",
"viti",
"prova_tenuta_abilitata",
"tempo_pre_riempimento",
"pressione_pre_riempimento",
"tempo_riempimento",
"tempo_assestamento",
"percentuale_minima_pressione_assestamento",
"percentuale_massima_pressione_assestamento",
"tempo_di_test",
"pressione_di_test_delta_minimo",
"pressione_di_test",
"pressione_di_test_delta_massimo",
"pid_pressure_correction",
"tempo_svuotamento",
"pressione_svuotamento",
"prova_tenuta_abilitata_2",
"tempo_pre_riempimento_2",
"pressione_pre_riempimento_2",
"tempo_riempimento_2",
"tempo_assestamento_2",
"percentuale_minima_pressione_assestamento_2",
"percentuale_massima_pressione_assestamento_2",
"tempo_di_test_2",
"pressione_di_test_delta_minimo_2",
"pressione_di_test_2",
"pressione_di_test_delta_massimo_2",
"pid_pressure_correction_2",
"tempo_svuotamento_2",
"pressione_svuotamento_2",
"test_visione_abilitato",
"ricetta_visione",
"stampa_etichetta_abilitata",
"modello_etichetta",
"labeltxt_1",
"labeltxt_2",
"labeltxt_3",
"labeltxt_4",
"labeltxt_5"
]
def export_recipes(config, csv_path=None, logger=None):
# ... (QFileDialog logic remains the same) ...
if csv_path is None:
options = QFileDialog.Options()
options |= QFileDialog.DontUseNativeDialog
@ -374,29 +315,31 @@ def export_recipes(config, csv_path=None, logger=None):
barcode_serial_field = config.get("recipe", {}).get("barcode_serial_field", "codice_a_barre").strip()
print_template_field = config.get("recipe", {}).get("label_template_field", "modello_etichetta").strip()
data = []
fieldnames = set() # Use a set to avoid duplicates
# Wrap database operations in a transaction for consistency
with db.atomic():
# Iterate over all recipes in the database
for recipe in Recipes.select():
try:
steps = recipe.get_steps_map()
# Create a dictionary with default values for all fields
exportable = {field: "" for field in FIELDNAMES}
# Populate the dictionary with recipe data
exportable.update({
exportable = {
# Base fields
recipe_name_field: recipe.name,
"cliente": recipe.client,
"part_number": recipe.part_number,
})
}
# Conditionally update the dictionary for each step
# Add base fields to the fieldnames
fieldnames.update([recipe_name_field, "cliente", "part_number"])
# Check and add steps conditionally
if "connector" in steps:
exportable.update({
"verifica_connettore_abilitata": "x",
"connettore": steps["connector"].spec["connector"]
})
fieldnames.update(["verifica_connettore_abilitata", "connettore"])
if "resistance" in steps:
exportable.update({
@ -406,18 +349,22 @@ def export_recipes(config, csv_path=None, logger=None):
"tolleranza_resistenza_pos": steps["resistance"].spec["tolerance_pos"],
"tolleranza_resistenza_neg": steps["resistance"].spec["tolerance_neg"],
})
fieldnames.update(["verifica_resistenza_connettore_abilitata", "scala_resistenza", "r nominale",
"tolleranza_resistenza_pos", "tolleranza_resistenza_neg"])
if "barcodes" in steps:
exportable.update({
barcode_enable_field: "x",
barcode_serial_field: steps["barcodes"].spec["serial"]
})
fieldnames.update([barcode_enable_field, barcode_serial_field])
if "screws" in steps:
exportable.update({
"avvitatura_abilitata": "x",
"viti": steps["screws"].spec["quantity"]
})
fieldnames.update(["avvitatura_abilitata", "viti"])
if "leak_1" in steps:
exportable.update({
@ -428,6 +375,8 @@ def export_recipes(config, csv_path=None, logger=None):
"pressione_di_test": steps["leak_1"].spec["test_pressure"],
"pid_pressure_correction": steps["leak_1"].spec["pid_pressure_correction"],
})
fieldnames.update(["prova_tenuta_abilitata", "tempo_pre_riempimento", "pressione_pre_riempimento",
"tempo_di_test", "pressione_di_test", "pid_pressure_correction"])
if "leak_2" in steps:
exportable.update({
@ -438,30 +387,29 @@ def export_recipes(config, csv_path=None, logger=None):
"pressione_di_test_2": steps["leak_2"].spec["test_pressure"],
"pid_pressure_correction": steps["leak_1"].spec["pid_pressure_correction"],
})
fieldnames.update(["prova_tenuta_abilitata_2", "tempo_pre_riempimento_2", "pressione_pre_riempimento_2",
"tempo_di_test_2", "pressione_di_test_2", "pid_pressure_correction"])
if "vision" in steps:
exportable.update({
"test_visione_abilitato": steps["vision"].spec.get("enabled", ""),
"ricetta_visione": steps["vision"].spec["recipe"]
})
fieldnames.update(["test_visione_abilitato", "ricetta_visione"])
if "print" in steps:
exportable.update({
"stampa_etichetta_abilitata": "x",
print_template_field: steps["print"].spec.get("template", ""),
# Add the labeltxt fields here
"labeltxt_1": steps["print"].spec.get("labeltxt_1", ""),
"labeltxt_2": steps["print"].spec.get("labeltxt_2", ""),
"labeltxt_3": steps["print"].spec.get("labeltxt_3", ""),
"labeltxt_4": steps["print"].spec.get("labeltxt_4", ""),
"labeltxt_5": steps["print"].spec.get("labeltxt_5", ""),
print_template_field: steps["print"].spec["template"],
})
fieldnames.update(["stampa_etichetta_abilitata", print_template_field])
# Append the exportable row to the data
data.append(exportable)
except Exception as e:
if logger:
logger.error(f"Error processing recipe {recipe.name}: {e}")
# Continue with next recipe instead of failing the entire export
continue
except Exception as e:
if logger:
@ -472,8 +420,8 @@ def export_recipes(config, csv_path=None, logger=None):
if len(data):
if logger:
logger.info(f"Exporting recipes to {csv_path}")
with open(csv_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
with open(csv_path, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=list(fieldnames))
writer.writeheader()
writer.writerows(data)
if logger:
@ -482,41 +430,36 @@ def export_recipes(config, csv_path=None, logger=None):
def backup_current_recipes(config, logger=None):
"""
Back up current recipes to a CSV file in a single common folder. The file name equals the
[machine]/description from the active machine config (sanitized). Saving overwrites any
previous backup with the same machine description.
Back up current recipes to a CSV file named after the current machine description.
Only one backup file is kept (overwritten on each call).
"""
# Get machine description from config; fall back to hostname if missing
machine_desc = None
try:
machine_desc = (config.get("machine", {}) or {}).get("description")
except Exception:
machine_desc = None
if not machine_desc:
machine_desc = socket.gethostname()
# Sanitize description to be safe as a file name
safe_name = str(machine_desc).strip()
# Replace path separators and common forbidden characters on Windows/Unix
for ch in ['\\', '/', ':', '*', '?', '"', '<', '>', '|']:
safe_name = safe_name.replace(ch, '_')
# Also collapse consecutive spaces
safe_name = ' '.join(safe_name.split())
# Ensure .csv extension
if not safe_name.lower().endswith('.csv'):
safe_name = f"{safe_name}.csv"
# Define the single backup directory and file path
# Define the backup directory
backup_dir = os.path.join('config', 'csv_import', 'backup_csv')
backup_path = os.path.join(backup_dir, safe_name)
# Read machine description from config
try:
machine_desc = (config.get('machine', {}) or {}).get('description')
if not machine_desc:
# Fallbacks
machine_desc = getattr(config, 'machine_id', None) or 'backup_recipes'
except Exception:
machine_desc = getattr(config, 'machine_id', None) or 'backup_recipes'
# Sanitize description to create a safe filename
safe_desc = re.sub(r"[^A-Za-z0-9._-]+", "_", str(machine_desc).strip())
if not safe_desc:
safe_desc = 'backup_recipes'
# Build backup file path (no timestamp => single rotating file)
backup_file = f"{safe_desc}.csv"
backup_path = os.path.join(backup_dir, backup_file)
# Ensure the backup directory exists
os.makedirs(backup_dir, exist_ok=True)
# Export current recipes to the backup path (export_recipes overwrites the file)
export_recipes(config=config, csv_path=backup_path, logger=logger)
if logger:
logger.info(f"Backup created at: {backup_path}")
# Export current recipes to the backup path (overwrites existing file)
# Suppress internal export logs during automatic backup
export_recipes(config=config, csv_path=backup_path, logger=None)
# Do not log here to avoid duplicate messages; caller will handle final log
return backup_path # Return the backup path for reference if needed

6
src/ui/crud/crud.py
View File

@ -246,6 +246,7 @@ class Json_External_Dialog_Editor_Cell_Widget(QPushButton, Cell):
class Crud(Widget):
modified = pyqtSignal(bool)
selected = pyqtSignal(object)
committed = pyqtSignal()
def __init__(self, table_name, readonly=False, select=None, filters=None, fields_aliases=None, autocomplete=None, sort=None, pagination=250, display_name=None, row_upgrader=None, widget_classes=None, row_filter=None):
super().__init__()
@ -549,6 +550,11 @@ class Crud(Widget):
# INDEX DATA WITH PK
try:
self.db.commit(data, deleted_rows=self.deleted_rows)
# Emit committed signal to notify successful save
try:
self.committed.emit()
except Exception:
pass
except Exception as e:
self.log.exception(traceback.format_exc())
QMessageBox.critical(None, "Errore Salvataggio DB", str(e))

21
src/ui/recipe_selection/recipe_selection.py
View File

@ -11,7 +11,7 @@ from PyQt5.QtGui import QKeySequence
from PyQt5.QtWidgets import QFileDialog, QMessageBox, QShortcut
import shutil
from lib.helpers.recipe_manager import export_recipes, import_recipes,recipe_manager_signals
from lib.helpers.recipe_manager import export_recipes, import_recipes, recipe_manager_signals, backup_current_recipes
from lib.helpers.step import Step
from ui.crud import Crud, Json_External_Dialog_Editor_Cell_Widget
from ui.helpers import replace_widget
@ -175,6 +175,11 @@ class Recipe_Selection(Widget):
pagination=25,
)
replace_widget(self, "crud_w", self.crud)
# Backup recipes automatically on successful save in CRUD
try:
self.crud.committed.connect(self.on_crud_committed)
except Exception:
pass
self.crud_modified = None
self.selected = None
self.select_b.setEnabled(False)
@ -219,6 +224,20 @@ class Recipe_Selection(Widget):
recipe_manager_signals.recipes_imported.connect(self.crud.refresh)
def on_crud_committed(self):
"""Triggered after successful save (commit) in the CRUD UI: creates a timestamped CSV backup."""
try:
backup_path = backup_current_recipes(config=self.config, logger=self.log)
try:
self.log.info(f"Backup CSV created: {backup_path}")
except Exception:
pass
except Exception as e:
try:
self.log.exception(f"Failed to create backup CSV after commit: {e}")
except Exception:
pass
def check_modified(self, modified):
self.crud_modified = modified
self.check(self.crud_modified, self.selected)