"""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 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 = "" @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.""" 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 virtual outbound cell 9000000.""" 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="9000000", ) 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="priority_high" if int(id_stato) == 1 else "priority_low", queue_label=queue_label, status_text="Pronto.", status_color=self.RED, destination_barcode="9000000", ) 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="9000000", ) 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.") if self._state.mode in ("priority_high", "priority_low") and destination == "9000000": expected = str(self._state.expected_pallet or "").strip() if expected and expected != 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) move = await self.repository.execute_legacy_move( operator_id=self.operator_id, barcode_cella=destination, barcode_pallet=pallet, numero_cella=int(destination), ) if move.rc != 0: self._state.scanned_pallet = pallet self._state.status_text = f"Operazione non riuscita (RC={move.rc})." self._state.status_color = self.RED return BarcodeActionResult(False, self._state, self._state.status_text) self._state = await self._build_post_move_state( barcode_pallet=pallet, destination_barcode=destination, last_priority_state=self._current_priority_state, ) return BarcodeActionResult(True, self._state, self._state.status_text) async def _build_post_move_state( self, *, barcode_pallet: str, destination_barcode: str, last_priority_state: 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, ) 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 == "9000000" else f"Ok Carico - {destination_barcode}" ), status_color=self.GREEN_YELLOW, source_location=str(destination_barcode or ""), document=( self.CONVENTIONAL_LOCATION_BY_CELL[9999] if destination_barcode == "9000000" and last_priority_state in (0, 1) else lotto ), customer=( lotto if destination_barcode == "9000000" and last_priority_state in (0, 1) else prodotto ), expected_pallet=( " - ".join(part for part in (prodotto, descrizione) if part) if destination_barcode == "9000000" and last_priority_state in (0, 1) else descrizione ), destination_barcode=destination_barcode, scanned_pallet=barcode_pallet, ) 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 "")