313 lines
13 KiB
Python
313 lines
13 KiB
Python
"""Service layer for the lightweight barcode WMS client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Literal
|
|
|
|
from barcode_repository import BarcodeRepository, LegacyMoveResult
|
|
from runtime_support import log_exception
|
|
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
|
|
|
|
|
|
@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: "5E1.1",
|
|
9999: "7G.1.1",
|
|
}
|
|
|
|
def __init__(self, repository: BarcodeRepository, operator_id: int):
|
|
self.repository = repository
|
|
self.operator_id = int(operator_id)
|
|
self._current_priority_state = 0
|
|
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 = 0
|
|
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 = 0
|
|
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 = 0
|
|
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
|
|
|
|
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:
|
|
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)
|
|
|
|
current_location = await self.repository.fetch_current_location_by_pallet(pallet)
|
|
picking_before_move = await self.repository.fetch_picking_by_pallet(pallet)
|
|
if not current_location and not picking_before_move:
|
|
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)
|
|
|
|
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 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
|
|
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,
|
|
)
|
|
|
|
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=5000 if (is_direct_unload or is_direct_load) else 1200 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 _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."""
|
|
|
|
picking_row = await self.repository.fetch_picking_by_pallet(barcode_pallet)
|
|
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")
|
|
return BarcodeViewState(
|
|
mode="confirm",
|
|
queue_label=queue_label,
|
|
status_text=(
|
|
"Ok Scarico" if destination_barcode in (self.NON_SCAFFALATA_BARCODE, self.SHIPPED_BARCODE) else f"Ok Carico - {destination_display}"
|
|
),
|
|
status_color=self.GREEN_YELLOW,
|
|
source_location=str(destination_display or destination_barcode or ""),
|
|
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_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 "")
|