747 lines
33 KiB
Python
747 lines
33 KiB
Python
"""Service layer for the lightweight barcode WMS client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from typing import Literal
|
|
|
|
from barcode_repository import BarcodeRepository, LegacyMoveResult
|
|
from runtime_support import log_exception, log_runtime_event
|
|
from version_info import module_version
|
|
|
|
__version__ = module_version(__name__)
|
|
|
|
ModeName = Literal["idle", "priority_high", "priority_low", "manual_load", "manual_unload", "confirm"]
|
|
|
|
|
|
@dataclass
|
|
class BarcodeViewState:
|
|
"""State projected from service logic into the barcode UI."""
|
|
|
|
mode: ModeName = "idle"
|
|
queue_label: str = ""
|
|
status_text: str = "Pronto."
|
|
status_color: str = "#d9d9d9"
|
|
source_location: str = ""
|
|
document: str = ""
|
|
customer: str = ""
|
|
expected_pallet: str = ""
|
|
destination_barcode: str = ""
|
|
scanned_pallet: str = ""
|
|
auto_advance_delay_ms: int = 0
|
|
destination_readonly: bool = False
|
|
focus_destination: bool = False
|
|
|
|
|
|
@dataclass
|
|
class BarcodeActionResult:
|
|
"""Standard result returned by user actions."""
|
|
|
|
ok: bool
|
|
state: BarcodeViewState
|
|
message: str = ""
|
|
|
|
|
|
class BarcodeService:
|
|
"""Faithful, but cleaner, port of the legacy barcode form logic."""
|
|
|
|
NON_SCAFFALATA_BARCODE = "9001000"
|
|
SHIPPED_BARCODE = "9000000"
|
|
GRAY = "#d9d9d9"
|
|
RED = "#f4cccc"
|
|
LIGHT_GREEN = "#d9ead3"
|
|
GREEN_YELLOW = "#e2f0cb"
|
|
CONVENTIONAL_LOCATION_BY_CELL = {
|
|
1000: "5E.1.1 - Non scaffalata",
|
|
9999: "7G.1.1 - Spedita",
|
|
}
|
|
|
|
def __init__(self, repository: BarcodeRepository, operator_id: int):
|
|
self.repository = repository
|
|
self.operator_id = int(operator_id)
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState()
|
|
|
|
@property
|
|
def state(self) -> BarcodeViewState:
|
|
"""Return a copy-safe reference to the current UI state."""
|
|
|
|
return self._state
|
|
|
|
def reset(self) -> BarcodeViewState:
|
|
"""Return the client to its neutral state."""
|
|
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState()
|
|
return self._state
|
|
|
|
def begin_manual_load(self) -> BarcodeViewState:
|
|
"""Prepare a real versamento into a physical warehouse cell."""
|
|
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState(
|
|
mode="manual_load",
|
|
queue_label="Versamento",
|
|
status_text="OP Carico",
|
|
status_color=self.GRAY,
|
|
)
|
|
return self._state
|
|
|
|
def begin_manual_unload(self) -> BarcodeViewState:
|
|
"""Prepare a direct unload toward the conventional non-shelved cell."""
|
|
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState(
|
|
mode="manual_unload",
|
|
queue_label="Prelievo diretto",
|
|
status_text="OP Scarico",
|
|
status_color=self.GRAY,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
destination_readonly=True,
|
|
)
|
|
return self._state
|
|
|
|
def begin_priority_pause(self, id_stato: int) -> BarcodeViewState:
|
|
"""Pause the active picking queue locally for exactly one free movement."""
|
|
|
|
self._current_priority_state = int(id_stato)
|
|
self._state = BarcodeViewState(
|
|
mode="manual_unload",
|
|
queue_label="Pausa PL",
|
|
status_text="Premi ESC per uscire dalla pausa",
|
|
status_color=self.GRAY,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
destination_readonly=False,
|
|
)
|
|
return self._state
|
|
|
|
def resume_priority_state(self, state: BarcodeViewState) -> BarcodeViewState:
|
|
"""Restore the exact picking pallet that was visible before a local pause."""
|
|
|
|
if state.mode not in ("priority_high", "priority_low"):
|
|
return self._state
|
|
self._current_priority_state = 1 if state.mode == "priority_high" else 0
|
|
self._state = replace(state, scanned_pallet="", auto_advance_delay_ms=0)
|
|
return self._state
|
|
|
|
async def skip_current_picking_pallet(self) -> BarcodeActionResult:
|
|
"""Mark the currently proposed picking pallet as skipped and advance."""
|
|
|
|
mode = str(self._state.mode or "")
|
|
if mode not in ("priority_high", "priority_low"):
|
|
return BarcodeActionResult(False, self._state, "Nessuna picking list attiva.")
|
|
|
|
id_stato = 1 if mode == "priority_high" else 0
|
|
documento = str(self._state.document or "").strip()
|
|
pallet = str(self._state.expected_pallet or "").strip()
|
|
if not documento or not pallet:
|
|
self._state.status_text = "Nessuna UDC da saltare."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
await self.repository.skip_picking_pallet(
|
|
documento=documento,
|
|
pallet=pallet,
|
|
id_stato=id_stato,
|
|
operator_id=self.operator_id,
|
|
)
|
|
queue_label = "Alta priorita' (F1)" if id_stato == 1 else "Bassa priorita' (F2)"
|
|
self._current_priority_state = id_stato
|
|
self._state = BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label=queue_label,
|
|
status_text="UDC saltata - chiusura manuale richiesta",
|
|
status_color=self.RED,
|
|
source_location=self._state.source_location,
|
|
document=documento,
|
|
customer="UDC non trovata",
|
|
expected_pallet=pallet,
|
|
destination_barcode=self.SHIPPED_BARCODE,
|
|
auto_advance_delay_ms=1500,
|
|
destination_readonly=True,
|
|
)
|
|
return BarcodeActionResult(True, self._state, self._state.status_text)
|
|
|
|
async def start_priority_queue(self, id_stato: int) -> BarcodeActionResult:
|
|
"""Load the next item of the selected legacy priority queue."""
|
|
|
|
row = await self.repository.fetch_next_picking(id_stato)
|
|
self._current_priority_state = int(id_stato)
|
|
queue_label = "Alta priorita' (F1)" if int(id_stato) == 1 else "Bassa priorita' (F2)"
|
|
if not row:
|
|
skipped_doc = await self.repository.fetch_active_skipped_document(int(id_stato))
|
|
if skipped_doc:
|
|
documento = str(skipped_doc.get("Documento") or "").strip()
|
|
skipped_count = int(skipped_doc.get("SkippedCount") or 0)
|
|
await self.repository.release_picking_document(
|
|
documento=documento,
|
|
operator_id=self.operator_id,
|
|
)
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState(
|
|
mode="manual_unload",
|
|
queue_label=queue_label,
|
|
status_text=f"PL {documento} sospesa: {skipped_count} UDC saltate. Riprenota per completare.",
|
|
status_color=self.RED,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
destination_readonly=False,
|
|
)
|
|
return BarcodeActionResult(True, self._state)
|
|
self._current_priority_state = -1
|
|
self._state = BarcodeViewState(
|
|
mode="manual_unload",
|
|
queue_label=queue_label,
|
|
status_text="Pronto.",
|
|
status_color=self.RED,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
destination_readonly=False,
|
|
)
|
|
return BarcodeActionResult(True, self._state)
|
|
|
|
customer = f"{row.get('CodNazione') or ''} - {row.get('NAZIONE') or ''}".strip(" -")
|
|
source_location = self._display_location(
|
|
cella=row.get("Cella"),
|
|
ubicazione=row.get("Ubicazione"),
|
|
)
|
|
self._state = BarcodeViewState(
|
|
mode="priority_high" if int(id_stato) == 1 else "priority_low",
|
|
queue_label=queue_label,
|
|
status_text=f"Ok Cella - {source_location}",
|
|
status_color=self.LIGHT_GREEN,
|
|
source_location=source_location,
|
|
document=str(row.get("Documento") or ""),
|
|
customer=customer,
|
|
expected_pallet=str(row.get("Pallet") or ""),
|
|
destination_barcode=self.SHIPPED_BARCODE,
|
|
destination_readonly=True,
|
|
)
|
|
return BarcodeActionResult(True, self._state)
|
|
|
|
async def submit(self, *, scanned_pallet: str, destination_barcode: str) -> BarcodeActionResult:
|
|
"""Execute the movement according to the current mode."""
|
|
|
|
pallet = str(scanned_pallet or "").strip()
|
|
destination = str(destination_barcode or "").strip()
|
|
if not pallet:
|
|
return BarcodeActionResult(False, self._state, "Inserisci o leggi il pallet.")
|
|
if not destination:
|
|
return BarcodeActionResult(False, self._state, "Inserisci o leggi la destinazione.")
|
|
if not destination.isdigit():
|
|
return BarcodeActionResult(False, self._state, "La destinazione deve essere numerica.")
|
|
|
|
is_priority_mode = self._state.mode in ("priority_high", "priority_low")
|
|
expected_before_move = str(self._state.expected_pallet or "").strip()
|
|
is_picking_unload = bool(is_priority_mode and expected_before_move and destination == self.SHIPPED_BARCODE)
|
|
is_direct_unload = bool(destination == self.NON_SCAFFALATA_BARCODE and not is_picking_unload)
|
|
is_direct_load = bool(destination not in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE) and not is_picking_unload)
|
|
|
|
if is_priority_mode and destination == self.SHIPPED_BARCODE:
|
|
if expected_before_move and expected_before_move != pallet:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.status_text = "Errata lettura: il pallet letto non coincide con quello atteso."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
if not is_picking_unload:
|
|
blocked = await self._block_closed_picking_free_move(pallet, destination)
|
|
if blocked is not None:
|
|
return blocked
|
|
|
|
target_barcode = destination
|
|
target_numero_cella = int(destination)
|
|
target_id_cella = 9999 if destination == self.SHIPPED_BARCODE else (1000 if destination == self.NON_SCAFFALATA_BARCODE else None)
|
|
target_display = destination
|
|
if is_direct_load:
|
|
resolved_cell = await self.repository.resolve_physical_cell(destination)
|
|
if not resolved_cell:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = f"Cella non valida: {destination}."
|
|
self._state.status_color = self.RED
|
|
self._state.focus_destination = True
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
occupants = await self.repository.fetch_open_pallets_by_cell(resolved_cell.id_cella)
|
|
if occupants:
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED DESTINATION_OCCUPIED "
|
|
f"pallet={pallet} "
|
|
f"cell={resolved_cell.id_cella} "
|
|
f"occupants={','.join(str(row.get('BarcodePallet') or '') for row in occupants)}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Errore: cella scelta occupata, cambiare cella di destinazione."
|
|
self._state.status_color = self.RED
|
|
self._state.destination_readonly = False
|
|
self._state.focus_destination = True
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
target_barcode = resolved_cell.barcode_cella
|
|
target_numero_cella = resolved_cell.numero_cella
|
|
target_id_cella = resolved_cell.id_cella
|
|
target_display = resolved_cell.ubicazione or destination
|
|
|
|
current_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
picking_before_move = (
|
|
None
|
|
if current_location
|
|
else await self.repository.fetch_picking_by_pallet(pallet)
|
|
)
|
|
if not current_location and not picking_before_move:
|
|
if is_direct_load:
|
|
trace_row = await self.repository.fetch_trace_by_pallet(pallet)
|
|
if not trace_row:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Errore: UDC non riconosciuta in ERP/SAM."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
f"AUTO INTAKE NON_SHELVED pallet={pallet} before_cell={target_id_cella}",
|
|
)
|
|
await self.repository.execute_legacy_move(
|
|
operator_id=self.operator_id,
|
|
barcode_cella=self.NON_SCAFFALATA_BARCODE,
|
|
barcode_pallet=pallet,
|
|
numero_cella=int(self.NON_SCAFFALATA_BARCODE),
|
|
refresh_picking_reservation=False,
|
|
)
|
|
current_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
try:
|
|
current_id_cella = int((current_location or {}).get("IDCella") or 0)
|
|
except Exception:
|
|
current_id_cella = 0
|
|
if current_id_cella != 1000:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Movimento non confermato: ingresso UDC non scaffalata non riuscito."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
else:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.status_text = "UDC non presente a magazzino."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
open_locations = await self.repository.fetch_open_locations_by_pallet(pallet)
|
|
distinct_cells = sorted({
|
|
int(row.get("IDCella") or 0)
|
|
for row in open_locations
|
|
if row.get("IDCella") is not None
|
|
})
|
|
if len(distinct_cells) > 1:
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED MULTI_LOCATION "
|
|
f"pallet={pallet} "
|
|
f"cells={','.join(str(cell) for cell in distinct_cells)}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "UDC presente in piu' celle. Movimento bloccato."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
if not is_direct_load and destination not in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE):
|
|
resolved_cell = await self.repository.resolve_physical_cell(destination)
|
|
if not resolved_cell:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.status_text = f"Cella non valida: {destination}."
|
|
self._state.status_color = self.RED
|
|
self._state.focus_destination = True
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
target_barcode = resolved_cell.barcode_cella
|
|
target_numero_cella = resolved_cell.numero_cella
|
|
target_id_cella = resolved_cell.id_cella
|
|
target_display = resolved_cell.ubicazione or destination
|
|
|
|
move = await self.repository.execute_legacy_move(
|
|
operator_id=self.operator_id,
|
|
barcode_cella=target_barcode,
|
|
barcode_pallet=pallet,
|
|
numero_cella=target_numero_cella,
|
|
refresh_picking_reservation=is_picking_unload,
|
|
)
|
|
|
|
final_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
try:
|
|
final_cell = int((final_location or {}).get("IDCella") or 0)
|
|
except Exception:
|
|
final_cell = 0
|
|
if target_id_cella is not None and final_cell != target_id_cella:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Movimento non confermato: la UDC non risulta nella cella attesa."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
try:
|
|
self._state = await self._build_post_move_state(
|
|
barcode_pallet=pallet,
|
|
destination_barcode=destination,
|
|
destination_display=target_display,
|
|
last_priority_state=self._current_priority_state,
|
|
auto_advance_delay_ms=3000 if is_picking_unload else 0,
|
|
)
|
|
except Exception as exc:
|
|
log_exception("Barcode WMS", exc, context=f"post move state pallet={pallet} destination={destination}")
|
|
self._state = BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label="Conferma movimento",
|
|
status_text="Movimento eseguito. Dettagli non aggiornati.",
|
|
status_color=self.GREEN_YELLOW,
|
|
destination_barcode=destination,
|
|
scanned_pallet=pallet,
|
|
)
|
|
return BarcodeActionResult(True, self._state, self._state.status_text)
|
|
|
|
async def submit_unload_with_source_check(self, *, scanned_pallet: str, source_barcode: str) -> BarcodeActionResult:
|
|
"""Unload to the non-shelved cell only if the scanned source cell matches DB."""
|
|
|
|
pallet = str(scanned_pallet or "").strip()
|
|
source = str(source_barcode or "").strip()
|
|
if not pallet:
|
|
return BarcodeActionResult(False, self._state, "Inserisci o leggi il pallet.")
|
|
if not source:
|
|
return BarcodeActionResult(False, self._state, "Leggi il codice della cella da scaricare.")
|
|
if not source.isdigit():
|
|
return BarcodeActionResult(False, self._state, "La cella da scaricare deve essere numerica.")
|
|
|
|
blocked = await self._block_closed_picking_free_move(pallet, source)
|
|
if blocked is not None:
|
|
return blocked
|
|
|
|
current_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
if not current_location:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.status_text = "UDC non presente a magazzino."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
open_locations = await self.repository.fetch_open_locations_by_pallet(pallet)
|
|
distinct_cells = sorted({
|
|
int(row.get("IDCella") or 0)
|
|
for row in open_locations
|
|
if row.get("IDCella") is not None
|
|
})
|
|
if len(distinct_cells) > 1:
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED MULTI_LOCATION "
|
|
f"pallet={pallet} "
|
|
f"cells={','.join(str(cell) for cell in distinct_cells)}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = source
|
|
self._state.status_text = "UDC presente in piu' celle. Movimento bloccato."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
if source == self.NON_SCAFFALATA_BARCODE:
|
|
source_id_cella = 1000
|
|
else:
|
|
resolved_source = await self.repository.resolve_physical_cell(source)
|
|
if not resolved_source:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = source
|
|
self._state.status_text = f"Cella non valida: {source}."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
source_id_cella = int(resolved_source.id_cella)
|
|
|
|
current_id_cella = int((current_location or {}).get("IDCella") or 0)
|
|
if current_id_cella != source_id_cella:
|
|
if source == self.NON_SCAFFALATA_BARCODE and current_id_cella != 1000:
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED SOURCE_REQUIRED "
|
|
f"pallet={pallet} "
|
|
f"db_cell={current_id_cella} "
|
|
f"default_cell={source_id_cella}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = source
|
|
self._state.status_text = "Errore: leggi la cella da scaricare."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED SOURCE_MISMATCH "
|
|
f"pallet={pallet} "
|
|
f"db_cell={current_id_cella} "
|
|
f"scanned_cell={source_id_cella}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = source
|
|
self._state.status_text = "Errore: UDC in cella diversa, rileggi il codice della cella da scaricare."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
await self.repository.execute_legacy_move(
|
|
operator_id=self.operator_id,
|
|
barcode_cella=self.NON_SCAFFALATA_BARCODE,
|
|
barcode_pallet=pallet,
|
|
numero_cella=int(self.NON_SCAFFALATA_BARCODE),
|
|
refresh_picking_reservation=False,
|
|
)
|
|
|
|
final_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
try:
|
|
final_cell = int((final_location or {}).get("IDCella") or 0)
|
|
except Exception:
|
|
final_cell = 0
|
|
if final_cell != 1000:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = source
|
|
self._state.status_text = "Movimento non confermato: la UDC non risulta non scaffalata."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
try:
|
|
self._state = await self._build_post_move_state(
|
|
barcode_pallet=pallet,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
destination_display=self.CONVENTIONAL_LOCATION_BY_CELL[1000],
|
|
last_priority_state=self._current_priority_state,
|
|
auto_advance_delay_ms=0,
|
|
)
|
|
except Exception as exc:
|
|
log_exception("Barcode WMS", exc, context=f"post source-check unload pallet={pallet} source={source}")
|
|
self._state = BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label="Conferma movimento",
|
|
status_text="Movimento eseguito. Dettagli non aggiornati.",
|
|
status_color=self.GREEN_YELLOW,
|
|
destination_barcode=self.NON_SCAFFALATA_BARCODE,
|
|
scanned_pallet=pallet,
|
|
)
|
|
return BarcodeActionResult(True, self._state, self._state.status_text)
|
|
|
|
async def _block_closed_picking_free_move(
|
|
self,
|
|
pallet: str,
|
|
destination_barcode: str,
|
|
) -> BarcodeActionResult | None:
|
|
closed_row = await self.repository.fetch_closed_picking_by_pallet(pallet)
|
|
if not closed_row:
|
|
return None
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE BLOCKED CLOSED_PICKING "
|
|
f"pallet={pallet} "
|
|
f"destination={destination_barcode} "
|
|
f"id_cella={closed_row.get('IDCella')} "
|
|
f"id_stato={closed_row.get('IDStato')}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination_barcode
|
|
self._state.status_text = "Errore: UDC gia' spedita, problema nella locazione o nella UDC."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
async def reconcile_after_submit_exception(
|
|
self,
|
|
*,
|
|
scanned_pallet: str,
|
|
destination_barcode: str,
|
|
) -> BarcodeActionResult:
|
|
"""Check whether a failed client round-trip actually committed on SQL Server."""
|
|
|
|
pallet = str(scanned_pallet or "").strip()
|
|
destination = str(destination_barcode or "").strip()
|
|
if not pallet or not destination or not destination.isdigit():
|
|
self._state.status_text = "Transazione non completata, ripeti l'operazione."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
target_id_cella = 9999 if destination == self.SHIPPED_BARCODE else (1000 if destination == self.NON_SCAFFALATA_BARCODE else None)
|
|
target_display = destination
|
|
if target_id_cella is None:
|
|
resolved_cell = await self.repository.resolve_physical_cell(destination)
|
|
if not resolved_cell:
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Transazione non completata, ripeti l'operazione."
|
|
self._state.status_color = self.RED
|
|
self._state.focus_destination = True
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
target_id_cella = int(resolved_cell.id_cella)
|
|
target_display = resolved_cell.ubicazione or destination
|
|
|
|
current_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
try:
|
|
current_id_cella = int((current_location or {}).get("IDCella") or 0)
|
|
except Exception:
|
|
current_id_cella = 0
|
|
|
|
if current_id_cella != target_id_cella:
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE RECOVERY NOT_CONFIRMED "
|
|
f"pallet={pallet} "
|
|
f"destination={destination} "
|
|
f"expected_cell={target_id_cella} "
|
|
f"actual_cell={current_id_cella}"
|
|
),
|
|
)
|
|
self._state.scanned_pallet = pallet
|
|
self._state.destination_barcode = destination
|
|
self._state.status_text = "Transazione non completata, ripeti l'operazione."
|
|
self._state.status_color = self.RED
|
|
return BarcodeActionResult(False, self._state, self._state.status_text)
|
|
|
|
last_priority_state = self._current_priority_state
|
|
is_recovered_picking = bool(destination == self.SHIPPED_BARCODE and last_priority_state in (0, 1))
|
|
auto_advance_delay_ms = 3000 if is_recovered_picking else 0
|
|
state = await self._build_post_move_state(
|
|
barcode_pallet=pallet,
|
|
destination_barcode=destination,
|
|
destination_display=target_display,
|
|
last_priority_state=last_priority_state,
|
|
auto_advance_delay_ms=auto_advance_delay_ms,
|
|
)
|
|
state.status_text = (
|
|
"Ok Scarico - risposta persa recuperata"
|
|
if is_recovered_picking
|
|
else "Movimento completato - risposta persa recuperata"
|
|
)
|
|
state.status_color = self.GREEN_YELLOW
|
|
self._state = state
|
|
log_runtime_event(
|
|
"Barcode WMS",
|
|
(
|
|
"MOVE RECOVERY CONFIRMED "
|
|
f"pallet={pallet} "
|
|
f"destination={destination} "
|
|
f"cell={target_id_cella}"
|
|
),
|
|
)
|
|
return BarcodeActionResult(True, self._state, self._state.status_text)
|
|
|
|
async def _build_post_move_state(
|
|
self,
|
|
*,
|
|
barcode_pallet: str,
|
|
destination_barcode: str,
|
|
destination_display: str,
|
|
last_priority_state: int,
|
|
auto_advance_delay_ms: int,
|
|
) -> BarcodeViewState:
|
|
"""Mirror the legacy confirmation flow after one stored-procedure move."""
|
|
|
|
is_priority_shipping_confirmation = (
|
|
destination_barcode == self.SHIPPED_BARCODE
|
|
and last_priority_state in (0, 1)
|
|
)
|
|
picking_row = (
|
|
await self.repository.fetch_picking_by_pallet(barcode_pallet)
|
|
if is_priority_shipping_confirmation
|
|
else None
|
|
)
|
|
if picking_row:
|
|
customer = f"{picking_row.get('CodNazione') or ''} - {picking_row.get('NAZIONE') or ''}".strip(" -")
|
|
source_location = self._display_location(
|
|
cella=picking_row.get("Cella"),
|
|
ubicazione=picking_row.get("Ubicazione"),
|
|
)
|
|
return BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label="Alta priorita' (F1)" if last_priority_state == 1 else ("Bassa priorita' (F2)" if last_priority_state == 0 else ""),
|
|
status_text=f"Ok Cella - {source_location}",
|
|
status_color=self.LIGHT_GREEN,
|
|
source_location=source_location,
|
|
document=str(picking_row.get("Documento") or ""),
|
|
customer=customer,
|
|
expected_pallet=str(picking_row.get("Pallet") or ""),
|
|
destination_barcode=destination_barcode,
|
|
auto_advance_delay_ms=auto_advance_delay_ms,
|
|
destination_readonly=True,
|
|
)
|
|
|
|
trace_row = await self.repository.fetch_trace_by_pallet(barcode_pallet)
|
|
if trace_row:
|
|
lotto = str(trace_row.get("Lotto") or "")
|
|
prodotto = str(trace_row.get("Prodotto") or "")
|
|
descrizione = str(trace_row.get("Descrizione") or "")
|
|
queue_label = "Alta priorita' (F1)" if last_priority_state == 1 else ("Bassa priorita' (F2)" if last_priority_state == 0 else "Conferma movimento")
|
|
destination_label = self._display_destination_location(
|
|
destination_barcode=destination_barcode,
|
|
destination_display=destination_display,
|
|
)
|
|
return BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label=queue_label,
|
|
status_text=(
|
|
f"Ok Scarico - {destination_label}"
|
|
if destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE)
|
|
else f"Ok Carico - {destination_label}"
|
|
),
|
|
status_color=self.GREEN_YELLOW,
|
|
source_location=destination_label,
|
|
document=(
|
|
self.CONVENTIONAL_LOCATION_BY_CELL[9999 if destination_barcode == self.SHIPPED_BARCODE else 1000]
|
|
if destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE) and last_priority_state in (0, 1)
|
|
else lotto
|
|
),
|
|
customer=(
|
|
lotto
|
|
if destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE) and last_priority_state in (0, 1)
|
|
else prodotto
|
|
),
|
|
expected_pallet=(
|
|
" - ".join(part for part in (prodotto, descrizione) if part)
|
|
if destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE) and last_priority_state in (0, 1)
|
|
else descrizione
|
|
),
|
|
destination_barcode=destination_barcode,
|
|
scanned_pallet=barcode_pallet,
|
|
auto_advance_delay_ms=auto_advance_delay_ms,
|
|
destination_readonly=destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE),
|
|
)
|
|
|
|
return BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label="Conferma movimento",
|
|
status_text="Movimento eseguito.",
|
|
status_color=self.GREEN_YELLOW,
|
|
destination_barcode=destination_barcode,
|
|
scanned_pallet=barcode_pallet,
|
|
)
|
|
|
|
def _display_destination_location(self, *, destination_barcode: str, destination_display: str) -> str:
|
|
"""Return the confirmation label for a movement destination."""
|
|
|
|
if destination_barcode == self.NON_SCAFFALATA_BARCODE:
|
|
return self.CONVENTIONAL_LOCATION_BY_CELL[1000]
|
|
if destination_barcode == self.SHIPPED_BARCODE:
|
|
return self.CONVENTIONAL_LOCATION_BY_CELL[9999]
|
|
return str(destination_display or destination_barcode or "")
|
|
|
|
def _display_location(self, *, cella: object, ubicazione: object) -> str:
|
|
"""Return the operator-facing location, honoring legacy conventional cells."""
|
|
|
|
try:
|
|
cella_int = int(cella)
|
|
except Exception:
|
|
cella_int = None
|
|
if cella_int in self.CONVENTIONAL_LOCATION_BY_CELL:
|
|
return self.CONVENTIONAL_LOCATION_BY_CELL[cella_int]
|
|
return str(ubicazione or cella or "")
|