6510b2ec6b
Use f-strings Avoid unnecessary casts to str()bool()int() Ensure file closure Merge nested ifs Simplify for-assigns-appends with comprehensions and internal functions Avoid shadowing internal function names Initialize variables Return value directly instead of assign then return Make some methods static
372 lines
14 KiB
Python
372 lines
14 KiB
Python
import gi
|
|
import logging
|
|
|
|
gi.require_version("Gtk", "3.0")
|
|
from gi.repository import Gdk, GLib
|
|
|
|
|
|
class Printer:
|
|
def __init__(self, printer_info, data, state_execute_cb):
|
|
self.klipper = {"version": printer_info['software_version']}
|
|
self.tempstore = None
|
|
self.config = None
|
|
self.state = "disconnected"
|
|
self.state_cb = state_execute_cb
|
|
self.power_devices = {}
|
|
self.store_timeout = False
|
|
self.data = data
|
|
self.devices = {}
|
|
self.power_devices = {}
|
|
self.state_callbacks = {
|
|
"disconnected": None,
|
|
"error": None,
|
|
"paused": None,
|
|
"printing": None,
|
|
"ready": None,
|
|
"startup": None,
|
|
"shutdown": None
|
|
}
|
|
self.tools = []
|
|
self.extrudercount = 0
|
|
self.tempdevcount = 0
|
|
self.fancount = 0
|
|
|
|
def reset(self):
|
|
GLib.source_remove(self.store_timeout)
|
|
self.store_timeout = None
|
|
self.state = None
|
|
self.state_cb = None
|
|
self.data = None
|
|
self.devices = None
|
|
self.power_devices = None
|
|
self.state_callbacks = None
|
|
self.tools = None
|
|
self.extrudercount = None
|
|
self.tempdevcount = None
|
|
self.fancount = None
|
|
self.config = None
|
|
self.klipper = None
|
|
self.tempstore = None
|
|
|
|
def reinit(self, printer_info, data):
|
|
logging.debug(f"Moonraker object status: {data}")
|
|
self.config = data['configfile']['config']
|
|
self.extrudercount = 0
|
|
self.tempdevcount = 0
|
|
self.fancount = 0
|
|
self.tools = []
|
|
self.devices = {}
|
|
self.data = data
|
|
self.klipper = {}
|
|
self.tempstore = {}
|
|
if not self.store_timeout:
|
|
self.store_timeout = GLib.timeout_add_seconds(1, self._update_temp_store)
|
|
|
|
self.klipper = {
|
|
"version": printer_info['software_version']
|
|
}
|
|
|
|
for x in self.config.keys():
|
|
if x[:8] == "extruder":
|
|
self.tools.append(x)
|
|
self.tools = sorted(self.tools)
|
|
self.extrudercount += 1
|
|
if x.startswith('extruder_stepper'):
|
|
continue
|
|
self.devices[x] = {
|
|
"temperature": 0,
|
|
"target": 0
|
|
}
|
|
if x == 'heater_bed' \
|
|
or x.startswith('heater_generic ') \
|
|
or x.startswith('temperature_sensor ') \
|
|
or x.startswith('temperature_fan '):
|
|
self.devices[x] = {
|
|
"temperature": 0,
|
|
"target": 0
|
|
}
|
|
# Support for hiding devices by name
|
|
if not " ".join(x.split(" ")[1:]).startswith("_"):
|
|
self.tempdevcount += 1
|
|
if x == 'fan' \
|
|
or x.startswith('controller_fan ') \
|
|
or x.startswith('heater_fan ') \
|
|
or x.startswith('fan_generic '):
|
|
# Support for hiding devices by name
|
|
if not " ".join(x.split(" ")[1:]).startswith("_"):
|
|
self.fancount += 1
|
|
if x.startswith('bed_mesh '):
|
|
r = self.config[x]
|
|
r['x_count'] = int(r['x_count'])
|
|
r['y_count'] = int(r['y_count'])
|
|
r['max_x'] = float(r['max_x'])
|
|
r['min_x'] = float(r['min_x'])
|
|
r['max_y'] = float(r['max_y'])
|
|
r['min_y'] = float(r['min_y'])
|
|
r['points'] = [[float(j.strip()) for j in i.split(",")] for i in r['points'].strip().split("\n")]
|
|
self.process_update(data)
|
|
|
|
logging.info(f"Klipper version: {self.klipper['version']}")
|
|
logging.info(f"# Extruders: {self.extrudercount}")
|
|
logging.info(f"# Temperature devices: {self.tempdevcount}")
|
|
logging.info(f"# Fans: {self.fancount}")
|
|
|
|
def process_update(self, data):
|
|
for x in (self.get_tools() + self.get_heaters() + self.get_filament_sensors()):
|
|
if x in data:
|
|
for i in data[x]:
|
|
self.set_dev_stat(x, i, data[x][i])
|
|
|
|
for x in data:
|
|
if x == "configfile":
|
|
continue
|
|
if x not in self.data:
|
|
self.data[x] = {}
|
|
self.data[x].update(data[x])
|
|
|
|
if "webhooks" in data or "idle_timeout" in data or "print_stats" in data or "pause_resume" in data:
|
|
self.evaluate_state()
|
|
|
|
def get_updates(self):
|
|
updates = self.data.copy()
|
|
updates.update(self.devices)
|
|
return updates
|
|
|
|
def evaluate_state(self):
|
|
wh_state = self.data['webhooks']['state'].lower() # possible values: startup, ready, shutdown, error
|
|
|
|
if wh_state == "ready":
|
|
new_state = "ready"
|
|
if self.data['pause_resume']['is_paused']:
|
|
new_state = "paused"
|
|
elif self.data['print_stats']:
|
|
print_state = self.data['print_stats']['state'].lower() # complete, error, paused, printing, standby
|
|
if print_state == "paused":
|
|
new_state = "paused"
|
|
if self.data['idle_timeout']:
|
|
idle_state = self.data['idle_timeout']['state'].lower() # idle, printing, ready
|
|
if idle_state == "printing":
|
|
if print_state == "complete":
|
|
new_state = "ready"
|
|
elif print_state != "printing": # Not printing a file, toolhead moving
|
|
new_state = "busy"
|
|
else:
|
|
new_state = "printing"
|
|
|
|
if new_state != "busy":
|
|
self.change_state(new_state)
|
|
else:
|
|
self.change_state(wh_state)
|
|
|
|
def process_power_update(self, data):
|
|
if data['device'] in self.power_devices:
|
|
self.power_devices[data['device']]['status'] = data['status']
|
|
|
|
def change_state(self, state):
|
|
if state == self.state or state not in list(self.state_callbacks):
|
|
return
|
|
|
|
logging.debug(f"Changing state from '{self.state}' to '{state}'")
|
|
prev_state = self.state
|
|
self.state = state
|
|
if self.state_callbacks[state] is not None:
|
|
logging.debug(f"Adding callback for state: {state}")
|
|
Gdk.threads_add_idle(
|
|
GLib.PRIORITY_HIGH_IDLE,
|
|
self.state_cb,
|
|
self.state_callbacks[state],
|
|
prev_state
|
|
)
|
|
|
|
def configure_power_devices(self, data):
|
|
self.power_devices = {}
|
|
|
|
logging.debug(f"Processing power devices: {data}")
|
|
for x in data['devices']:
|
|
self.power_devices[x['device']] = {
|
|
"status": "on" if x['status'] == "on" else "off"
|
|
}
|
|
logging.debug(f"Power devices: {self.power_devices}")
|
|
|
|
def get_config_section_list(self, search=""):
|
|
return [i for i in list(self.config) if i.startswith(search)] if hasattr(self, "config") else []
|
|
|
|
def get_config_section(self, section):
|
|
return self.config[section] if section in self.config else False
|
|
|
|
def get_data(self):
|
|
return self.data
|
|
|
|
def get_fans(self):
|
|
fans = []
|
|
if self.config_section_exists("fan"):
|
|
fans.append("fan")
|
|
fan_types = ["controller_fan", "fan_generic", "heater_fan"]
|
|
for fan_type in fan_types:
|
|
fans.extend(iter(self.get_config_section_list(f"{fan_type} ")))
|
|
return fans
|
|
|
|
def get_gcode_macros(self):
|
|
return self.get_config_section_list("gcode_macro ")
|
|
|
|
def get_heaters(self):
|
|
heaters = []
|
|
if self.has_heated_bed():
|
|
heaters.append("heater_bed")
|
|
heaters.extend(iter(self.get_config_section_list("heater_generic ")))
|
|
heaters.extend(iter(self.get_config_section_list("temperature_sensor ")))
|
|
heaters.extend(iter(self.get_config_section_list("temperature_fan ")))
|
|
return heaters
|
|
|
|
def get_filament_sensors(self):
|
|
sensors = list(self.get_config_section_list("filament_switch_sensor "))
|
|
sensors.extend(iter(self.get_config_section_list("filament_motion_sensor ")))
|
|
return sensors
|
|
|
|
def get_printer_status_data(self):
|
|
data = {
|
|
"printer": {
|
|
"extruders": {"count": self.extrudercount},
|
|
"temperature_devices": {"count": self.tempdevcount},
|
|
"fans": {"count": self.fancount},
|
|
"bltouch": self.config_section_exists("bltouch"),
|
|
"gcode_macros": {"count": len(self.get_gcode_macros())},
|
|
"idle_timeout": self.get_stat("idle_timeout").copy(),
|
|
"pause_resume": {"is_paused": self.state == "paused"},
|
|
"power_devices": {"count": len(self.get_power_devices())},
|
|
"probe": self.config_section_exists("probe"),
|
|
"firmware_retraction": self.config_section_exists("firmware_retraction"),
|
|
"input_shaper": self.config_section_exists("input_shaper")
|
|
}
|
|
}
|
|
|
|
sections = ["bed_mesh", "bltouch", "probe", "quad_gantry_level", "z_tilt"]
|
|
for section in sections:
|
|
if self.config_section_exists(section):
|
|
data["printer"][section] = self.get_config_section(section).copy()
|
|
|
|
return data
|
|
|
|
def get_klipper_version(self):
|
|
return self.klipper['version']
|
|
|
|
def get_power_devices(self):
|
|
return list(self.power_devices)
|
|
|
|
def get_power_device_status(self, device):
|
|
if device not in self.power_devices:
|
|
return
|
|
return self.power_devices[device]['status']
|
|
|
|
def get_stat(self, stat, substat=None):
|
|
if self.data is None or stat not in self.data:
|
|
return {}
|
|
if substat is not None:
|
|
return self.data[stat][substat] if substat in self.data[stat] else {}
|
|
return self.data[stat]
|
|
|
|
def get_state(self):
|
|
return self.state
|
|
|
|
def set_dev_temps(self, dev, temp, target=None):
|
|
if dev in self.devices:
|
|
self.devices[dev]['temperature'] = temp
|
|
if target is not None:
|
|
self.devices[dev]['target'] = target
|
|
|
|
def get_dev_stats(self, dev):
|
|
return self.devices[dev] if dev in self.devices else None
|
|
|
|
def get_dev_stat(self, dev, stat):
|
|
if dev in self.devices and stat in self.devices[dev]:
|
|
return self.devices[dev][stat]
|
|
return None
|
|
|
|
def get_fan_speed(self, fan="fan", speed=None):
|
|
if fan not in self.config or fan not in self.data:
|
|
logging.debug(f"Error getting {fan} config")
|
|
return speed if speed is not None else 0
|
|
if speed is None and "speed" in self.data[fan]:
|
|
speed = self.data[fan]["speed"]
|
|
if 'max_power' in self.config[fan]:
|
|
max_power = float(self.config[fan]['max_power'])
|
|
if max_power > 0:
|
|
speed = speed / max_power
|
|
if 'off_below' in self.config[fan]:
|
|
off_below = float(self.config[fan]['off_below'])
|
|
if speed < off_below:
|
|
speed = 0
|
|
return speed
|
|
|
|
def get_extruder_count(self):
|
|
return self.extrudercount
|
|
|
|
def get_temp_store_devices(self):
|
|
if self.tempstore is not None:
|
|
return list(self.tempstore)
|
|
|
|
def get_temp_store_device_has_target(self, device):
|
|
return device in self.tempstore and "targets" in self.tempstore[device]
|
|
|
|
def get_temp_store(self, device, section=False, results=0):
|
|
if device not in self.tempstore:
|
|
return False
|
|
|
|
if section is not False:
|
|
if section not in self.tempstore[device]:
|
|
return False
|
|
if results == 0 or results >= len(self.tempstore[device][section]):
|
|
return self.tempstore[device][section]
|
|
return self.tempstore[device][section][-results:]
|
|
|
|
temp = {}
|
|
for section in self.tempstore[device]:
|
|
if results == 0 or results >= len(self.tempstore[device][section]):
|
|
temp[section] = self.tempstore[device][section]
|
|
temp[section] = self.tempstore[device][section][-results:]
|
|
return temp
|
|
|
|
def get_tools(self):
|
|
return self.tools
|
|
|
|
def get_tool_number(self, tool):
|
|
return self.tools.index(tool)
|
|
|
|
def has_heated_bed(self):
|
|
if "heater_bed" in self.devices:
|
|
return True
|
|
|
|
def init_temp_store(self, result):
|
|
for dev in result:
|
|
self.tempstore[dev] = {}
|
|
if "targets" in result[dev]:
|
|
self.tempstore[dev]["targets"] = result[dev]["targets"]
|
|
if "temperatures" in result[dev]:
|
|
self.tempstore[dev]["temperatures"] = result[dev]["temperatures"]
|
|
logging.info(f"Temp store: {list(self.tempstore)}")
|
|
|
|
def config_section_exists(self, section):
|
|
return section in self.get_config_section_list()
|
|
|
|
def set_callbacks(self, callbacks):
|
|
for name, cb in callbacks.items():
|
|
if name in list(self.state_callbacks):
|
|
self.state_callbacks[name] = cb
|
|
|
|
def set_dev_stat(self, dev, stat, value):
|
|
if dev not in self.devices:
|
|
return
|
|
|
|
self.devices[dev][stat] = value
|
|
|
|
def _update_temp_store(self):
|
|
for device in self.tempstore:
|
|
for x in self.tempstore[device]:
|
|
if len(self.tempstore[device][x]) >= 1200:
|
|
self.tempstore[device][x].pop(0)
|
|
temp = self.get_dev_stat(device, x[:-1])
|
|
if temp is None:
|
|
temp = 0
|
|
self.tempstore[device][x].append(round(temp, 2))
|
|
return True
|