"""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 "")