administrator/ware_house
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Aggiungi pausa picking list nel barcode

Browse Source
This commit is contained in:
allebonvi
2026-07-03 10:13:49 +02:00
parent 29687c8094
commit 7ecc906d59
4 changed files with 238 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
barcode_service.py
View File

@@ -6,7 +6,7 @@ from dataclasses import dataclass
from typing import Literal
from barcode_repository import BarcodeRepository, LegacyMoveResult
from runtime_support import log_exception
from runtime_support import log_exception, log_runtime_event
from version_info import module_version
__version__ = module_version(__name__)
@@ -58,7 +58,7 @@ class BarcodeService:
def __init__(self, repository: BarcodeRepository, operator_id: int):
self.repository = repository
self.operator_id = int(operator_id)
self._current_priority_state = 0
self._current_priority_state = -1
self._state = BarcodeViewState()
@property
@@ -70,14 +70,14 @@ class BarcodeService:
def reset(self) -> BarcodeViewState:
"""Return the client to its neutral state."""
self._current_priority_state = 0
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 = 0
self._current_priority_state = -1
self._state = BarcodeViewState(
mode="manual_load",
queue_label="Versamento",
@@ -89,7 +89,7 @@ class BarcodeService:
def begin_manual_unload(self) -> BarcodeViewState:
"""Prepare a direct unload toward the conventional non-shelved cell."""
self._current_priority_state = 0
self._current_priority_state = -1
self._state = BarcodeViewState(
mode="manual_unload",
queue_label="Prelievo diretto",
@@ -100,6 +100,20 @@ class BarcodeService:
)
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
async def start_priority_queue(self, id_stato: int) -> BarcodeActionResult:
"""Load the next item of the selected legacy priority queue."""
@@ -107,6 +121,7 @@ class BarcodeService:
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._current_priority_state = -1
self._state = BarcodeViewState(
mode="manual_unload",
queue_label=queue_label,
@@ -169,6 +184,27 @@ class BarcodeService:
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)
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)
@@ -210,7 +246,7 @@ class BarcodeService:
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,
auto_advance_delay_ms=5000 if (is_direct_unload or is_direct_load) else 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}")
@@ -224,6 +260,92 @@ class BarcodeService:
)
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.")
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)
return await self.submit(scanned_pallet=pallet, destination_barcode=self.NON_SCAFFALATA_BARCODE)
async def _build_post_move_state(
self,
*,