|
| 1 | +import os |
| 2 | +import yaml |
| 3 | +import requests |
| 4 | + |
| 5 | +class Prusa: |
| 6 | + """ |
| 7 | + This is not a game, but a demo that shows the status of a Prusa printer |
| 8 | + using the PrusaLink API. It is designed for an 'SSSMini' which is a 1x1 screen. |
| 9 | + To use this demo, you need to have a Prusa printer with PrusaLink enabled |
| 10 | + and the API key set up in the config.yaml file. |
| 11 | + Example config.yaml: |
| 12 | + prusalink_ip: "<PRUSA_LINK_IP>" |
| 13 | + api_key: "<API_KEY>" |
| 14 | + You can also add the service file to your systemd services to run it automatically. |
| 15 | + Ensure that your system is connected to the same network as the Prusa printer, |
| 16 | + and you have a connection to the screen. |
| 17 | + """ |
| 18 | + |
| 19 | + demo_time = None # Number of seconds or None if its game |
| 20 | + |
| 21 | + def __init__(self, input_queue, output_queue, screen): |
| 22 | + """ |
| 23 | + Constructor |
| 24 | +
|
| 25 | + Args: |
| 26 | + input_queue (Queue): Queue for user input |
| 27 | + output_queue (Queue): Queue for game output |
| 28 | + screen (Screen): Screen object |
| 29 | + """ |
| 30 | + # Provide the framerate in frames/seconds and the amount of time of the demo in seconds |
| 31 | + self.frame_rate = 0.5 |
| 32 | + |
| 33 | + self.input_queue = input_queue |
| 34 | + self.output_queue = output_queue |
| 35 | + self.screen = screen |
| 36 | + |
| 37 | + # init demo/game specific variables here |
| 38 | + config_path = os.path.join(os.path.dirname(__file__), 'config.yaml') |
| 39 | + with open(config_path, 'r') as f: |
| 40 | + config = yaml.safe_load(f) |
| 41 | + self.prusalink_ip = config.get('prusalink_ip') |
| 42 | + self.api_key = config.get('api_key') |
| 43 | + self.URL = f"http://{self.prusalink_ip}/api/v1/job" |
| 44 | + self.HEADERS = { |
| 45 | + "X-Api-Key": self.api_key, |
| 46 | + } |
| 47 | + self.response = None |
| 48 | + self.data = None |
| 49 | + self.state = "N/A" |
| 50 | + self.filename = "N/A" |
| 51 | + self.progress = 0 |
| 52 | + self.time_elapsed = "N/A" |
| 53 | + self.time_left = "N/A" |
| 54 | + self.minutes_left = "N/A" |
| 55 | + self.display_filename = "N/A" |
| 56 | + self.display_status = "N/A" |
| 57 | + self.display_time_elapsed = "N/A" |
| 58 | + self.display_time_left = "N/A" |
| 59 | + self.display_progress = 0 |
| 60 | + |
| 61 | + |
| 62 | + def get_stats(self): |
| 63 | + """ |
| 64 | + Returns static information. |
| 65 | + """ |
| 66 | + try: |
| 67 | + self.response = requests.get(self.URL, headers=self.HEADERS, timeout=5) |
| 68 | + self.response.raise_for_status() |
| 69 | + |
| 70 | + # Check if response has content before parsing JSON |
| 71 | + if not self.response.text.strip(): |
| 72 | + # print("Empty response received") |
| 73 | + return self._get_default_stats() |
| 74 | + |
| 75 | + # Check content type |
| 76 | + content_type = self.response.headers.get('content-type', '') |
| 77 | + if 'application/json' not in content_type: |
| 78 | + print(f"Unexpected content type: {content_type}") |
| 79 | + print(f"Response body: {self.response.text[:200]}") # First 200 chars |
| 80 | + return self._get_default_stats() |
| 81 | + |
| 82 | + self.data = self.response.json() |
| 83 | + |
| 84 | + self.state = self.data.get("state", "N/A") |
| 85 | + self.filename = self.data.get("file", {}).get("display_name", "N/A") |
| 86 | + self.progress = self.data.get("progress", 0) |
| 87 | + self.time_elapsed = self.data.get("time_printing", "N/A") |
| 88 | + self.time_left = self.data.get("time_remaining", "N/A") |
| 89 | + self.minutes_left = self.time_left // 60 if isinstance(self.time_left, int) else "N/A" |
| 90 | + |
| 91 | + except requests.RequestException as e: |
| 92 | + # print(f"Request failed: {e}") |
| 93 | + return self._get_default_stats() |
| 94 | + except ValueError as e: |
| 95 | + # print(f"JSON parsing failed: {e}") |
| 96 | + # print(f"Response status: {self.response.status_code}") |
| 97 | + # print(f"Response headers: {dict(self.response.headers)}") |
| 98 | + # print(f"Response body: {self.response.text[:200]}") # First 200 chars |
| 99 | + return self._get_default_stats() |
| 100 | + |
| 101 | + return { |
| 102 | + "state": self.state, |
| 103 | + "filename": self.filename, |
| 104 | + "progress": self.progress, |
| 105 | + "time_elapsed": self.time_elapsed, |
| 106 | + "time_left": self.time_left, |
| 107 | + "minutes_left": self.minutes_left, |
| 108 | + } |
| 109 | + |
| 110 | + def _get_default_stats(self): |
| 111 | + """Return default stats when API call fails""" |
| 112 | + return { |
| 113 | + "state": "IDLE", |
| 114 | + "filename": "N/A", |
| 115 | + "progress": 0, |
| 116 | + "time_elapsed": "N/A", |
| 117 | + "time_left": "N/A", |
| 118 | + "minutes_left": "N/A", |
| 119 | + } |
| 120 | + |
| 121 | + def run(self): |
| 122 | + """Main loop for the demo""" |
| 123 | + # Create generator here |
| 124 | + while True: |
| 125 | + self.stats = self.get_stats() |
| 126 | + |
| 127 | + # print("=== Printer Status ===") |
| 128 | + # print(f"time left: {self.stats['time_left']}") |
| 129 | + # print(f"Stats: {self.stats}") |
| 130 | + # draw the filename |
| 131 | + self.display_status = self.stats['state'] |
| 132 | + # print(f"Status: {self.display_status}") |
| 133 | + if self.stats['state'] == "IDLE" and int(self.display_time_elapsed[0:8].strip()) > 0: |
| 134 | + self.display_progress = 0 |
| 135 | + self.display_time_elapsed = "0" |
| 136 | + self.display_time_left = "0" |
| 137 | + self.display_filename = "N/A" |
| 138 | + self.screen.clear() |
| 139 | + print("hit") |
| 140 | + |
| 141 | + |
| 142 | + self.screen.draw_text( |
| 143 | + self.screen.x_width // 2 - 8, |
| 144 | + self.screen.y_height // 2 - 6, |
| 145 | + self.display_status, |
| 146 | + push=True, |
| 147 | + ) |
| 148 | + |
| 149 | + if self.stats['state'] == "IDLE": |
| 150 | + yield |
| 151 | + continue |
| 152 | + |
| 153 | + self.display_filename = self.stats['filename'][:16].upper() |
| 154 | + # print(f"Filename: {self.display_filename}") |
| 155 | + self.screen.draw_text( |
| 156 | + self.screen.x_width // 2 - 8, |
| 157 | + self.screen.y_height // 2 - 4, |
| 158 | + self.display_filename, |
| 159 | + push=True, |
| 160 | + ) |
| 161 | + |
| 162 | + |
| 163 | + # print(f"Time left: {self.stats['minutes_left']} minutes") |
| 164 | + self.display_time_elapsed = f"{str(self.stats['time_elapsed']):>5} ELAPSED" |
| 165 | + self.screen.draw_text( |
| 166 | + self.screen.x_width // 2 - 8, |
| 167 | + self.screen.y_height // 2 - 2, |
| 168 | + self.display_time_elapsed, |
| 169 | + push=True, |
| 170 | + ) |
| 171 | + |
| 172 | + # print(f"Time left: {self.stats['minutes_left']} minutes") |
| 173 | + self.display_time_left = f"{str(self.stats['time_left']):>5} LEFT{self.stats['minutes_left']/60:>4.1f}" |
| 174 | + self.screen.draw_text( |
| 175 | + self.screen.x_width // 2 - 8, |
| 176 | + self.screen.y_height // 2, |
| 177 | + self.display_time_left, |
| 178 | + push=True, |
| 179 | + ) |
| 180 | + |
| 181 | + # Map progress (0-100) to a value between 0 and 16 |
| 182 | + percentage_complete = int((self.stats['progress'] / 100) * 16) |
| 183 | + percentage_complete = max(0, min(percentage_complete, 16)) |
| 184 | + # print(f"Progress: {self.stats['progress']}%, (mapped to {percentage_complete})") |
| 185 | + self.display_progress = f"{int(self.stats['progress']):>5} PROGRESS" |
| 186 | + self.screen.draw_text( |
| 187 | + self.screen.x_width // 2 - 8, |
| 188 | + self.screen.y_height // 2 + 2, |
| 189 | + self.display_progress, |
| 190 | + push=True, |
| 191 | + ) |
| 192 | + for i in range(percentage_complete): |
| 193 | + self.screen.draw_pixel( |
| 194 | + self.screen.x_width // 2 - 8 + i, |
| 195 | + self.screen.y_height // 2 + 5, |
| 196 | + 15, |
| 197 | + combine=False, |
| 198 | + push=True, |
| 199 | + ) |
| 200 | + yield |
| 201 | + |
| 202 | + def stop(self): |
| 203 | + """Reset the state of the demo if needed, else leave blank""" |
| 204 | + pass |
0 commit comments