Platform-specific build lock.
| 175 | |
| 176 | |
| 177 | class PlatformLock: |
| 178 | """Platform-specific build lock.""" |
| 179 | |
| 180 | def __init__(self, lock_file: Path) -> None: |
| 181 | self.lock_file_path = lock_file |
| 182 | # Use global lock DB for platform locks |
| 183 | home_dir = Path.home() |
| 184 | db_path = home_dir / ".fastled" / "locks.db" |
| 185 | self._db = LockDatabase(db_path) |
| 186 | self._lock_name = f"platform:{lock_file}" |
| 187 | self.is_locked = False |
| 188 | |
| 189 | def _check_stale_lock(self) -> bool: |
| 190 | """Check if lock is stale (dead process) and remove if stale. |
| 191 | |
| 192 | Returns: |
| 193 | True if lock was stale and removed, False otherwise |
| 194 | """ |
| 195 | try: |
| 196 | if self._db.is_lock_stale(self._lock_name): |
| 197 | print( |
| 198 | f"Detected stale platform lock at {self.lock_file_path}. Removing..." |
| 199 | ) |
| 200 | if self._db.break_stale_lock(self._lock_name): |
| 201 | print(f"Removed stale lock file: {self.lock_file_path}") |
| 202 | return True |
| 203 | return False |
| 204 | except KeyboardInterrupt as ki: |
| 205 | handle_keyboard_interrupt(ki) |
| 206 | raise |
| 207 | except Exception as e: |
| 208 | print(f"Warning: Could not check for stale lock: {e}") |
| 209 | return False |
| 210 | |
| 211 | def acquire(self) -> None: |
| 212 | """Acquire the platform lock.""" |
| 213 | if self.is_locked: |
| 214 | return # Already acquired |
| 215 | |
| 216 | my_pid = os.getpid() |
| 217 | hostname = platform.node() |
| 218 | start_time = time.time() |
| 219 | warning_shown = False |
| 220 | last_stale_check = 0.0 |
| 221 | |
| 222 | while True: |
| 223 | # Check for keyboard interrupt |
| 224 | if is_interrupted(): |
| 225 | print("\nKeyboardInterrupt: Aborting platform lock acquisition") |
| 226 | raise KeyboardInterrupt() |
| 227 | |
| 228 | elapsed = time.time() - start_time |
| 229 | |
| 230 | # Check for stale lock periodically (every ~1 second) |
| 231 | if elapsed - last_stale_check >= 1.0: |
| 232 | if self._check_stale_lock(): |
| 233 | print("Stale platform lock removed, retrying acquisition...") |
| 234 | last_stale_check = elapsed |