Read current daemon status with corruption recovery. Returns: DaemonStatus object (or default status if file doesn't exist or corrupted)
()
| 103 | |
| 104 | |
| 105 | def read_status_file() -> DaemonStatus: |
| 106 | """Read current daemon status with corruption recovery. |
| 107 | |
| 108 | Returns: |
| 109 | DaemonStatus object (or default status if file doesn't exist or corrupted) |
| 110 | """ |
| 111 | if not STATUS_FILE.exists(): |
| 112 | return DaemonStatus( |
| 113 | state=DaemonState.UNKNOWN, |
| 114 | message="Status file not found", |
| 115 | updated_at=time.time(), |
| 116 | ) |
| 117 | |
| 118 | try: |
| 119 | with open(STATUS_FILE, "r") as f: |
| 120 | data = json.load(f) |
| 121 | |
| 122 | # Parse into typed DaemonStatus |
| 123 | return DaemonStatus.from_dict(data) |
| 124 | |
| 125 | except (json.JSONDecodeError, ValueError): |
| 126 | # Corrupted JSON - return default status |
| 127 | return DaemonStatus( |
| 128 | state=DaemonState.UNKNOWN, |
| 129 | message="Status file corrupted (invalid JSON)", |
| 130 | updated_at=time.time(), |
| 131 | ) |
| 132 | except KeyboardInterrupt as ki: |
| 133 | handle_keyboard_interrupt(ki) |
| 134 | raise |
| 135 | except Exception: |
| 136 | return DaemonStatus( |
| 137 | state=DaemonState.UNKNOWN, |
| 138 | message="Failed to read status", |
| 139 | updated_at=time.time(), |
| 140 | ) |
| 141 | |
| 142 | |
| 143 | def write_request_file(request: PackageRequest) -> None: |
no test coverage detected