676 lines
27 KiB
Python
676 lines
27 KiB
Python
"""Window used to inspect and logically empty an entire warehouse aisle.
|
|
|
|
The tool summarizes the current occupancy of one aisle and, after explicit
|
|
confirmation, unloads every active UDC through the same logical movement
|
|
semantics used by the rest of the WMS.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import sys
|
|
import tkinter as tk
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
from tkinter import messagebox, simpledialog, ttk
|
|
from typing import Any
|
|
|
|
import customtkinter as ctk
|
|
|
|
from busy_overlay import InlineBusyOverlay
|
|
from gestione_aree import AsyncRunner
|
|
from gestione_scarico import move_pallet_async
|
|
from locale_text import load_locale_catalog, text as loc_text
|
|
from tooltips import WidgetToolTip, load_tooltip_catalog, tooltip_text
|
|
from ui_theme import theme_color, theme_font, theme_padding, theme_section, theme_value
|
|
from window_placement import place_window_fullsize_below_parent_later
|
|
|
|
try:
|
|
from loguru import logger
|
|
except Exception: # pragma: no cover - fallback used only when Loguru is not available
|
|
class _FallbackLogger:
|
|
"""Minimal adapter used only when Loguru is not installed yet."""
|
|
|
|
def __init__(self):
|
|
self._logger = logging.getLogger(MODULE_LOG_NAME if "MODULE_LOG_NAME" in globals() else __name__)
|
|
self._logger.setLevel(logging.DEBUG)
|
|
self._logger.propagate = False
|
|
|
|
def bind(self, **_kwargs):
|
|
return self
|
|
|
|
def add(self, sink, level="INFO", format=None, encoding="utf-8", **_kwargs):
|
|
handler: logging.Handler
|
|
if hasattr(sink, "write"):
|
|
handler = logging.StreamHandler(sink)
|
|
else:
|
|
handler = logging.FileHandler(str(sink), encoding=encoding)
|
|
handler.setLevel(getattr(logging, str(level).upper(), logging.INFO))
|
|
handler.setFormatter(
|
|
logging.Formatter("%(asctime)s | %(levelname)-8s | %(name)s | %(message)s")
|
|
)
|
|
self._logger.addHandler(handler)
|
|
return 0
|
|
|
|
def log(self, level, message):
|
|
getattr(self._logger, str(level).lower(), self._logger.info)(message)
|
|
|
|
def debug(self, message):
|
|
self._logger.debug(message)
|
|
|
|
def info(self, message):
|
|
self._logger.info(message)
|
|
|
|
def exception(self, message):
|
|
self._logger.exception(message)
|
|
|
|
logger = _FallbackLogger()
|
|
|
|
|
|
RESET_CORSIE_LOG_MODE = "INFO" # "OFF" | "INFO" | "DEBUG"
|
|
MODULE_LOG_NAME = Path(__file__).stem
|
|
MODULE_LOG_PATH = Path(__file__).with_suffix(".log")
|
|
_MODULE_LOG_ENABLED = RESET_CORSIE_LOG_MODE.upper() != "OFF"
|
|
_MODULE_LOG_LEVEL = "DEBUG" if RESET_CORSIE_LOG_MODE.upper() == "DEBUG" else "INFO"
|
|
_MODULE_LOGGER = logger.bind(warehouse_module=MODULE_LOG_NAME)
|
|
_MODULE_LOGGING_CONFIGURED = False
|
|
|
|
|
|
def _configure_module_logger():
|
|
"""Configure console and file logging for this module."""
|
|
global _MODULE_LOGGING_CONFIGURED
|
|
if _MODULE_LOGGING_CONFIGURED:
|
|
return
|
|
if not _MODULE_LOG_ENABLED:
|
|
_MODULE_LOGGING_CONFIGURED = True
|
|
return
|
|
|
|
record_filter = lambda record: record["extra"].get("warehouse_module") == MODULE_LOG_NAME
|
|
|
|
logger.add(
|
|
sys.stderr,
|
|
level=_MODULE_LOG_LEVEL,
|
|
colorize=True,
|
|
filter=record_filter,
|
|
format=(
|
|
"<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
|
|
"<level>{level: <8}</level> | "
|
|
"<cyan>" + MODULE_LOG_NAME + "</cyan> | "
|
|
"<level>{message}</level>"
|
|
),
|
|
)
|
|
logger.add(
|
|
MODULE_LOG_PATH,
|
|
level=_MODULE_LOG_LEVEL,
|
|
colorize=False,
|
|
encoding="utf-8",
|
|
filter=record_filter,
|
|
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | " + MODULE_LOG_NAME + " | {message}",
|
|
)
|
|
_MODULE_LOGGING_CONFIGURED = True
|
|
|
|
|
|
def _format_payload(payload: Any) -> str:
|
|
"""Serialize payloads for human-readable logging."""
|
|
try:
|
|
return json.dumps(payload, ensure_ascii=False, indent=2, default=str)
|
|
except Exception:
|
|
return repr(payload)
|
|
|
|
|
|
def _log_call(level: str | None = None):
|
|
"""Trace entry, exit and failure of selected high-level functions."""
|
|
|
|
def decorator(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
effective_level = level or _MODULE_LOG_LEVEL
|
|
_MODULE_LOGGER.log(
|
|
effective_level,
|
|
f"CALL {func.__qualname__} args={_format_payload(args[1:] if args else ())} kwargs={_format_payload(kwargs)}",
|
|
)
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
except Exception:
|
|
_MODULE_LOGGER.exception(f"FAIL {func.__qualname__}")
|
|
raise
|
|
_MODULE_LOGGER.log(effective_level, f"RETURN {func.__qualname__}")
|
|
return result
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def _log_sql(query_name: str, sql: str, params: dict[str, Any] | None = None):
|
|
"""Log one SQL statement and its parameters."""
|
|
_MODULE_LOGGER.log(_MODULE_LOG_LEVEL, f"SQL {query_name} params={_format_payload(params or {})}")
|
|
_MODULE_LOGGER.debug(f"SQL {query_name} text:\n{sql.strip()}")
|
|
|
|
|
|
def _log_dataset(query_name: str, rows: list[Any]):
|
|
"""Log query results at summary or full-debug level depending on the flag."""
|
|
_MODULE_LOGGER.log(_MODULE_LOG_LEVEL, f"SQL {query_name} returned {len(rows)} rows")
|
|
if RESET_CORSIE_LOG_MODE.upper() == "DEBUG":
|
|
_MODULE_LOGGER.debug(f"SQL {query_name} dataset:\n{_format_payload(rows)}")
|
|
|
|
|
|
_configure_module_logger()
|
|
if _MODULE_LOG_ENABLED:
|
|
_MODULE_LOGGER.info(
|
|
f"Logging inizializzato su {MODULE_LOG_PATH.name} livello={_MODULE_LOG_LEVEL} mode={RESET_CORSIE_LOG_MODE.upper()}"
|
|
)
|
|
|
|
|
|
SQL_CORSIE = """
|
|
WITH C AS (
|
|
SELECT DISTINCT LTRIM(RTRIM(Corsia)) AS Corsia
|
|
FROM dbo.Celle
|
|
WHERE ID <> 9999 AND (DelDataOra IS NULL) AND LTRIM(RTRIM(Corsia)) <> '7G'
|
|
)
|
|
SELECT Corsia
|
|
FROM C
|
|
ORDER BY
|
|
CASE
|
|
WHEN LEFT(Corsia,3)='MAG' AND TRY_CONVERT(int, SUBSTRING(Corsia,4,50)) IS NOT NULL THEN 0
|
|
WHEN TRY_CONVERT(int, Corsia) IS NOT NULL THEN 1
|
|
ELSE 2
|
|
END,
|
|
CASE WHEN LEFT(Corsia,3)='MAG' THEN TRY_CONVERT(int, SUBSTRING(Corsia,4,50)) END,
|
|
CASE WHEN TRY_CONVERT(int, Corsia) IS NOT NULL THEN TRY_CONVERT(int, Corsia) END,
|
|
CASE WHEN TRY_CONVERT(int, Corsia) IS NOT NULL THEN SUBSTRING(Corsia, LEN(CAST(TRY_CONVERT(int, Corsia) AS varchar(20)))+1, 50) END,
|
|
Corsia;
|
|
"""
|
|
|
|
SQL_RIEPILOGO = """
|
|
WITH C AS (
|
|
SELECT ID, LTRIM(RTRIM(Corsia)) AS Corsia,
|
|
LTRIM(RTRIM(Colonna)) AS Colonna,
|
|
LTRIM(RTRIM(Fila)) AS Fila
|
|
FROM dbo.Celle
|
|
WHERE ID <> 9999 AND (DelDataOra IS NULL)
|
|
AND LTRIM(RTRIM(Corsia)) = :corsia
|
|
),
|
|
S AS (
|
|
SELECT c.ID, COUNT(DISTINCT g.BarcodePallet) AS n
|
|
FROM C AS c LEFT JOIN dbo.XMag_GiacenzaPallet AS g ON g.IDCella = c.ID
|
|
GROUP BY c.ID
|
|
)
|
|
SELECT
|
|
COUNT(*) AS TotCelle,
|
|
SUM(CASE WHEN s.n>0 THEN 1 ELSE 0 END) AS CelleOccupate,
|
|
SUM(CASE WHEN s.n>1 THEN 1 ELSE 0 END) AS CelleDoppie,
|
|
SUM(COALESCE(s.n,0)) AS TotPallet
|
|
FROM C LEFT JOIN S s ON s.ID = C.ID;
|
|
"""
|
|
|
|
SQL_DETTAGLIO = """
|
|
WITH C AS (
|
|
SELECT ID, LTRIM(RTRIM(Corsia)) AS Corsia,
|
|
LTRIM(RTRIM(Colonna)) AS Colonna,
|
|
LTRIM(RTRIM(Fila)) AS Fila
|
|
FROM dbo.Celle
|
|
WHERE ID <> 9999 AND (DelDataOra IS NULL)
|
|
AND LTRIM(RTRIM(Corsia)) = :corsia
|
|
),
|
|
S AS (
|
|
SELECT c.ID, COUNT(DISTINCT g.BarcodePallet) AS n
|
|
FROM C AS c LEFT JOIN dbo.XMag_GiacenzaPallet AS g ON g.IDCella = c.ID
|
|
GROUP BY c.ID
|
|
)
|
|
SELECT c.ID AS IDCella,
|
|
CONCAT(c.Corsia, '.', c.Colonna, '.', c.Fila) AS Ubicazione,
|
|
COALESCE(s.n,0) AS NumUDC
|
|
FROM C c LEFT JOIN S s ON s.ID = c.ID
|
|
WHERE COALESCE(s.n,0) > 0
|
|
ORDER BY TRY_CONVERT(int,c.Colonna), c.Colonna, TRY_CONVERT(int,c.Fila), c.Fila;
|
|
"""
|
|
|
|
SQL_COUNT_RESET = """
|
|
SELECT
|
|
COUNT(DISTINCT g.BarcodePallet) AS TotUDC,
|
|
COUNT(DISTINCT g.IDCella) AS TotCelle
|
|
FROM dbo.XMag_GiacenzaPallet g
|
|
JOIN dbo.Celle c ON c.ID = g.IDCella
|
|
WHERE c.ID <> 9999
|
|
AND LTRIM(RTRIM(c.Corsia)) = :corsia;
|
|
"""
|
|
|
|
SQL_UDC_RESET = """
|
|
WITH U AS (
|
|
SELECT DISTINCT
|
|
g.BarcodePallet AS BarcodePallet,
|
|
g.IDCella AS IDCella,
|
|
CONCAT(LTRIM(RTRIM(c.Corsia)), '.', LTRIM(RTRIM(c.Colonna)), '.', LTRIM(RTRIM(c.Fila))) AS Ubicazione,
|
|
TRY_CONVERT(int, c.Colonna) AS SortColNum,
|
|
LTRIM(RTRIM(c.Colonna)) AS SortColTxt,
|
|
TRY_CONVERT(int, c.Fila) AS SortFilaNum,
|
|
LTRIM(RTRIM(c.Fila)) AS SortFilaTxt
|
|
FROM dbo.XMag_GiacenzaPallet g
|
|
JOIN dbo.Celle c ON c.ID = g.IDCella
|
|
WHERE c.ID <> 9999
|
|
AND LTRIM(RTRIM(c.Corsia)) = :corsia
|
|
)
|
|
SELECT
|
|
BarcodePallet,
|
|
IDCella,
|
|
Ubicazione
|
|
FROM U
|
|
ORDER BY
|
|
SortColNum,
|
|
SortColTxt,
|
|
SortFilaNum,
|
|
SortFilaTxt,
|
|
BarcodePallet;
|
|
"""
|
|
|
|
|
|
class ResetCorsieWindow(ctk.CTkToplevel):
|
|
"""Toplevel used to inspect and clear the pallets assigned to an aisle."""
|
|
|
|
@_log_call()
|
|
def __init__(self, parent, db_client, session=None):
|
|
"""Create the window and immediately load the list of aisles."""
|
|
super().__init__(parent)
|
|
self._theme = theme_section("reset_corsie", {})
|
|
self._locale_catalog = load_locale_catalog()
|
|
self.title(loc_text("reset.title", catalog=self._locale_catalog, default="Gestione Corsie - svuotamento celle per corsia"))
|
|
self.geometry(str(theme_value(self._theme, "window_geometry", "1000x680")))
|
|
minsize = theme_value(self._theme, "window_minsize", [880, 560])
|
|
self.minsize(int(minsize[0]), int(minsize[1]))
|
|
self.resizable(True, True)
|
|
try:
|
|
self.configure(fg_color=theme_color(self._theme, "window_fg_color", ("#efefef", "#2f2f2f")))
|
|
except Exception:
|
|
pass
|
|
|
|
self.db = db_client
|
|
self.session = session
|
|
self._busy = InlineBusyOverlay(self, self._theme)
|
|
self._async = AsyncRunner(self)
|
|
self._refresh_token = 0
|
|
self._tooltip_catalog = load_tooltip_catalog()
|
|
|
|
self._build_ui()
|
|
self._load_corsie()
|
|
|
|
def _setup_tree_style(self):
|
|
"""Apply a denser, spreadsheet-like style to the main result grid."""
|
|
|
|
style = ttk.Style(self)
|
|
style.configure(
|
|
"ResetCorsie.Treeview.Heading",
|
|
font=theme_font(self._theme, "tree_heading_font", ("Segoe UI", 10, "bold")),
|
|
background=theme_value(self._theme, "tree_heading_bg", "#b8c7db"),
|
|
foreground=theme_value(self._theme, "tree_heading_fg", "#10243e"),
|
|
relief="flat",
|
|
padding=theme_padding(self._theme, "tree_heading_padding", (8, 6)),
|
|
)
|
|
style.map(
|
|
"ResetCorsie.Treeview.Heading",
|
|
background=[("active", theme_value(self._theme, "tree_heading_bg_active", "#aebfd6"))],
|
|
relief=[("pressed", "groove"), ("!pressed", "flat")],
|
|
)
|
|
style.configure(
|
|
"ResetCorsie.Treeview",
|
|
font=theme_font(self._theme, "tree_body_font", ("Segoe UI", 10)),
|
|
rowheight=int(theme_value(self._theme, "tree_row_height", 30)),
|
|
background=theme_value(self._theme, "tree_body_bg", "#ffffff"),
|
|
fieldbackground=theme_value(self._theme, "tree_body_bg", "#ffffff"),
|
|
foreground=theme_value(self._theme, "tree_body_fg", "#1f1f1f"),
|
|
borderwidth=0,
|
|
)
|
|
style.map(
|
|
"ResetCorsie.Treeview",
|
|
background=[("selected", theme_value(self._theme, "tree_selected_bg", "#cfe4ff"))],
|
|
foreground=[("selected", theme_value(self._theme, "tree_selected_fg", "#10243e"))],
|
|
)
|
|
|
|
@_log_call()
|
|
def _build_ui(self):
|
|
"""Create selectors, summary widgets and the occupied-cell grid."""
|
|
self._setup_tree_style()
|
|
top = ctk.CTkFrame(self)
|
|
top.pack(
|
|
fill="x",
|
|
padx=int(theme_value(self._theme, "frame_padx", 8)),
|
|
pady=int(theme_value(self._theme, "frame_pady", 8)),
|
|
)
|
|
try:
|
|
top.configure(fg_color=theme_color(self._theme, "top_frame_fg_color", ("#d7d7d7", "#3b3b3b")))
|
|
except Exception:
|
|
pass
|
|
ctk.CTkLabel(
|
|
top,
|
|
text=loc_text("reset.label.aisle", catalog=self._locale_catalog, default="Corsia:"),
|
|
font=theme_font(self._theme, "toolbar_label_font", ("Segoe UI", 10)),
|
|
).pack(side="left")
|
|
self.cmb = ctk.CTkComboBox(
|
|
top,
|
|
width=int(theme_value(self._theme, "combobox_width", 140)),
|
|
height=int(theme_value(self._theme, "combobox_height", 28)),
|
|
values=[],
|
|
font=theme_font(self._theme, "combobox_font", ("Segoe UI", 10)),
|
|
dropdown_font=theme_font(self._theme, "combobox_font", ("Segoe UI", 10)),
|
|
)
|
|
self.cmb.pack(side="left", padx=(6, 10))
|
|
btn_refresh = ctk.CTkButton(
|
|
top,
|
|
text=loc_text("reset.button.refresh", catalog=self._locale_catalog, default="Carica"),
|
|
command=self.refresh,
|
|
width=int(theme_value(self._theme, "toolbar_button_width", 140)),
|
|
height=int(theme_value(self._theme, "toolbar_button_height", 28)),
|
|
corner_radius=int(theme_value(self._theme, "toolbar_button_corner_radius", 6)),
|
|
font=theme_font(self._theme, "toolbar_button_font", ("Segoe UI", 10, "bold")),
|
|
)
|
|
btn_refresh.pack(side="left")
|
|
btn_reset = ctk.CTkButton(
|
|
top,
|
|
text=loc_text("reset.button.empty", catalog=self._locale_catalog, default="Svuota corsia..."),
|
|
command=self._ask_reset,
|
|
width=int(theme_value(self._theme, "toolbar_button_width", 140)),
|
|
height=int(theme_value(self._theme, "toolbar_button_height", 28)),
|
|
corner_radius=int(theme_value(self._theme, "toolbar_button_corner_radius", 6)),
|
|
font=theme_font(self._theme, "toolbar_button_font", ("Segoe UI", 10, "bold")),
|
|
)
|
|
btn_reset.pack(side="right")
|
|
WidgetToolTip(btn_refresh, tooltip_text("reset_corsie.refresh", catalog=self._tooltip_catalog))
|
|
WidgetToolTip(btn_reset, tooltip_text("reset_corsie.empty_aisle", catalog=self._tooltip_catalog))
|
|
|
|
mid = ctk.CTkFrame(self)
|
|
mid.pack(
|
|
fill="both",
|
|
expand=True,
|
|
padx=int(theme_value(self._theme, "frame_padx", 8)),
|
|
pady=(0, int(theme_value(self._theme, "frame_pady", 8))),
|
|
)
|
|
try:
|
|
mid.configure(fg_color=theme_color(self._theme, "mid_frame_fg_color", ("#e5e5e5", "#383838")))
|
|
except Exception:
|
|
pass
|
|
mid.grid_columnconfigure(0, weight=1)
|
|
mid.grid_rowconfigure(0, weight=1)
|
|
|
|
self.tree = ttk.Treeview(
|
|
mid,
|
|
columns=("Ubicazione", "NumUDC"),
|
|
show="headings",
|
|
selectmode="browse",
|
|
style="ResetCorsie.Treeview",
|
|
)
|
|
self.tree.heading("Ubicazione", text="Ubicazione")
|
|
self.tree.heading("NumUDC", text="UDC in cella")
|
|
self.tree.column(
|
|
"Ubicazione",
|
|
width=int(theme_value(self._theme, "tree_col_ubicazione_width", 340)),
|
|
anchor=str(theme_value(self._theme, "tree_col_ubicazione_anchor", "center")),
|
|
)
|
|
self.tree.column(
|
|
"NumUDC",
|
|
width=int(theme_value(self._theme, "tree_col_num_udc_width", 160)),
|
|
anchor=str(theme_value(self._theme, "tree_col_num_udc_anchor", "center")),
|
|
)
|
|
self.tree.tag_configure("odd", background=theme_value(self._theme, "tree_row_odd_bg", "#ffffff"))
|
|
self.tree.tag_configure("even", background=theme_value(self._theme, "tree_row_even_bg", "#f3f6fb"))
|
|
|
|
sy = ttk.Scrollbar(mid, orient="vertical", command=self.tree.yview)
|
|
sx = ttk.Scrollbar(mid, orient="horizontal", command=self.tree.xview)
|
|
self.tree.configure(yscrollcommand=sy.set, xscrollcommand=sx.set)
|
|
self.tree.grid(row=0, column=0, sticky="nsew")
|
|
sy.grid(row=0, column=1, sticky="ns")
|
|
sx.grid(row=1, column=0, sticky="ew")
|
|
|
|
bottom = ctk.CTkFrame(self)
|
|
bottom.pack(
|
|
fill="x",
|
|
padx=int(theme_value(self._theme, "frame_padx", 8)),
|
|
pady=(0, int(theme_value(self._theme, "frame_pady", 8))),
|
|
)
|
|
try:
|
|
bottom.configure(fg_color=theme_color(self._theme, "bottom_frame_fg_color", ("#dcdcdc", "#363636")))
|
|
except Exception:
|
|
pass
|
|
ctk.CTkLabel(
|
|
bottom,
|
|
text=loc_text("reset.summary", catalog=self._locale_catalog, default="Riepilogo"),
|
|
font=theme_font(self._theme, "summary_title_font", ("Segoe UI", 12, "bold")),
|
|
).pack(anchor="w", padx=8, pady=(8, 0))
|
|
|
|
g = ctk.CTkFrame(bottom)
|
|
g.pack(fill="x", padx=8, pady=8)
|
|
try:
|
|
g.configure(fg_color=theme_color(self._theme, "inner_summary_frame_fg_color", ("#d4d4d4", "#404040")))
|
|
except Exception:
|
|
pass
|
|
self.var_tot_celle = tk.StringVar(value="0")
|
|
self.var_occ = tk.StringVar(value="0")
|
|
self.var_dbl = tk.StringVar(value="0")
|
|
self.var_pallet = tk.StringVar(value="0")
|
|
|
|
def _kv(parent_widget, label, var, col):
|
|
"""Build a compact summary label/value pair."""
|
|
ctk.CTkLabel(
|
|
parent_widget,
|
|
text=label,
|
|
font=theme_font(self._theme, "summary_label_font", ("Segoe UI", 9, "bold")),
|
|
).grid(row=0, column=col * 2, sticky="w", padx=(0, 6))
|
|
ctk.CTkLabel(
|
|
parent_widget,
|
|
textvariable=var,
|
|
font=theme_font(self._theme, "summary_value_font", ("Segoe UI", 9)),
|
|
).grid(row=0, column=col * 2 + 1, sticky="w", padx=(0, 18))
|
|
|
|
g.grid_columnconfigure(7, weight=1)
|
|
_kv(g, "Tot. celle:", self.var_tot_celle, 0)
|
|
_kv(g, "Celle occupate:", self.var_occ, 1)
|
|
_kv(g, "Celle doppie:", self.var_dbl, 2)
|
|
_kv(g, "Tot. pallet:", self.var_pallet, 3)
|
|
|
|
@_log_call()
|
|
def _load_corsie(self):
|
|
"""Load available aisles and preselect ``1A`` when present."""
|
|
_log_sql("reset_corsie_corsie", SQL_CORSIE, {})
|
|
|
|
def _ok(res):
|
|
rows = res.get("rows", []) if isinstance(res, dict) else []
|
|
_log_dataset("reset_corsie_corsie", rows)
|
|
items = [r[0] for r in rows]
|
|
self.cmb.configure(values=items)
|
|
if items:
|
|
sel = "1A" if "1A" in items else items[0]
|
|
self.cmb.set(sel)
|
|
self.refresh()
|
|
else:
|
|
messagebox.showinfo("Info", "Nessuna corsia trovata.", parent=self)
|
|
|
|
def _err(ex):
|
|
_MODULE_LOGGER.exception(f"Errore caricamento corsie reset corsie: {ex}")
|
|
messagebox.showerror("Errore", f"Caricamento corsie fallito:\n{ex}", parent=self)
|
|
|
|
self._async.run(self.db.query_json(SQL_CORSIE, {}), _ok, _err, busy=self._busy, message="Carico corsie...")
|
|
|
|
@_log_call()
|
|
def refresh(self):
|
|
"""Refresh both the summary counters and the occupied-cell list."""
|
|
corsia = self.cmb.get().strip()
|
|
if not corsia:
|
|
return
|
|
_log_sql("reset_corsie_riepilogo", SQL_RIEPILOGO, {"corsia": corsia})
|
|
_log_sql("reset_corsie_dettaglio", SQL_DETTAGLIO, {"corsia": corsia})
|
|
|
|
async def _q():
|
|
riepilogo = await self.db.query_json(SQL_RIEPILOGO, {"corsia": corsia})
|
|
dettaglio = await self.db.query_json(SQL_DETTAGLIO, {"corsia": corsia})
|
|
return {"riepilogo": riepilogo, "dettaglio": dettaglio}
|
|
|
|
def _ok(payload):
|
|
try:
|
|
riepilogo = payload.get("riepilogo", {}) if isinstance(payload, dict) else {}
|
|
dettaglio = payload.get("dettaglio", {}) if isinstance(payload, dict) else {}
|
|
|
|
sum_rows = riepilogo.get("rows", []) if isinstance(riepilogo, dict) else []
|
|
det_rows = dettaglio.get("rows", []) if isinstance(dettaglio, dict) else []
|
|
_log_dataset("reset_corsie_riepilogo", sum_rows)
|
|
_log_dataset("reset_corsie_dettaglio", det_rows)
|
|
|
|
if sum_rows:
|
|
tot, occ, dbl, pallet = sum_rows[0]
|
|
self.var_tot_celle.set(str(tot or 0))
|
|
self.var_occ.set(str(occ or 0))
|
|
self.var_dbl.set(str(dbl or 0))
|
|
self.var_pallet.set(str(pallet or 0))
|
|
else:
|
|
self.var_tot_celle.set("0")
|
|
self.var_occ.set("0")
|
|
self.var_dbl.set("0")
|
|
self.var_pallet.set("0")
|
|
|
|
for item in self.tree.get_children():
|
|
self.tree.delete(item)
|
|
for idx, (_idc, ubi, n) in enumerate(det_rows):
|
|
tag = "even" if idx % 2 else "odd"
|
|
self.tree.insert("", "end", values=(ubi, n), tags=(tag,))
|
|
except Exception as ex:
|
|
_MODULE_LOGGER.exception(f"Errore UI refresh reset corsie corsia={corsia}: {ex}")
|
|
messagebox.showerror("Errore", f"Aggiornamento interfaccia fallito:\n{ex}", parent=self)
|
|
|
|
def _err(ex):
|
|
_MODULE_LOGGER.exception(f"Errore refresh reset corsie corsia={corsia}: {ex}")
|
|
messagebox.showerror("Errore", f"Refresh fallito:\n{ex}", parent=self)
|
|
|
|
self._async.run(_q(), _ok, _err, busy=self._busy, message=f"Riepilogo {corsia}...")
|
|
|
|
@_log_call()
|
|
def _ask_reset(self):
|
|
"""Ask for confirmation and start the logical unload flow for the selected aisle."""
|
|
corsia = self.cmb.get().strip()
|
|
if not corsia:
|
|
return
|
|
_log_sql("reset_corsie_count_reset", SQL_COUNT_RESET, {"corsia": corsia})
|
|
|
|
def _ok_count(res):
|
|
rows = res.get("rows", []) if isinstance(res, dict) else []
|
|
_log_dataset("reset_corsie_count_reset", rows)
|
|
tot_udc = int(rows[0][0] or 0) if rows else 0
|
|
tot_celle = int(rows[0][1] or 0) if rows else 0
|
|
if tot_udc <= 0:
|
|
messagebox.showinfo("Svuota corsia", f"Nessuna UDC attiva da scaricare per la corsia {corsia}.", parent=self)
|
|
return
|
|
msg = (
|
|
f"Verranno scaricate logicamente {tot_udc} UDC attive distribuite su {tot_celle} celle della corsia {corsia}.",
|
|
"L'operazione verra' eseguita come scarico verso 9000000 / 9999, senza cancellazioni fisiche dirette.",
|
|
"Digitare il nome della corsia per confermare:",
|
|
)
|
|
confirm = simpledialog.askstring("Conferma", "\n".join(msg), parent=self)
|
|
if confirm is None:
|
|
_MODULE_LOGGER.info(f"Reset corsia {corsia}: conferma annullata dall'utente")
|
|
return
|
|
if confirm.strip().upper() != corsia.upper():
|
|
_MODULE_LOGGER.info(f"Reset corsia {corsia}: testo conferma non corrispondente ({confirm!r})")
|
|
messagebox.showwarning("Annullato", "Testo di conferma non corrispondente.", parent=self)
|
|
return
|
|
self._do_reset(corsia)
|
|
|
|
def _err_count(ex):
|
|
_MODULE_LOGGER.exception(f"Errore conteggio reset corsie corsia={corsia}: {ex}")
|
|
messagebox.showerror("Errore", f"Conteggio UDC da scaricare fallito:\n{ex}", parent=self)
|
|
|
|
self._async.run(self.db.query_json(SQL_COUNT_RESET, {"corsia": corsia}), _ok_count, _err_count, busy=self._busy, message="Verifico...")
|
|
|
|
@_log_call()
|
|
def _do_reset(self, corsia: str):
|
|
"""Execute the logical unload of every active UDC in the selected aisle."""
|
|
_log_sql("reset_corsie_udc_reset", SQL_UDC_RESET, {"corsia": corsia})
|
|
|
|
async def _q():
|
|
payload = await self.db.query_json(SQL_UDC_RESET, {"corsia": corsia})
|
|
rows = payload.get("rows", []) if isinstance(payload, dict) else []
|
|
_log_dataset("reset_corsie_udc_reset", rows)
|
|
|
|
success = 0
|
|
failed: list[dict[str, Any]] = []
|
|
utente = str(getattr(self.session, "login", "") or "warehouse_ui").strip()
|
|
|
|
for barcode_pallet, idcella, ubicazione in rows:
|
|
try:
|
|
await move_pallet_async(
|
|
self.db,
|
|
barcode_pallet=str(barcode_pallet or "").strip(),
|
|
target_idcella=9999,
|
|
target_barcode_cella="9000000",
|
|
utente=utente,
|
|
)
|
|
success += 1
|
|
except Exception as ex:
|
|
failed.append(
|
|
{
|
|
"barcode_pallet": str(barcode_pallet or ""),
|
|
"idcella": int(idcella or 0),
|
|
"ubicazione": str(ubicazione or ""),
|
|
"error": str(ex),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"total": len(rows),
|
|
"success": success,
|
|
"failed": failed,
|
|
}
|
|
|
|
def _ok_del(result):
|
|
total = int((result or {}).get("total", 0))
|
|
success = int((result or {}).get("success", 0))
|
|
failed = list((result or {}).get("failed", []))
|
|
_MODULE_LOGGER.info(
|
|
f"Reset corsia {corsia}: scarico logico completato success={success} total={total} failed={len(failed)}"
|
|
)
|
|
if failed:
|
|
messagebox.showwarning(
|
|
"Completato con errori",
|
|
(
|
|
f"Corsia {corsia}: scaricate {success} UDC su {total}.\n"
|
|
f"Errori su {len(failed)} UDC. Controllare {MODULE_LOG_PATH.name}."
|
|
),
|
|
parent=self,
|
|
)
|
|
else:
|
|
messagebox.showinfo(
|
|
"Completato",
|
|
f"Corsia {corsia}: svuotamento logico completato su {success} UDC.",
|
|
parent=self,
|
|
)
|
|
self.refresh()
|
|
|
|
def _err_del(ex):
|
|
_MODULE_LOGGER.exception(f"Errore reset logico corsie corsia={corsia}: {ex}")
|
|
messagebox.showerror("Errore", f"Svuotamento fallito:\n{ex}", parent=self)
|
|
|
|
self._async.run(_q(), _ok_del, _err_del, busy=self._busy, message=f"Svuoto {corsia}...")
|
|
|
|
|
|
def open_reset_corsie_window(parent, db_app, session=None):
|
|
"""Create, focus and return the aisle reset window."""
|
|
key = "_reset_corsie_window_singleton"
|
|
ex = getattr(parent, key, None)
|
|
if ex and ex.winfo_exists():
|
|
try:
|
|
ex.deiconify()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
ex.lift()
|
|
ex.focus_force()
|
|
return ex
|
|
except Exception:
|
|
pass
|
|
win = ResetCorsieWindow(parent, db_app, session=session)
|
|
setattr(parent, key, win)
|
|
place_window_fullsize_below_parent_later(parent, win)
|
|
try:
|
|
win.lift()
|
|
win.focus_force()
|
|
except Exception:
|
|
pass
|
|
return win
|