jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-14 11:04:29 +05:00
Code Issues Packages Projects Releases Wiki Activity

fix: CRLF to LF

Browse Source
This commit is contained in:
th33xitus
2022-02-26 19:17:22 +01:00
parent 5c45bc7617
commit 4517415e9d
1 changed files with 87 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
resources/gcode_shell_command.py
View File

@@ -1,87 +1,87 @@
# Run a shell command via gcode # Run a shell command via gcode
# #
# Copyright (C) 2019 Eric Callahan <arksine.code@gmail.com> # Copyright (C) 2019 Eric Callahan <arksine.code@gmail.com>
# #
# This file may be distributed under the terms of the GNU GPLv3 license. # This file may be distributed under the terms of the GNU GPLv3 license.
import os import os
import shlex import shlex
import subprocess import subprocess
import logging import logging
class ShellCommand: class ShellCommand:
def __init__(self, config): def __init__(self, config):
self.name = config.get_name().split()[-1] self.name = config.get_name().split()[-1]
self.printer = config.get_printer() self.printer = config.get_printer()
self.gcode = self.printer.lookup_object('gcode') self.gcode = self.printer.lookup_object('gcode')
cmd = config.get('command') cmd = config.get('command')
cmd = os.path.expanduser(cmd) cmd = os.path.expanduser(cmd)
self.command = shlex.split(cmd) self.command = shlex.split(cmd)
self.timeout = config.getfloat('timeout', 2., above=0.) self.timeout = config.getfloat('timeout', 2., above=0.)
self.verbose = config.getboolean('verbose', True) self.verbose = config.getboolean('verbose', True)
self.proc_fd = None self.proc_fd = None
self.partial_output = "" self.partial_output = ""
self.gcode.register_mux_command( self.gcode.register_mux_command(
"RUN_SHELL_COMMAND", "CMD", self.name, "RUN_SHELL_COMMAND", "CMD", self.name,
self.cmd_RUN_SHELL_COMMAND, self.cmd_RUN_SHELL_COMMAND,
desc=self.cmd_RUN_SHELL_COMMAND_help) desc=self.cmd_RUN_SHELL_COMMAND_help)
def _process_output(self, eventime): def _process_output(self, eventime):
if self.proc_fd is None: if self.proc_fd is None:
return return
try: try:
data = os.read(self.proc_fd, 4096) data = os.read(self.proc_fd, 4096)
except Exception: except Exception:
pass pass
data = self.partial_output + data data = self.partial_output + data
if '\n' not in data: if '\n' not in data:
self.partial_output = data self.partial_output = data
return return
elif data[-1] != '\n': elif data[-1] != '\n':
split = data.rfind('\n') + 1 split = data.rfind('\n') + 1
self.partial_output = data[split:] self.partial_output = data[split:]
data = data[:split] data = data[:split]
else: else:
self.partial_output = "" self.partial_output = ""
self.gcode.respond_info(data) self.gcode.respond_info(data)
cmd_RUN_SHELL_COMMAND_help = "Run a linux shell command" cmd_RUN_SHELL_COMMAND_help = "Run a linux shell command"
def cmd_RUN_SHELL_COMMAND(self, params): def cmd_RUN_SHELL_COMMAND(self, params):
gcode_params = params.get('PARAMS','') gcode_params = params.get('PARAMS','')
gcode_params = shlex.split(gcode_params) gcode_params = shlex.split(gcode_params)
reactor = self.printer.get_reactor() reactor = self.printer.get_reactor()
try: try:
proc = subprocess.Popen( proc = subprocess.Popen(
self.command + gcode_params, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) self.command + gcode_params, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
except Exception: except Exception:
logging.exception( logging.exception(
"shell_command: Command {%s} failed" % (self.name)) "shell_command: Command {%s} failed" % (self.name))
raise self.gcode.error("Error running command {%s}" % (self.name)) raise self.gcode.error("Error running command {%s}" % (self.name))
if self.verbose: if self.verbose:
self.proc_fd = proc.stdout.fileno() self.proc_fd = proc.stdout.fileno()
self.gcode.respond_info("Running Command {%s}...:" % (self.name)) self.gcode.respond_info("Running Command {%s}...:" % (self.name))
hdl = reactor.register_fd(self.proc_fd, self._process_output) hdl = reactor.register_fd(self.proc_fd, self._process_output)
eventtime = reactor.monotonic() eventtime = reactor.monotonic()
endtime = eventtime + self.timeout endtime = eventtime + self.timeout
complete = False complete = False
while eventtime < endtime: while eventtime < endtime:
eventtime = reactor.pause(eventtime + .05) eventtime = reactor.pause(eventtime + .05)
if proc.poll() is not None: if proc.poll() is not None:
complete = True complete = True
break break
if not complete: if not complete:
proc.terminate() proc.terminate()
if self.verbose: if self.verbose:
if self.partial_output: if self.partial_output:
self.gcode.respond_info(self.partial_output) self.gcode.respond_info(self.partial_output)
self.partial_output = "" self.partial_output = ""
if complete: if complete:
msg = "Command {%s} finished\n" % (self.name) msg = "Command {%s} finished\n" % (self.name)
else: else:
msg = "Command {%s} timed out" % (self.name) msg = "Command {%s} timed out" % (self.name)
self.gcode.respond_info(msg) self.gcode.respond_info(msg)
reactor.unregister_fd(hdl) reactor.unregister_fd(hdl)
self.proc_fd = None self.proc_fd = None
def load_config_prefix(config): def load_config_prefix(config):
return ShellCommand(config) return ShellCommand(config)