# -*- coding: utf-8 -*- #!/usr/bin/env python3 """ Bridge TPE v3 -- copie EXACTE du TpeDriver du luge-print-agent ============================================================= Meme comportement, meme format de message, meme logique. """ import logging import os import socket import struct import sys import time from secrets import randbelow # ─── Config ───────────────────────────────────────────── DEFAULT_IP = "192.168.1.42" DEFAULT_PORT = 8888 DEFAULT_CAISSE_NUM = "01" TPE_SOCKET_TIMEOUT = 90 # 90s TPE_FIRST_FRAME_TIMEOUT = 90 # 90s TPE_FINAL_TIMEOUT = 120 # 120s RECONNECT_DELAYS = [1, 2, 5] BUFFER_SIZE = 4096 LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bridge.log") logger = logging.getLogger("bridge") logger.setLevel(logging.DEBUG) fh = logging.FileHandler(LOG_FILE) fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logger.addHandler(fh) ch = logging.StreamHandler(sys.stderr) ch.setLevel(logging.INFO) ch.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(ch) # ─── Message Concert V3 (EXACT comme TpeDriver) ───────── def build_message(amount_cents, pos_number="01"): """Copie EXACTE de TpeDriver.buildCaisseApMessage().""" tlv = lambda tag, val: tag + str(len(str(val))).zfill(3) + str(val) tx_id = str(randbelow(10 ** 12)).zfill(12) return ( tlv("CZ", "0300") + tlv("CJ", tx_id) + tlv("CA", str(pos_number or "01").zfill(2)) + tlv("CB", str(amount_cents)) + tlv("CD", "0") + tlv("CE", "978") ) def parse_response(raw): """Parse une reponse Concert V3 (comme parseCaisseApResponse).""" cleaned = raw.replace("\r", "").replace("\n", "") tags = {} i = 0 while i + 5 <= len(cleaned): tag = cleaned[i:i+2] if not tag.isalnum(): i += 1 continue try: length = int(cleaned[i+2:i+5]) except ValueError: break i += 5 if i + length > len(cleaned): break value = cleaned[i:i+length] tags[tag] = value i += length # Logique parseCaisseApResponse: # - CC trouve -> final + success # - AF trouve -> final + error # - sinon -> pending if "CC" in tags: return {"final": True, "success": True, "tags": tags, "summary": "approved"} if "AF" in tags: return {"final": True, "success": False, "tags": tags, "summary": f"error: AF={tags['AF']}"} return {"final": False, "success": False, "tags": tags, "summary": "no_af_tag_found"} # ─── TpeDriver Python (copie conforme) ────────────────── class TpeDriver: """Copie Python EXACTE du TpeDriver Node.js.""" def __init__(self, host, port, pos_number="01"): self.host = host self.port = port self.pos_number = pos_number self.sock = None self.connected = False self.is_busy = False self.current_tx = None self.reconnect_attempt = 0 self._rx_buf = b"" # ─── Public ─── def start(self): self._connect() def stop(self): self._destroy_socket() def start_payment(self, amount_cents): # Validation (comme TpeDriver) if not isinstance(amount_cents, (int, float)) or amount_cents <= 0: return {"ok": False, "error": "INVALID_AMOUNT"} # Reset si zombie ou occupe (comme TpeDriver) needs_reset = self.is_busy or (self.sock and not self.connected) if needs_reset: if self.is_busy: self._cancel() self._fail_tx("OVERRIDDEN", "Nouvelle transaction prioritaire") self._destroy_socket() self.connected = False if not self.sock: self._connect() if not self.sock: return {"ok": False, "error": "TPE_DISCONNECTED"} # Initialiser la transaction (comme TpeDriver) self.is_busy = True self.current_tx = { "amount": amount_cents, "started_at": time.time(), "first_frame_received": False, "response_data": b"", "final_result": None, } # Envoyer le message (comme TpeDriver) msg = build_message(amount_cents, self.pos_number) frame = msg.encode("ascii") logger.info(f" [SEND] {msg}") try: self.sock.sendall(frame) except OSError as e: self._fail_tx("TPE_SOCKET_ERROR", str(e)) return {"ok": False, "error": str(e)} # Attendre la reponse -- comportement synchrone # mais avec gestion des timeouts TpeDriver deadline = time.time() + TPE_FINAL_TIMEOUT first_frame_deadline = time.time() + TPE_FIRST_FRAME_TIMEOUT self.sock.settimeout(TPE_SOCKET_TIMEOUT) while time.time() < deadline: remaining = deadline - time.time() if remaining <= 0: break try: chunk = self.sock.recv(BUFFER_SIZE) except socket.timeout: logger.warning(" [TIMEOUT] socket timeout") continue except OSError as e: self._fail_tx("TPE_SOCKET_ERROR", str(e)) return {"ok": False, "error": str(e)} if not chunk: logger.warning(" [DISC] connexion fermee par le TPE") break # Premiere frame recue (comme TpeDriver) if not self.current_tx["first_frame_received"]: self.current_tx["first_frame_received"] = True logger.info(f" [RECV] recu {len(chunk)} octets: {chunk.hex()}") # Accumuler self.current_tx["response_data"] += chunk # Parser (comme TpeDriver handleData) raw = self.current_tx["response_data"].decode("ascii", errors="replace") parsed = parse_response(raw) logger.info(f" [PARSE] parse: final={parsed['final']} success={parsed['success']} summary={parsed['summary']}") # Si pas final, on continue d'attendre (comme TpeDriver emitProgress + return) if not parsed["final"]: continue # Final ! (comme TpeDriver emit tx_success / tx_error) self.is_busy = False self.current_tx = None if parsed["success"]: logger.info(f" [OK] SUCCÈS ! Tags: {parsed['tags']}") return {"ok": True, "tags": parsed["tags"], "raw": raw} else: logger.warning(f" [FAIL] ECHEC: {parsed['summary']}") return {"ok": False, "error": parsed["summary"], "tags": parsed["tags"], "raw": raw} # Timeout atteint (comme TpeDriver TPE_FINAL_TIMEOUT) self._fail_tx("TPE_FINAL_TIMEOUT", "tx_timeout") return {"ok": False, "error": "Timeout TPE"} def cancel(self): """Annule la transaction (envoie 0x18 comme TpeDriver).""" if not self.sock or not self.connected or not self.current_tx: return False try: self.sock.sendall(b"\x18") logger.info(" [CANCEL] ANNULATION (0x18)") return True except OSError: return False # ─── Prive (copie EXACTE du TpeDriver) ─── def _connect(self): """Copie EXACTE de TpeDriver.connect().""" if self.sock: if self.is_busy: self._fail_tx("TPE_RECONNECT_FORCED", "socket_reconnected_manually") self._destroy_socket() sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(10) # timeout de connexion initial sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # setKeepAlive(true, 10000) if hasattr(socket, "TCP_KEEPIDLE"): sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10) self.sock = sock try: sock.connect((self.host, self.port)) except (socket.timeout, ConnectionRefusedError, OSError) as e: logger.error(f" [FAIL] Connexion echouee: {e}") self.connected = False sock.close() self.sock = None self._schedule_reconnect() return self.connected = True self.reconnect_attempt = 0 logger.info(f" [OK] Connecte au TPE {self.host}:{self.port}") def _destroy_socket(self): """Detruit la socket proprement.""" if self.sock: try: self.sock.close() except Exception: pass self.sock = None self.connected = False def _schedule_reconnect(self): """Reconnexion avec backoff.""" idx = min(self.reconnect_attempt, len(RECONNECT_DELAYS) - 1) delay = RECONNECT_DELAYS[idx] self.reconnect_attempt += 1 logger.info(f" [RETRY] Reconnexion dans {delay}s...") time.sleep(delay) self._connect() def _cancel(self): """Cancel interne.""" try: if self.sock and self.connected: self.sock.sendall(b"\x18") except OSError: pass def _fail_tx(self, code, message): """Echoue la transaction (comme TpeDriver.failCurrentTx).""" if self.current_tx: logger.warning(f" [FAIL] {code}: {message}") self.current_tx = None self.is_busy = False # ─── CLI ───────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser(description="Bridge TPE v3 (copie TpeDriver)") parser.add_argument("-i", "--ip", default=DEFAULT_IP, help="IP du TPE") parser.add_argument("-p", "--port", type=int, default=DEFAULT_PORT, help="Port TCP") parser.add_argument("--caisse", default=DEFAULT_CAISSE_NUM, help="N° caisse") parser.add_argument("--test", type=float, metavar="€", help="Paiement test") parser.add_argument("--debug", action="store_true", help="Logs detailles") args = parser.parse_args() if args.debug: ch.setLevel(logging.DEBUG) driver = TpeDriver(args.ip, args.port, args.caisse) driver.start() # Attendre connexion time.sleep(0.5) if not driver.connected: logger.error(" [FAIL] Connexion impossible") sys.exit(1) try: if args.test is not None: cents = int(round(abs(args.test) * 100)) logger.info(f"[PAY] Paiement {args.test:.2f}€") result = driver.start_payment(cents) if result.get("ok"): logger.info(" [OK] SUCCÈS !") for k, v in result.get("tags", {}).items(): logger.info(f" {k} = {v}") else: logger.error(f" [FAIL] {result.get('error', 'Echec')}") else: logger.info(" Commandes: test <€>, cancel, quit") while driver.connected: try: cmd = input(" > ").strip() if cmd == "quit": break if cmd == "cancel": driver.cancel() elif cmd.startswith("test"): parts = cmd.split() if len(parts) > 1: amt = float(parts[1]) cents = int(round(abs(amt) * 100)) logger.info(f"[PAY] Paiement {amt:.2f}€") r = driver.start_payment(cents) if r.get("ok"): logger.info(" [OK] SUCCÈS !") else: logger.error(f" [FAIL] {r.get('error', 'Echec')}") except (KeyboardInterrupt, EOFError): break finally: driver.stop() if __name__ == "__main__": main()