1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
# Generic Filament Sensor Module
#
# Copyright (C) 2019 Eric Callahan <arksine.code@gmail.com>
#
# This file may be distributed under the terms of the GNU GPLv3 license.
import logging
class RunoutHelper:
def __init__(self, config):
self.name = config.get_name().split()[-1]
self.printer = config.get_printer()
self.reactor = self.printer.get_reactor()
self.gcode = self.printer.lookup_object('gcode')
self.runout_pause = config.getboolean('pause_on_runout', True)
if self.runout_pause:
self.printer.try_load_module(config, 'pause_resume')
self.runout_gcode = self.insert_gcode = None
gcode_macro = self.printer.try_load_module(config, 'gcode_macro')
if self.runout_pause or config.get('runout_gcode', None) is not None:
self.runout_gcode = gcode_macro.load_template(
config, 'runout_gcode', '')
if config.get('insert_gcode', None) is not None:
self.insert_gcode = gcode_macro.load_template(
config, 'insert_gcode')
self.pause_delay = config.getfloat('pause_delay', .5, above=.0)
self.event_delay = config.getfloat('event_delay', 3., above=0.)
self.min_event_systime = self.reactor.NEVER
self.filament_present = False
self.runout_enabled = False
self.insert_enabled = self.insert_gcode is not None
self.sensor_enabled = True
self.print_status = "idle"
self.printer.register_event_handler("klippy:ready", self._handle_ready)
self.printer.register_event_handler(
"idle_timeout:idle",
(lambda e, s=self, st="idle": s._update_print_status(e, st)))
self.printer.register_event_handler(
"idle_timeout:ready",
(lambda e, s=self, st="ready": s._update_print_status(e, st)))
self.printer.register_event_handler(
"idle_timeout:printing",
(lambda e, s=self, st="printing": s._update_print_status(e, st)))
self.gcode.register_mux_command(
"QUERY_FILAMENT_SENSOR", "SENSOR", self.name,
self.cmd_QUERY_FILAMENT_SENSOR,
desc=self.cmd_QUERY_FILAMENT_SENSOR_help)
self.gcode.register_mux_command(
"SET_FILAMENT_SENSOR", "SENSOR", self.name,
self.cmd_SET_FILAMENT_SENSOR,
desc=self.cmd_SET_FILAMENT_SENSOR_help)
def _handle_ready(self):
self.min_event_systime = self.reactor.monotonic() + 2.
def _update_print_status(self, eventtime, status):
if status == "printing":
runout_en = self.runout_gcode is not None
self.set_enable(runout_en, False)
else:
insert_en = self.insert_gcode is not None
self.set_enable(False, insert_en)
def _runout_event_handler(self, eventtime):
# Pausing from inside an event requires that the pause portion
# of pause_resume execute immediately.
pause_prefix = ""
if self.runout_pause:
pause_resume = self.printer.lookup_object('pause_resume')
pause_resume.send_pause_command()
pause_prefix = "PAUSE\n"
self.printer.get_reactor().pause(eventtime + self.pause_delay)
self._exec_gcode(pause_prefix, self.runout_gcode)
def _insert_event_handler(self, eventtime):
self._exec_gcode("", self.insert_gcode)
def _exec_gcode(self, prefix, template):
try:
self.gcode.run_script(prefix + template.render() + "\nM400")
except Exception:
logging.exception("Script running error")
self.min_event_systime = self.reactor.monotonic() + self.event_delay
def set_enable(self, runout, insert):
if runout and insert:
# both cannot be enabled
insert = False
self.runout_enabled = runout
self.insert_enabled = insert
def note_filament_present(self, is_filament_present):
if is_filament_present == self.filament_present:
return
self.filament_present = is_filament_present
eventtime = self.reactor.monotonic()
if eventtime < self.min_event_systime or not self.sensor_enabled:
# do not process during the initialization time, duplicates,
# during the event delay time, while an event is running, or
# when the sensor is disabled
return
if is_filament_present:
if self.insert_enabled:
# insert detected
self.min_event_systime = self.reactor.NEVER
logging.info(
"Filament Sensor %s: insert event detected, Time %.2f" %
(self.name, eventtime))
self.reactor.register_callback(self._insert_event_handler)
elif self.runout_enabled:
# runout detected
self.min_event_systime = self.reactor.NEVER
logging.info(
"Filament Sensor %s: runout event detected, Time %.2f" %
(self.name, eventtime))
self.reactor.register_callback(self._runout_event_handler)
def get_status(self, eventtime):
return {"filament_detected": bool(self.filament_present)}
cmd_QUERY_FILAMENT_SENSOR_help = "Query the status of the Filament Sensor"
def cmd_QUERY_FILAMENT_SENSOR(self, params):
if self.filament_present:
msg = "Filament Sensor %s: filament detected" % (self.name)
else:
msg = "Filament Sensor %s: filament not detected" % (self.name)
self.gcode.respond_info(msg)
cmd_SET_FILAMENT_SENSOR_help = "Sets the filament sensor on/off"
def cmd_SET_FILAMENT_SENSOR(self, params):
self.sensor_enabled = self.gcode.get_int("ENABLE", params, 1)
class SwitchSensor:
def __init__(self, config):
printer = config.get_printer()
buttons = printer.try_load_module(config, 'buttons')
switch_pin = config.get('switch_pin')
buttons.register_buttons([switch_pin], self._button_handler)
self.runout_helper = RunoutHelper(config)
self.get_status = self.runout_helper.get_status
def _button_handler(self, eventtime, state):
self.runout_helper.note_filament_present(state)
def load_config_prefix(config):
return SwitchSensor(config)
|
