host.nfc_tool.app
NFC Battery Tag Tool — main application window.
Coordinates the serial panel, tabs, and background NFC worker thread. All NFC I/O runs on a background thread; results are dispatched to tabs via tkinter after() polling.
1"""NFC Battery Tag Tool — main application window. 2 3Coordinates the serial panel, tabs, and background NFC worker thread. 4All NFC I/O runs on a background thread; results are dispatched to tabs 5via tkinter after() polling. 6""" 7 8import logging 9import queue 10import threading 11import tkinter as tk 12from tkinter import ttk 13from pathlib import Path 14 15from utils.nfc.nfc_serial_transport import NfcSerialTransport 16from utils.nfc.nfc_reader import NfcReader 17from utils.nfc.nfc_writer import build_battery_ndef 18 19from host.nfc_tool.serial_panel import SerialPanel 20from host.nfc_tool.scan_tab import ScanTab 21from host.nfc_tool.read_tab import ReadTab 22from host.nfc_tool.write_tab import WriteTab 23 24logger = logging.getLogger(__name__) 25 26# Commands sent to worker thread 27CMD_CONNECT = 'connect' 28CMD_DISCONNECT = 'disconnect' 29CMD_POLL = 'poll' 30CMD_WRITE = 'write' 31CMD_STOP = 'stop' 32 33 34class NfcWorker(threading.Thread): 35 """Background thread for all NFC serial I/O.""" 36 37 def __init__(self): 38 super().__init__(daemon=True) 39 self.cmd_queue = queue.Queue() 40 self.result_queue = queue.Queue() 41 self._transport = None 42 self._reader = None 43 44 def run(self): 45 while True: 46 try: 47 cmd = self.cmd_queue.get(timeout=0.5) 48 except queue.Empty: 49 continue 50 51 if cmd[0] == CMD_STOP: 52 self._cleanup() 53 break 54 elif cmd[0] == CMD_CONNECT: 55 self._do_connect(cmd[1]) 56 elif cmd[0] == CMD_DISCONNECT: 57 self._do_disconnect() 58 elif cmd[0] == CMD_POLL: 59 self._do_poll(cmd[1]) 60 elif cmd[0] == CMD_WRITE: 61 self._do_write(cmd[1]) 62 63 def _do_connect(self, port): 64 try: 65 self._transport = NfcSerialTransport(port, timeout=0.2) 66 if not self._transport.open(): 67 self.result_queue.put( 68 ('connected', False, "Failed to open serial port")) 69 return 70 self._reader = NfcReader(self._transport) 71 if not self._reader.begin(): 72 self._transport.close() 73 self.result_queue.put( 74 ('connected', False, "PN532 init failed")) 75 return 76 self.result_queue.put(('connected', True, '')) 77 except Exception as e: 78 self.result_queue.put(('connected', False, str(e))) 79 80 def _do_disconnect(self): 81 self._cleanup() 82 self.result_queue.put(('disconnected',)) 83 84 def _do_poll(self, source): 85 if self._reader is None or not self._reader.is_ready: 86 self.result_queue.put(('tag_data', source, None)) 87 return 88 try: 89 tag_data = self._reader.read_full_tag( 90 user_data_start=4, user_data_pages=128) 91 self.result_queue.put(('tag_data', source, tag_data)) 92 except Exception as e: 93 logger.debug("Poll error: %s", e) 94 self.result_queue.put(('tag_data', source, None)) 95 96 def _do_write(self, params): 97 if self._reader is None or not self._reader.is_ready: 98 self.result_queue.put( 99 ('write_result', False, "Reader not connected")) 100 return 101 try: 102 # Build NDEF data 103 ndef_bytes = build_battery_ndef( 104 sn=params['sn'], 105 year=params['year'], 106 note=params.get('note', ''), 107 uri=params.get('uri', 'http://www.raptacon.org'), 108 ) 109 110 # Detect tag (fresh target listing for write) 111 result = self._reader.read_passive_target() 112 if result is None: 113 self.result_queue.put( 114 ('write_result', False, "No tag detected")) 115 return 116 117 tg, uid = result 118 119 # Pad NDEF with zeros to clear any old data beyond it 120 write_data = bytearray(ndef_bytes) 121 write_data.extend(b'\x00' * 32) 122 123 # Write starting at page 4 124 if not self._reader.write_ntag_pages( 125 tg, 4, bytes(write_data)): 126 self.result_queue.put( 127 ('write_result', False, "Page write failed")) 128 return 129 130 self.result_queue.put(('write_result', True, '')) 131 132 except ValueError as e: 133 self.result_queue.put(('write_result', False, str(e))) 134 except Exception as e: 135 logger.exception("Write error") 136 self.result_queue.put(('write_result', False, str(e))) 137 138 def _cleanup(self): 139 if self._transport is not None: 140 try: 141 self._transport.close() 142 except Exception as e: 143 logger.debug("Error closing NFC transport: %s", e) 144 self._transport = None 145 self._reader = None 146 147 148class NfcToolApp(tk.Tk): 149 """Main NFC Battery Tag Tool window.""" 150 151 def __init__(self, project_root=None): 152 super().__init__() 153 self.title("NFC Battery Tag Tool - Team 3200") 154 self.geometry("650x550") 155 self.minsize(500, 400) 156 157 self._project_root = Path(project_root) if project_root else None 158 self._load_icon() 159 160 # Worker thread 161 self._worker = NfcWorker() 162 self._worker.start() 163 164 self._build_ui() 165 166 # Poll worker results 167 self._check_results() 168 169 self.protocol("WM_DELETE_WINDOW", self._on_close) 170 171 def _load_icon(self): 172 """Set the window icon from the Raptacon logo.""" 173 if self._project_root is None: 174 return 175 icon_path = self._project_root / "images" / "Raptacon3200-BG-BW.png" 176 if icon_path.exists(): 177 try: 178 self._icon_image = tk.PhotoImage(file=str(icon_path)) 179 self.iconphoto(True, self._icon_image) 180 except Exception as e: 181 logger.debug("Failed to load window icon: %s", e) 182 183 def _load_gear_logo(self, parent): 184 """Load the gear logo and return a Label widget, or None.""" 185 if self._project_root is None: 186 return None 187 logo_path = self._project_root / "images" / "raptacongear.png" 188 if not logo_path.exists(): 189 return None 190 try: 191 img = tk.PhotoImage(file=str(logo_path)) 192 # Subsample to ~64px (original is large) 193 w = img.width() 194 factor = max(1, w // 64) 195 img = img.subsample(factor, factor) 196 self._gear_image = img # prevent GC 197 label = ttk.Label(parent, image=img) 198 return label 199 except Exception as e: 200 logger.debug("Failed to load gear logo: %s", e) 201 return None 202 203 def _build_ui(self): 204 # Header with serial panel and logo 205 header = ttk.Frame(self) 206 header.pack(fill=tk.X, padx=8, pady=(8, 4)) 207 208 # Serial panel (left side) 209 self._serial_panel = SerialPanel( 210 header, 211 on_connect=self._on_connect, 212 on_disconnect=self._on_disconnect, 213 ) 214 self._serial_panel.pack(side=tk.LEFT, fill=tk.X, expand=True) 215 216 # Gear logo (right side) 217 logo = self._load_gear_logo(header) 218 if logo: 219 logo.pack(side=tk.RIGHT, padx=(8, 0)) 220 221 ttk.Separator(self, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=8) 222 223 # Tabs 224 self._notebook = ttk.Notebook(self) 225 226 self._scan_tab = ScanTab( 227 self._notebook, request_poll=self._request_poll) 228 self._read_tab = ReadTab( 229 self._notebook, request_poll=self._request_poll) 230 self._write_tab = WriteTab( 231 self._notebook, 232 request_poll=self._request_poll, 233 request_write=self._request_write, 234 ) 235 236 self._notebook.add(self._scan_tab, text="Scan") 237 self._notebook.add(self._read_tab, text="Read") 238 self._notebook.add(self._write_tab, text="Write Battery Tag") 239 240 self._notebook.pack(fill=tk.BOTH, expand=True, padx=8, pady=(4, 8)) 241 self._notebook.bind("<<NotebookTabChanged>>", self._on_tab_changed) 242 243 self._tabs = [self._scan_tab, self._read_tab, self._write_tab] 244 self._active_tab_idx = 0 245 246 def _on_tab_changed(self, event): 247 # Notify old tab 248 if 0 <= self._active_tab_idx < len(self._tabs): 249 self._tabs[self._active_tab_idx].on_tab_deselected() 250 251 # Notify new tab 252 idx = self._notebook.index(self._notebook.select()) 253 self._active_tab_idx = idx 254 if 0 <= idx < len(self._tabs): 255 self._tabs[idx].on_tab_selected() 256 257 def _on_connect(self, port): 258 self._worker.cmd_queue.put((CMD_CONNECT, port)) 259 260 def _on_disconnect(self): 261 # Stop any tab polling 262 for tab in self._tabs: 263 tab.on_tab_deselected() 264 self._worker.cmd_queue.put((CMD_DISCONNECT,)) 265 266 def _request_poll(self, source): 267 """Tabs call this to request a tag poll.""" 268 if not self._serial_panel.is_connected: 269 return 270 self._worker.cmd_queue.put((CMD_POLL, source)) 271 272 def _request_write(self, sn, year, note='', uri=''): 273 """Write tab calls this to perform the actual write.""" 274 self._worker.cmd_queue.put((CMD_WRITE, { 275 'sn': sn, 'year': year, 'note': note, 'uri': uri, 276 })) 277 278 def _check_results(self): 279 """Poll worker result queue and dispatch to tabs.""" 280 try: 281 while True: 282 result = self._worker.result_queue.get_nowait() 283 self._handle_result(result) 284 except queue.Empty: 285 pass 286 self.after(100, self._check_results) 287 288 def _handle_result(self, result): 289 cmd = result[0] 290 291 if cmd == 'connected': 292 success, error_msg = result[1], result[2] 293 self._serial_panel.set_connected(success, error_msg) 294 if success: 295 # Activate current tab's polling 296 idx = self._active_tab_idx 297 if 0 <= idx < len(self._tabs): 298 self._tabs[idx].on_tab_selected() 299 300 elif cmd == 'disconnected': 301 self._serial_panel.set_disconnected() 302 303 elif cmd == 'tag_data': 304 source, tag_data = result[1], result[2] 305 if source == 'scan': 306 self._scan_tab.update_tag_data(tag_data) 307 elif source == 'read': 308 self._read_tab.update_tag_data(tag_data) 309 elif source == 'write_search': 310 self._write_tab.on_search_result(tag_data) 311 elif source == 'write_verify': 312 self._write_tab.on_verify_result(tag_data) 313 314 elif cmd == 'write_result': 315 success, error_msg = result[1], result[2] 316 self._write_tab.on_write_result(success, error_msg) 317 318 def _on_close(self): 319 # Stop all tab polling 320 for tab in self._tabs: 321 tab.on_tab_deselected() 322 # Stop worker 323 self._worker.cmd_queue.put((CMD_STOP,)) 324 self._worker.join(timeout=2.0) 325 self.destroy()
logger =
<Logger host.nfc_tool.app (DEBUG)>
CMD_CONNECT =
'connect'
CMD_DISCONNECT =
'disconnect'
CMD_POLL =
'poll'
CMD_WRITE =
'write'
CMD_STOP =
'stop'
class
NfcWorker(threading.Thread):
35class NfcWorker(threading.Thread): 36 """Background thread for all NFC serial I/O.""" 37 38 def __init__(self): 39 super().__init__(daemon=True) 40 self.cmd_queue = queue.Queue() 41 self.result_queue = queue.Queue() 42 self._transport = None 43 self._reader = None 44 45 def run(self): 46 while True: 47 try: 48 cmd = self.cmd_queue.get(timeout=0.5) 49 except queue.Empty: 50 continue 51 52 if cmd[0] == CMD_STOP: 53 self._cleanup() 54 break 55 elif cmd[0] == CMD_CONNECT: 56 self._do_connect(cmd[1]) 57 elif cmd[0] == CMD_DISCONNECT: 58 self._do_disconnect() 59 elif cmd[0] == CMD_POLL: 60 self._do_poll(cmd[1]) 61 elif cmd[0] == CMD_WRITE: 62 self._do_write(cmd[1]) 63 64 def _do_connect(self, port): 65 try: 66 self._transport = NfcSerialTransport(port, timeout=0.2) 67 if not self._transport.open(): 68 self.result_queue.put( 69 ('connected', False, "Failed to open serial port")) 70 return 71 self._reader = NfcReader(self._transport) 72 if not self._reader.begin(): 73 self._transport.close() 74 self.result_queue.put( 75 ('connected', False, "PN532 init failed")) 76 return 77 self.result_queue.put(('connected', True, '')) 78 except Exception as e: 79 self.result_queue.put(('connected', False, str(e))) 80 81 def _do_disconnect(self): 82 self._cleanup() 83 self.result_queue.put(('disconnected',)) 84 85 def _do_poll(self, source): 86 if self._reader is None or not self._reader.is_ready: 87 self.result_queue.put(('tag_data', source, None)) 88 return 89 try: 90 tag_data = self._reader.read_full_tag( 91 user_data_start=4, user_data_pages=128) 92 self.result_queue.put(('tag_data', source, tag_data)) 93 except Exception as e: 94 logger.debug("Poll error: %s", e) 95 self.result_queue.put(('tag_data', source, None)) 96 97 def _do_write(self, params): 98 if self._reader is None or not self._reader.is_ready: 99 self.result_queue.put( 100 ('write_result', False, "Reader not connected")) 101 return 102 try: 103 # Build NDEF data 104 ndef_bytes = build_battery_ndef( 105 sn=params['sn'], 106 year=params['year'], 107 note=params.get('note', ''), 108 uri=params.get('uri', 'http://www.raptacon.org'), 109 ) 110 111 # Detect tag (fresh target listing for write) 112 result = self._reader.read_passive_target() 113 if result is None: 114 self.result_queue.put( 115 ('write_result', False, "No tag detected")) 116 return 117 118 tg, uid = result 119 120 # Pad NDEF with zeros to clear any old data beyond it 121 write_data = bytearray(ndef_bytes) 122 write_data.extend(b'\x00' * 32) 123 124 # Write starting at page 4 125 if not self._reader.write_ntag_pages( 126 tg, 4, bytes(write_data)): 127 self.result_queue.put( 128 ('write_result', False, "Page write failed")) 129 return 130 131 self.result_queue.put(('write_result', True, '')) 132 133 except ValueError as e: 134 self.result_queue.put(('write_result', False, str(e))) 135 except Exception as e: 136 logger.exception("Write error") 137 self.result_queue.put(('write_result', False, str(e))) 138 139 def _cleanup(self): 140 if self._transport is not None: 141 try: 142 self._transport.close() 143 except Exception as e: 144 logger.debug("Error closing NFC transport: %s", e) 145 self._transport = None 146 self._reader = None
Background thread for all NFC serial I/O.
NfcWorker()
38 def __init__(self): 39 super().__init__(daemon=True) 40 self.cmd_queue = queue.Queue() 41 self.result_queue = queue.Queue() 42 self._transport = None 43 self._reader = None
This constructor should always be called with keyword arguments. Arguments are:
group should be None; reserved for future extension when a ThreadGroup class is implemented.
target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.
name is the thread name. By default, a unique name is constructed of the form "Thread-N" where N is a small decimal number.
args is a list or tuple of arguments for the target invocation. Defaults to ().
kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.
context is the contextvars.Context value to use for the thread. The default value is None, which means to check sys.flags.thread_inherit_context. If that flag is true, use a copy of the context of the caller. If false, use an empty context. To explicitly start with an empty context, pass a new instance of contextvars.Context(). To explicitly start with a copy of the current context, pass the value from contextvars.copy_context().
If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.
def
run(self):
45 def run(self): 46 while True: 47 try: 48 cmd = self.cmd_queue.get(timeout=0.5) 49 except queue.Empty: 50 continue 51 52 if cmd[0] == CMD_STOP: 53 self._cleanup() 54 break 55 elif cmd[0] == CMD_CONNECT: 56 self._do_connect(cmd[1]) 57 elif cmd[0] == CMD_DISCONNECT: 58 self._do_disconnect() 59 elif cmd[0] == CMD_POLL: 60 self._do_poll(cmd[1]) 61 elif cmd[0] == CMD_WRITE: 62 self._do_write(cmd[1])
Method representing the thread's activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object's constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
class
NfcToolApp(tkinter.Tk):
149class NfcToolApp(tk.Tk): 150 """Main NFC Battery Tag Tool window.""" 151 152 def __init__(self, project_root=None): 153 super().__init__() 154 self.title("NFC Battery Tag Tool - Team 3200") 155 self.geometry("650x550") 156 self.minsize(500, 400) 157 158 self._project_root = Path(project_root) if project_root else None 159 self._load_icon() 160 161 # Worker thread 162 self._worker = NfcWorker() 163 self._worker.start() 164 165 self._build_ui() 166 167 # Poll worker results 168 self._check_results() 169 170 self.protocol("WM_DELETE_WINDOW", self._on_close) 171 172 def _load_icon(self): 173 """Set the window icon from the Raptacon logo.""" 174 if self._project_root is None: 175 return 176 icon_path = self._project_root / "images" / "Raptacon3200-BG-BW.png" 177 if icon_path.exists(): 178 try: 179 self._icon_image = tk.PhotoImage(file=str(icon_path)) 180 self.iconphoto(True, self._icon_image) 181 except Exception as e: 182 logger.debug("Failed to load window icon: %s", e) 183 184 def _load_gear_logo(self, parent): 185 """Load the gear logo and return a Label widget, or None.""" 186 if self._project_root is None: 187 return None 188 logo_path = self._project_root / "images" / "raptacongear.png" 189 if not logo_path.exists(): 190 return None 191 try: 192 img = tk.PhotoImage(file=str(logo_path)) 193 # Subsample to ~64px (original is large) 194 w = img.width() 195 factor = max(1, w // 64) 196 img = img.subsample(factor, factor) 197 self._gear_image = img # prevent GC 198 label = ttk.Label(parent, image=img) 199 return label 200 except Exception as e: 201 logger.debug("Failed to load gear logo: %s", e) 202 return None 203 204 def _build_ui(self): 205 # Header with serial panel and logo 206 header = ttk.Frame(self) 207 header.pack(fill=tk.X, padx=8, pady=(8, 4)) 208 209 # Serial panel (left side) 210 self._serial_panel = SerialPanel( 211 header, 212 on_connect=self._on_connect, 213 on_disconnect=self._on_disconnect, 214 ) 215 self._serial_panel.pack(side=tk.LEFT, fill=tk.X, expand=True) 216 217 # Gear logo (right side) 218 logo = self._load_gear_logo(header) 219 if logo: 220 logo.pack(side=tk.RIGHT, padx=(8, 0)) 221 222 ttk.Separator(self, orient=tk.HORIZONTAL).pack(fill=tk.X, padx=8) 223 224 # Tabs 225 self._notebook = ttk.Notebook(self) 226 227 self._scan_tab = ScanTab( 228 self._notebook, request_poll=self._request_poll) 229 self._read_tab = ReadTab( 230 self._notebook, request_poll=self._request_poll) 231 self._write_tab = WriteTab( 232 self._notebook, 233 request_poll=self._request_poll, 234 request_write=self._request_write, 235 ) 236 237 self._notebook.add(self._scan_tab, text="Scan") 238 self._notebook.add(self._read_tab, text="Read") 239 self._notebook.add(self._write_tab, text="Write Battery Tag") 240 241 self._notebook.pack(fill=tk.BOTH, expand=True, padx=8, pady=(4, 8)) 242 self._notebook.bind("<<NotebookTabChanged>>", self._on_tab_changed) 243 244 self._tabs = [self._scan_tab, self._read_tab, self._write_tab] 245 self._active_tab_idx = 0 246 247 def _on_tab_changed(self, event): 248 # Notify old tab 249 if 0 <= self._active_tab_idx < len(self._tabs): 250 self._tabs[self._active_tab_idx].on_tab_deselected() 251 252 # Notify new tab 253 idx = self._notebook.index(self._notebook.select()) 254 self._active_tab_idx = idx 255 if 0 <= idx < len(self._tabs): 256 self._tabs[idx].on_tab_selected() 257 258 def _on_connect(self, port): 259 self._worker.cmd_queue.put((CMD_CONNECT, port)) 260 261 def _on_disconnect(self): 262 # Stop any tab polling 263 for tab in self._tabs: 264 tab.on_tab_deselected() 265 self._worker.cmd_queue.put((CMD_DISCONNECT,)) 266 267 def _request_poll(self, source): 268 """Tabs call this to request a tag poll.""" 269 if not self._serial_panel.is_connected: 270 return 271 self._worker.cmd_queue.put((CMD_POLL, source)) 272 273 def _request_write(self, sn, year, note='', uri=''): 274 """Write tab calls this to perform the actual write.""" 275 self._worker.cmd_queue.put((CMD_WRITE, { 276 'sn': sn, 'year': year, 'note': note, 'uri': uri, 277 })) 278 279 def _check_results(self): 280 """Poll worker result queue and dispatch to tabs.""" 281 try: 282 while True: 283 result = self._worker.result_queue.get_nowait() 284 self._handle_result(result) 285 except queue.Empty: 286 pass 287 self.after(100, self._check_results) 288 289 def _handle_result(self, result): 290 cmd = result[0] 291 292 if cmd == 'connected': 293 success, error_msg = result[1], result[2] 294 self._serial_panel.set_connected(success, error_msg) 295 if success: 296 # Activate current tab's polling 297 idx = self._active_tab_idx 298 if 0 <= idx < len(self._tabs): 299 self._tabs[idx].on_tab_selected() 300 301 elif cmd == 'disconnected': 302 self._serial_panel.set_disconnected() 303 304 elif cmd == 'tag_data': 305 source, tag_data = result[1], result[2] 306 if source == 'scan': 307 self._scan_tab.update_tag_data(tag_data) 308 elif source == 'read': 309 self._read_tab.update_tag_data(tag_data) 310 elif source == 'write_search': 311 self._write_tab.on_search_result(tag_data) 312 elif source == 'write_verify': 313 self._write_tab.on_verify_result(tag_data) 314 315 elif cmd == 'write_result': 316 success, error_msg = result[1], result[2] 317 self._write_tab.on_write_result(success, error_msg) 318 319 def _on_close(self): 320 # Stop all tab polling 321 for tab in self._tabs: 322 tab.on_tab_deselected() 323 # Stop worker 324 self._worker.cmd_queue.put((CMD_STOP,)) 325 self._worker.join(timeout=2.0) 326 self.destroy()
Main NFC Battery Tag Tool window.
NfcToolApp(project_root=None)
152 def __init__(self, project_root=None): 153 super().__init__() 154 self.title("NFC Battery Tag Tool - Team 3200") 155 self.geometry("650x550") 156 self.minsize(500, 400) 157 158 self._project_root = Path(project_root) if project_root else None 159 self._load_icon() 160 161 # Worker thread 162 self._worker = NfcWorker() 163 self._worker.start() 164 165 self._build_ui() 166 167 # Poll worker results 168 self._check_results() 169 170 self.protocol("WM_DELETE_WINDOW", self._on_close)
Return a new top level widget on screen SCREENNAME. A new Tcl interpreter will be created. BASENAME will be used for the identification of the profile file (see readprofile). It is constructed from sys.argv[0] without extensions if None is given. CLASSNAME is the name of the widget class.