# coding: utf-8 # python 2 only # Copyright (c) 2026 TormachTips.com. All rights reserved. # Licensed under the TormachTips Personal Use License. # Permission is granted only for private personal use and private personal modification. # No sharing, publication, distribution, resale, sublicensing, screenshots, code excerpts, # benchmarks, or videos are permitted without prior written permission. # Requests: tormach.1100m@gmail.com # Information page: https://tormachtips.com/plugins.htm ########################################### ## ## ## Cycle Time Monitor 1.22 ## ## www.tormachtips.com ## ## ## ########################################### # 1.22 - publish mean vs median, just for curiosity - 6/18/2026 # 1.21 - pushes segment estimation to a separate file for tools_on_deck to read, if available - 6/16/2026 # 1.20 - Major refactor. Cycle and segment estimation greatly improved and, with tab plugin, is shown as a tab. - 6/13/2026 # 1.10 - Uses median historical cycle and tool-segment times for outlier-resistant typical runtime display - 6/06/2026 # 1.09 - Combines Cycle Time Estimator & Program Completion, adds Tool Segment Estimation vs Actual Logging - 6/01/2026 # This combined plugin estimates cycle time on file load, tracks actual per-tool segment # elapsed time during program execution, writes a clean estimate-vs-actual pipe log, # and optionally sends email notifications on tool-change prompts and completion. # # This script supercedes the program_completion and cycle_estimator scripts. # They should be disabled and this used instead. # # The email notifier is controlled by notify_mode in: # /home/operator/gcode/python/program_completion_notifier_config.ini # # Ctrl+H cycles: # 0 = Script disabled # 1 = Notify on completion only # 2 = Notify on tool changes and completion # # Legacy enabled is still accepted: # enabled = 0 maps to notify_mode = 0 # enabled = 1 maps to notify_mode = 1 import linuxcnc import threading import time from datetime import datetime import os import math import subprocess import sys import types import ConfigParser import smtplib import gtk import glib import re import sqlite3 import Tkinter as tk from email.mime.text import MIMEText from ui_hooks import plugin import singletons import constants CURRENT_VER = "1.22" SCRIPT_NAME = "Cycle Time Monitor" DESCRIPTION = "Estimates cycle time, logs actual per-tool segment time, and sends optional completion/tool-change notifications." ENABLED = 1 DEV_MACHINE = 1 DEV_MACHINE_FLAG = "/home/operator/gcode/python/dev_machine.txt" NOTIFIER_CONFIG = "/home/operator/gcode/python/cycle_time_config.ini" ARCHIVE_GCODE_LOG = "/home/operator/gcode/python/tormachtips_gcode_log.txt" SEGMENT_LOG = "/home/operator/gcode/python/cycle_time_main_log.txt" TOOL_SEGMENT_TIMING_LOG = "/home/operator/gcode/python/cycle_time_tool_log.txt" CYCLE_RUN_TIMING_LOG = "/home/operator/gcode/python/cycle_time_cycle_log.txt" CYCLE_TIME_TAB_STATUS = "/home/operator/gcode/python/cycle_time_tab_status.txt" CYCLE_TIME_TOOLS_ON_DECK_STATUS = "/home/operator/gcode/python/cycle_time_tools_on_deck_status.txt" CYCLE_TIME_TAB_HISTORY = "/home/operator/gcode/python/cycle_time_tab_history_status.txt" CYCLE_TIME_TAB_SNAPSHOT_DIR = "/home/operator/gcode/python/cycle_time_snapshots" CYCLE_TIME_HISTORY_EDIT_DB = "/home/operator/gcode/python/cycle_time_history_edits.db" CYCLE_TIME_TAB_REFRESH = 1000 MODE_DISABLED = 0 MODE_COMPLETE_ONLY = 1 MODE_TOOL_AND_DONE = 2 DEFAULT_UNITS = "inches" DEFAULT_FEEDRATE = 50 DEFAULT_RAPIDRATE = 100 TOOL_CHANGE_SECS = 15.0 TOOLCHANGE_SECONDS = 5 G30_SECONDS = 5 M30_CHECKER = 1 M30_POPUP_DELAY = 5 WRITE_ANNO_FILE = 0 TOOL_LOG_PROGRAM_WIDTH = 100 TOOL_LOG_RUN_WIDTH = 20 TOOL_LOG_SEGMENT_WIDTH = 10 TOOL_LOG_TOOL_WIDTH = 6 TOOL_LOG_RESULT_WIDTH = 10 TOOL_LOG_TIME_WIDTH = 10 TOOL_LOG_DELTA_WIDTH = 10 TOOL_LOG_PERCENT_WIDTH = 8 LOG_FILENAME_WIDTH = 100 LOG_SEGMENT_WIDTH = 10 LOG_TOOL_WIDTH = 6 LOG_EVENT_WIDTH = 28 CYCLE_TIME_PROGRESS_WIDTH = 60 CYCLE_TIME_PROGRESS_BLOCK = u"\u2588" CYCLE_TIME_PROGRESS_GREEN_OVER = 5.0 CYCLE_TIME_PROGRESS_YELLOW_OVER = 10.0 CYCLE_TIME_PROGRESS_GREEN_MARK = "[TT_PROGRESS_GREEN]" CYCLE_TIME_PROGRESS_YELLOW_MARK = "[TT_PROGRESS_YELLOW]" CYCLE_TIME_PROGRESS_RED_MARK = "[TT_PROGRESS_RED]" CYCLE_TIME_HISTORY_EXCLUDED_MARK = "[TT_HISTORY_EXCLUDED]" CYCLE_TIME_ACTIVE_SEGMENT_MARK = "[TT_ACTIVE_SEGMENT]" DEFAULT_CONFIG_TEXT = """[settings] notify_mode = 1 ; notify_mode: ; 0 = Script disabled ; 1 = Notify on completion only ; 2 = Notify on tool changes and completion ; ; Ctrl+H cycles through these three modes. [email] sender = your_email@example.com recipient = recipient@example.com subject = {event} body = {event}\\n\\nProgram: {filename}\\nElapsed: {elapsed}\\nTool: {tool}\\nPath: {path} smtp_server = smtp.gmail.com smtp_port = 465 smtp_username = your_email@example.com smtp_password = app_password""" class M30PopupProcess(object): def __init__(self, full_path): self.full_path = full_path def _is_nc_file(self): return os.path.splitext(self.full_path)[1].lower() == '.nc' def _get_last_non_whitespace_line(self): last_line = None with open(self.full_path, 'r') as f: for raw in f: stripped = raw.strip() if stripped: last_line = stripped return last_line def _last_line_has_end_code(self): try: last_line = self._get_last_non_whitespace_line() if not last_line: return False return re.search(r'^\s*(?:M30|M2)\s*(?:\(.*\)|;.*)?\s*$', last_line, re.IGNORECASE) is not None except: return True def _append_m30(self): with open(self.full_path, 'a') as f: f.write('\nM30\n') def run(self): if not os.path.isfile(self.full_path): return 0 if self._is_nc_file(): pass else: return 0 if self._last_line_has_end_code(): return 0 root = tk.Tk() root.title("EOF Marker Not Found") root.resizable(False, False) msg = "G-code does not end with M2 or M30.\n\nAdd M30 to end of this program?\n\nTo disable this message, open cycle_time_monitor_plugin.py\nand set M30_CHECKER to 0." tk.Label(root, text=msg, justify='left', padx=20, pady=15).pack() btn_frame = tk.Frame(root, padx=10, pady=10) btn_frame.pack() def do_yes(): try: if not self._last_line_has_end_code(): self._append_m30() finally: root.destroy() def do_no(): root.destroy() tk.Button(btn_frame, text="Yes", width=10, command=do_yes).pack(side='left', padx=5) tk.Button(btn_frame, text="No", width=10, command=do_no).pack(side='left', padx=5) try: root.attributes('-topmost', True) root.lift() except: pass root.mainloop() return 0 class LastRunHistory(object): def __init__(self, archive_log_path, error_handler=None): self.archive_log_path = archive_log_path self.error_handler = error_handler def _write_error(self, msg): if self.error_handler: self.error_handler.write(msg, constants.ALARM_LEVEL_LOW) def format_elapsed_since(self, delta): total_seconds = int(delta.total_seconds()) if total_seconds < 0: total_seconds = 0 days = total_seconds // 86400 hours = (total_seconds % 86400) // 3600 minutes = (total_seconds % 3600) // 60 if days > 0: return "{}d {}h ago".format(days, hours) if hours > 0: return "{}h {}m ago".format(hours, minutes) return "{}m ago".format(minutes) def get_run_history(self, full_path): if not os.path.isfile(self.archive_log_path): return 0, None, None target_candidates = set([ full_path, os.path.relpath(full_path, '/home/operator'), os.path.relpath(full_path, '/home/operator/gcode')]) last_run_dt = None last_run_type = None run_count = 0 try: with open(self.archive_log_path, 'r') as f: for raw in f: line = raw.strip() if not line: continue parts = [p.strip() for p in line.split('|', 3)] if len(parts) != 4: continue dt_txt, machine, logged_file, status = parts logged_file = logged_file.strip() if logged_file not in target_candidates: continue status_l = status.lower() if not status_l.startswith('program stopped'): continue run_count += 1 try: dt_val = datetime.strptime(dt_txt, '%Y-%m-%d %H:%M:%S.%f') except: try: dt_val = datetime.strptime(dt_txt, '%Y-%m-%d %H:%M:%S') except: continue run_type = None if 'complete run' in status_l: run_type = 'complete' elif 'partial run' in status_l: run_type = 'partial' if last_run_dt is None or dt_val > last_run_dt: last_run_dt = dt_val last_run_type = run_type except Exception, e: self._write_error("Archive history read error: " + str(e)) return 0, None, None return run_count, last_run_dt, last_run_type class CycleTimeEstimator(object): def __init__(self, error_handler=None): self.error_handler = error_handler self.tool_segments = [] def _write_error(self, msg): if self.error_handler: self.error_handler.write(msg, constants.ALARM_LEVEL_LOW) def _format_time_comment(self, seconds): seconds = max(1, int(math.ceil(seconds))) if seconds < 60: return '{}s'.format(seconds) m, s = divmod(seconds, 60) return '{}m {}s'.format(m, s) def estimate_runtime(self, gcode_path): self.tool_segments = [] if not os.path.isfile(gcode_path): return 0.0, DEFAULT_UNITS total_estimate, units = self._first_pass_runtime(gcode_path) total_actual, _, self.tool_segments = self._second_pass_annotation(gcode_path, total_estimate, units) return round(total_actual / 60.0, 2), units def _first_pass_runtime(self, gcode_path): units = DEFAULT_UNITS tmp_feed = DEFAULT_FEEDRATE tmp_mode = None tmp_pos = {'X': 0.0, 'Y': 0.0, 'Z': 0.0} total_seconds_estimate = 0.0 with open(gcode_path) as f: for raw in f: l = re.split(r'[;#(]', raw.strip())[0].upper() if not l: continue if 'G20' in l: units = 'inches' elif 'G21' in l: units = 'mm' g4_match = re.search(r'\bG4\b(?:\s+P\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+)))?', l) if g4_match: try: dwell_seconds = float(g4_match.group(1) or 0.0) total_seconds_estimate += max(0.0, dwell_seconds) except: pass continue if 'G0' in l: tmp_mode = 'G0' elif 'G1' in l: tmp_mode = 'G1' fm = re.search(r'F\s*(\d+\.?\d*|\.\d+)', l) if fm: try: tmp_feed = float(fm.group(1)) if units == 'mm': tmp_feed /= 25.4 except: pass if re.match(r'^G[23]\s', l): t = self._process_arc(l, tmp_pos, tmp_feed, units, tmp_mode) if t: total_seconds_estimate += t continue coords = self._extract_coords(l) if coords: t = self._process_linear_move(tmp_pos, coords, tmp_mode, tmp_feed, DEFAULT_RAPIDRATE, units) total_seconds_estimate += t tmp_pos.update(coords) return total_seconds_estimate, units def _extract_tool_change(self, code): m6_tool_match = re.search(r'\bM6\b[^;#(]*?\bT\s*(\d+)\b', code) if m6_tool_match: return int(m6_tool_match.group(1)) tool_m6_match = re.search(r'\bT\s*(\d+)\b[^;#(]*?\bM6\b', code) if tool_m6_match: return int(tool_m6_match.group(1)) return None def _second_pass_annotation(self, gcode_path, total_estimate, units): pos = {'X': 0.0, 'Y': 0.0, 'Z': 0.0} last_feedrate = DEFAULT_FEEDRATE motion_mode = None prev_tool = None total_seconds = 0.0 inject_g30_delay = False annotated_lines = [] output_path = None tool_segments = [] active_tool = None segment_start_seconds = 0.0 if WRITE_ANNO_FILE: filename = os.path.basename(gcode_path) output_path = os.path.join('/home/operator/gcode', filename + '.time.nc') with open(gcode_path) as f: for raw_line in f: code = re.split(r'[;#(]', raw_line.strip())[0].strip().upper() tool_num = self._extract_tool_change(code) if tool_num is not None and tool_num != active_tool: if active_tool is not None or total_seconds > segment_start_seconds: tool_segments.append((active_tool, total_seconds - segment_start_seconds)) active_tool = tool_num segment_start_seconds = total_seconds annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay = \ self._process_line(raw_line, pos, motion_mode, last_feedrate, prev_tool, total_seconds, total_estimate, inject_g30_delay, units) if WRITE_ANNO_FILE: annotated_lines.append(annotated) if active_tool is not None or total_seconds > segment_start_seconds: tool_segments.append((active_tool, total_seconds - segment_start_seconds)) if WRITE_ANNO_FILE: with open(output_path, 'w') as outf: for line in annotated_lines: outf.write(line + '\n') return total_seconds, output_path, tool_segments def _process_line(self, raw_line, pos, motion_mode, last_feedrate, prev_tool, total_seconds, total_estimate, inject_g30_delay, units): original_line = raw_line.rstrip('\r\n') line = original_line.strip() if not line or line.startswith(('#', ';', '(')): return original_line, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay code = re.split(r'[;#(]', line)[0].strip().upper() tool_num = self._extract_tool_change(code) if tool_num is not None: if tool_num != prev_tool: total_seconds += TOOLCHANGE_SECONDS prev_tool = tool_num remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(TOOLCHANGE_SECONDS) + ' | ' + self._format_time_comment(remaining) + ' left (5 sec padding for tool change)' return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay return original_line, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay if 'G20' in code: units = 'inches' elif 'G21' in code: units = 'mm' g4_match = re.search(r'\bG4\b(?:\s+P\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+)))?', code) if g4_match: try: dwell_seconds = max(0.0, float(g4_match.group(1) or 0.0)) except: dwell_seconds = 0.0 total_seconds += dwell_seconds remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(dwell_seconds) + ' | ' + self._format_time_comment(remaining) + ' left (G4 dwell)' return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay if 'G0' in code: motion_mode = 'G0' elif 'G1' in code: motion_mode = 'G1' feed_match = re.search(r'F\s*(\d+\.?\d*|\.\d+)', code) if feed_match: try: last_feedrate = float(feed_match.group(1)) if units == 'mm': last_feedrate /= 25.4 except: pass if re.match(r'^G[23]\s', code): arc_time = self._process_arc(code, pos, last_feedrate, units, motion_mode) if arc_time is not None: total_seconds += arc_time remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(arc_time) + ' | ' + self._format_time_comment(remaining) + ' left' return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay return original_line, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay if 'G30' in code: total_seconds += G30_SECONDS remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(G30_SECONDS) + ' | ' + self._format_time_comment(remaining) + ' left (%d second G30 pre-move)' % G30_SECONDS inject_g30_delay = True return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay z_motion_match = re.search(r'\bG[01]\b[^;#(]*\bZ\s*[+-]?(?:\d+(?:\.\d*)?|\.\d+)', code) if inject_g30_delay and z_motion_match: total_seconds += G30_SECONDS remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(G30_SECONDS) + ' | ' + self._format_time_comment(remaining) + ' left (%d second G30 pad)' % G30_SECONDS inject_g30_delay = False return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay coords = self._extract_coords(code) if coords: move_time = self._process_linear_move(pos, coords, motion_mode, last_feedrate, DEFAULT_RAPIDRATE, units) total_seconds += move_time pos.update(coords) remaining = max(0.0, total_estimate - total_seconds) annotated = original_line + ' ; takes ' + self._format_time_comment(move_time) + ' | ' + self._format_time_comment(remaining) + ' left' return annotated, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay return original_line, total_seconds, pos, motion_mode, last_feedrate, prev_tool, inject_g30_delay def _extract_coords(self, line): pattern = r'([XYZ])\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))' return {a: float(v) for a, v in re.findall(pattern, line)} def _process_arc(self, line, pos, feedrate, units, mode): try: clockwise = 'G2' in line num = r'[+-]?(?:\d+(?:\.\d*)?|\.\d+)' end_coords = {a: float(v) for a, v in re.findall(r'([XYZ])\s*(' + num + ')', line)} ijk = {a: float(v) for a, v in re.findall(r'([IJK])\s*(' + num + ')', line)} p_match = re.search(r'P\s*([+-]?\d+\.?\d*|\.\d+)', line) turns = float(p_match.group(1)) if p_match else None x0, y0 = pos.get('X', 0.0), pos.get('Y', 0.0) x1 = end_coords.get('X', x0) y1 = end_coords.get('Y', y0) i = ijk.get('I', 0.0) j = ijk.get('J', 0.0) cx, cy = x0 + i, y0 + j r = math.hypot(i, j) start_angle = math.atan2(y0 - cy, x0 - cx) end_angle = math.atan2(y1 - cy, x1 - cx) if turns is not None: arc_length = 2 * math.pi * r * turns else: delta = end_angle - start_angle delta = (delta + math.pi) % (2 * math.pi) - math.pi if clockwise and delta > 0: delta -= 2 * math.pi elif not clockwise and delta < 0: delta += 2 * math.pi arc_length = abs(delta) * r z0 = pos.get('Z', 0.0) z1 = end_coords.get('Z', z0) z_delta = z1 - z0 path_length = math.hypot(arc_length, z_delta) if abs(z_delta) > 1e-6 else arc_length if units == 'mm': path_length /= 25.4 rate = feedrate if feedrate > 0 else DEFAULT_FEEDRATE arc_time_sec = (path_length / rate) * 60.0 pos.update(end_coords) return arc_time_sec except Exception, e: self._write_error("Arc parse failed: " + str(e)) return None def _process_linear_move(self, pos, new_coords, mode, feedrate, rapidrate, units): if not mode: return 0.0 dist = math.sqrt(sum((new_coords.get(a, pos[a]) - pos[a]) ** 2 for a in 'XYZ')) if dist < 1e-6: return 0.0 if units == 'mm': dist /= 25.4 rate = rapidrate if mode == 'G0' else (feedrate if feedrate > 0 else DEFAULT_FEEDRATE) return (dist / rate) * 60.0 def format_minutes_and_seconds(total_minutes): total_seconds = int(round(total_minutes * 60)) minutes = total_seconds // 60 seconds = total_seconds % 60 return "%d min %d sec" % (minutes, seconds) class CycleTimeHistoryEdits(object): def __init__(self, db_path): self.db_path = db_path self.cycle_edits = {} self.segment_edits = {} def load(self): # Missing database means no edits exist yet. self.cycle_edits = {} self.segment_edits = {} if os.path.isfile(self.db_path): self._load_existing_database() def _load_existing_database(self): connection = sqlite3.connect(self.db_path) try: cursor = connection.cursor() self._load_cycle_edits(cursor) self._load_segment_edits(cursor) finally: connection.close() def _load_cycle_edits(self, cursor): try: cursor.execute("select run_id, actual_seconds, excluded from cycle_history_edits") for run_id, actual_seconds, excluded in cursor.fetchall(): self.cycle_edits[str(run_id)] = { "actual_seconds": actual_seconds, "excluded": int(excluded or 0)} except Exception: pass def _load_segment_edits(self, cursor): try: cursor.execute("select run_id, segment_number, actual_seconds, excluded from segment_history_edits") for run_id, segment_number, actual_seconds, excluded in cursor.fetchall(): self.segment_edits[(str(run_id), int(segment_number))] = { "actual_seconds": actual_seconds, "excluded": int(excluded or 0)} except Exception: pass def get_cycle_seconds(self, run_id, raw_seconds): edit = self.cycle_edits.get(str(run_id)) if edit is None: return raw_seconds if edit.get("excluded"): return None edited_seconds = edit.get("actual_seconds") if edited_seconds is not None: return edited_seconds return raw_seconds def get_segment_seconds(self, run_id, segment_number, raw_seconds): cycle_edit = self.cycle_edits.get(str(run_id)) if cycle_edit is not None and cycle_edit.get("excluded"): return None edit = self.segment_edits.get((str(run_id), int(segment_number))) if edit is None: return raw_seconds if edit.get("excluded"): return None edited_seconds = edit.get("actual_seconds") if edited_seconds is not None: return edited_seconds return raw_seconds class UserPlugin(plugin): def __init__(self): plugin.__init__(self, SCRIPT_NAME) dev_machine_found = os.path.exists(DEV_MACHINE_FLAG) if dev_machine_found: plugin_enabled = DEV_MACHINE else: plugin_enabled = ENABLED if plugin_enabled: self.ui = None self.config_file = NOTIFIER_CONFIG self.active_program_path = None self.active_start_time = None self.original_interp_task_status_change = None self.original_set_message_line_text = None self.original_load_gcode_file = None self.last_tool_change_key = "" self.last_tool_change_time = 0.0 self.active_segment_tool = "" self.active_segment_start_time = None self.active_segment_number = 0 self.active_run_id = "" self.active_total_estimate_seconds = None self.loaded_estimate_rows = [] self.loaded_total_estimate_seconds = None self.loaded_estimate_program_path = "" self.loaded_estimate_program_signature = (0, 0) self.segment_estimates = {} self._history_cache_signature = None self._history_cache = None self._history_edit_db_signature = None self.cycle_time_tab_actual_segments = {} self.cycle_time_tab_cycle_actual_seconds = None self.cycle_time_tab_cycle_result = "" self.cycle_time_tab_run_id = "" self.cycle_time_tab_waiting_for_tool_change = False self.cycle_time_tab_waiting_tool_text = "" self.cycle_time_tab_waiting_since = None self.cycle_time_tab_refresh_timer = None self.cycle_time = CycleTimeEstimator() self.last_run = LastRunHistory(ARCHIVE_GCODE_LOG) self._m30_prompt_lock = threading.Lock() self._m30_prompt_active = {} self.ensure_default_config() t = threading.Thread(target=self.setup_hook) t.daemon = True t.start() glib.timeout_add(3000, self.setup_key_handler) self.cycle_time_tab_refresh_timer = glib.timeout_add(CYCLE_TIME_TAB_REFRESH, self.refresh_cycle_time_tab_snapshot) self.write_status("loaded (Ctrl+H cycles notification mode)") self.write_status("current mode: " + self.get_mode_text(self.get_notify_mode())) return if dev_machine_found: self.write_status("dev machine found. Plugin loaded, but disabled by DEV_MACHINE.") else: self.write_status("loaded, but disabled.") self.write_status("To enable, open script, find ENABLED = 0 and change to ENABLED = 1") def write_status(self, msg, level=constants.ALARM_LEVEL_QUIET): try: self.error_handler.write("[" + SCRIPT_NAME + "] " + msg, level) except Exception: try: ui = getattr(singletons, "g_Machine", None) err_handler = getattr(ui, "error_handler", None) if err_handler: err_handler.write("[" + SCRIPT_NAME + "] " + msg, level) except Exception: pass def ensure_default_config(self): try: if os.path.exists(self.config_file): self.upgrade_config_if_needed() return config_dir = os.path.dirname(self.config_file) if config_dir and not os.path.isdir(config_dir): os.makedirs(config_dir) handle = open(self.config_file, "wb") try: handle.write(DEFAULT_CONFIG_TEXT) finally: handle.close() self.write_status("created default config: " + self.config_file) except Exception as e: self.write_status("failed to create default config: " + str(e), constants.ALARM_LEVEL_QUIET) def upgrade_config_if_needed(self): try: config = ConfigParser.ConfigParser() config.read(self.config_file) changed = False if not config.has_section("settings"): config.add_section("settings") changed = True if not config.has_option("settings", "notify_mode"): if config.has_option("settings", "enabled"): try: enabled = config.getint("settings", "enabled") except Exception: enabled = 1 if enabled: config.set("settings", "notify_mode", str(MODE_COMPLETE_ONLY)) else: config.set("settings", "notify_mode", str(MODE_DISABLED)) else: config.set("settings", "notify_mode", str(MODE_COMPLETE_ONLY)) changed = True if not config.has_section("email"): config.add_section("email") changed = True defaults = ConfigParser.ConfigParser() defaults.readfp(FakeConfigFile(DEFAULT_CONFIG_TEXT)) for option in defaults.options("email"): if not config.has_option("email", option): config.set("email", option, defaults.get("email", option)) changed = True if changed: with open(self.config_file, "w") as f: config.write(f) self.write_status("updated config format: " + self.config_file) except Exception as e: self.write_status("config upgrade skipped: " + str(e), constants.ALARM_LEVEL_QUIET) def _load_config(self): if not os.path.exists(self.config_file): raise IOError("Config file not found") config = ConfigParser.ConfigParser() config.read(self.config_file) return config def get_notify_mode(self): try: config = self._load_config() if config.has_option("settings", "notify_mode"): mode = config.getint("settings", "notify_mode") if mode < MODE_DISABLED: return MODE_DISABLED if mode > MODE_TOOL_AND_DONE: return MODE_TOOL_AND_DONE return mode if config.has_option("settings", "enabled"): enabled = config.getint("settings", "enabled") if enabled: return MODE_COMPLETE_ONLY return MODE_DISABLED except Exception: pass return MODE_DISABLED def set_notify_mode(self, mode): config = self._load_config() if not config.has_section("settings"): config.add_section("settings") config.set("settings", "notify_mode", str(mode)) if config.has_option("settings", "enabled"): config.set("settings", "enabled", "1" if mode else "0") with open(self.config_file, "w") as f: config.write(f) def get_mode_text(self, mode): if mode == MODE_DISABLED: return "Email notifications disabled. Segment timing remains active." if mode == MODE_COMPLETE_ONLY: return "Email on completion only. Segment timing remains active." if mode == MODE_TOOL_AND_DONE: return "Email on tool changes and completion. Segment timing remains active." return "Unknown mode" def toggle_enabled(self): self.cycle_notify_mode() def cycle_notify_mode(self): try: current_mode = self.get_notify_mode() if current_mode == MODE_DISABLED: new_mode = MODE_COMPLETE_ONLY elif current_mode == MODE_COMPLETE_ONLY: new_mode = MODE_TOOL_AND_DONE else: new_mode = MODE_DISABLED self.set_notify_mode(new_mode) self.write_status("Mode: " + self.get_mode_text(new_mode), constants.ALARM_LEVEL_QUIET) except Exception as e: self.write_status("toggle error: " + str(e), constants.ALARM_LEVEL_QUIET) def notify_completion_enabled(self): return self.get_notify_mode() in (MODE_COMPLETE_ONLY, MODE_TOOL_AND_DONE) def notify_tool_change_enabled(self): return self.get_notify_mode() == MODE_TOOL_AND_DONE def get_current_program_path(self, ui): try: if hasattr(ui, "get_current_gcode_path"): path = ui.get_current_gcode_path() if path: return path if hasattr(ui, "current_gcode_file_path"): path = ui.current_gcode_file_path if path: return path if hasattr(ui, "last_gcode_program_path"): path = ui.last_gcode_program_path if path: return path except Exception: pass return "" def get_first_program_tool(self, program_path): if not program_path or not os.path.isfile(program_path): return "" try: with open(program_path, "r") as f: for raw_line in f: code = re.split(r"[;#(]", raw_line.strip())[0].strip().upper() if not code: continue match = re.search(r"\bM0?6\b[^;#(]*?\bT\s*(\d+)\b", code) if match is None: match = re.search(r"\bT\s*(\d+)\b[^;#(]*?\bM0?6\b", code) if match: return "T" + match.group(1) except Exception as e: self.write_status("first tool read error: " + str(e), constants.ALARM_LEVEL_LOW) return "" def get_loaded_tool(self, ui): try: tool_number = int(getattr(ui.status, "tool_in_spindle", 0)) if tool_number > 0: return "T%d" % tool_number except Exception: pass return "Tool unknown" def update_tool_change_wait_state(self): if not self.cycle_time_tab_waiting_for_tool_change: return try: if self.cycle_time_tab_waiting_since is not None: if (time.time() - self.cycle_time_tab_waiting_since) < 2.0: return ui = self.ui or getattr(singletons, "g_Machine", None) if not ui: return loaded_tool = self.get_loaded_tool(ui) if ( self.cycle_time_tab_waiting_tool_text and loaded_tool == self.cycle_time_tab_waiting_tool_text ): self.cycle_time_tab_waiting_for_tool_change = False self.cycle_time_tab_waiting_tool_text = "" self.cycle_time_tab_waiting_since = None self.publish_cycle_time_tab_snapshot() except Exception as e: self.write_status("tool-change wait-state update error: " + str(e), constants.ALARM_LEVEL_QUIET) def get_initial_segment_tool(self, ui, program_path): tool_text = self.get_first_program_tool(program_path) if tool_text: return tool_text return self.get_loaded_tool(ui) def get_program_display_name(self, program_path): if program_path: return os.path.basename(program_path) return "Unknown program" def write_file_loaded_log(self, program_path): old_program_path = self.active_program_path self.active_program_path = program_path try: self.write_segment_log("File loaded", 0, "", "", program_path) finally: self.active_program_path = old_program_path def ensure_segment_log_header(self): try: if os.path.exists(SEGMENT_LOG) and os.path.getsize(SEGMENT_LOG) > 0: return log_dir = os.path.dirname(SEGMENT_LOG) if log_dir and os.path.isdir(log_dir) == False: os.makedirs(log_dir) with open(SEGMENT_LOG, "a") as f: f.write("%-19s | %-*s | %-*s | %-*s | %-*s | %-8s | %s\n" % ( "Timestamp", LOG_FILENAME_WIDTH, "Program", LOG_EVENT_WIDTH, "Event", LOG_SEGMENT_WIDTH, "Segment", LOG_TOOL_WIDTH, "Tool", "Elapsed", "Detail")) f.write("%s | %s | %s | %s | %s | %s | %s\n" % ( "-" * 19, "-" * LOG_FILENAME_WIDTH, "-" * LOG_EVENT_WIDTH, "-" * LOG_SEGMENT_WIDTH, "-" * LOG_TOOL_WIDTH, "-" * 8, "-" * 40)) except Exception as e: self.write_status("segment log header error: " + str(e), constants.ALARM_LEVEL_LOW) def write_segment_log(self, event_text, segment_number=0, tool_text="", elapsed_text="", detail_text=""): try: timestamp = time.strftime("%Y-%m-%d %H:%M:%S") program_path = self.active_program_path if not program_path and self.ui: program_path = self.get_current_program_path(self.ui) filename = self.get_program_display_name(program_path) if segment_number > 0: segment_text = "Segment %d" % segment_number else: segment_text = "" self.ensure_segment_log_header() with open(SEGMENT_LOG, "a") as f: f.write("%s | %-*s | %-*.*s | %-*.*s | %-*.*s | %-8s | %s\n" % ( timestamp, LOG_FILENAME_WIDTH, filename, LOG_EVENT_WIDTH, LOG_EVENT_WIDTH, event_text, LOG_SEGMENT_WIDTH, LOG_SEGMENT_WIDTH, segment_text, LOG_TOOL_WIDTH, LOG_TOOL_WIDTH, tool_text, elapsed_text, detail_text)) except Exception as e: self.write_status("segment log write error: " + str(e), constants.ALARM_LEVEL_LOW) def clean_log_field(self, value): # Keep fixed-width pipe logs one-line and visually aligned. if value is None: return "" text = str(value) text = text.replace("|", "/") text = text.replace("\t", " ") text = text.replace("\r", " ") text = text.replace("\n", " ") return text.strip() def make_run_id(self, start_time): # Human-readable enough for troubleshooting, unique enough for repeated runs. millis = int((start_time - int(start_time)) * 1000) return time.strftime("%Y%m%d-%H%M%S", time.localtime(start_time)) + "-%03d" % millis def get_program_signature(self, program_path): # Used to make sure the in-memory estimate still belongs to the exact loaded file. try: if program_path and os.path.isfile(program_path): return int(os.path.getmtime(program_path)), int(os.path.getsize(program_path)) except Exception as e: self.write_status("program signature error: " + str(e), constants.ALARM_LEVEL_LOW) return 0, 0 def format_signed_seconds(self, seconds): try: seconds = int(round(seconds)) except Exception: seconds = 0 if seconds > 0: return "+" + self.format_elapsed_seconds(seconds) if seconds < 0: return "-" + self.format_elapsed_seconds(abs(seconds)) return "0s" def parse_elapsed_text_to_seconds(self, text): try: text = str(text).strip().lower() if not text: return None total = 0 matched = False for value, unit in re.findall(r'(\d+)\s*([dhms])', text): matched = True value = int(value) if unit == "d": total += value * 86400 elif unit == "h": total += value * 3600 elif unit == "m": total += value * 60 elif unit == "s": total += value if matched: return total if text.endswith("sec"): return int(float(text.replace("sec", "").strip())) return int(float(text)) except Exception: return None def split_pipe_log_line(self, line): try: parts = [p.strip() for p in line.rstrip("\r\n").split("|")] if len(parts) < 2: return [] if parts[0] == "Timestamp" or parts[0].startswith("-"): return [] return parts except Exception: return [] def median_seconds(self, values): try: clean_values = sorted([float(v) for v in values]) count = len(clean_values) if count == 0: return None middle = count // 2 if count % 2: return clean_values[middle] return (clean_values[middle - 1] + clean_values[middle]) / 2.0 except Exception: return None def mean_seconds(self, values): try: clean_values = [float(v) for v in values] count = len(clean_values) if count == 0: return None return sum(clean_values) / float(count) except Exception: return None def get_expected_seconds(self, estimate_seconds, history_seconds, sample_count): expected_seconds, source_text = self.get_expected_detail(estimate_seconds, history_seconds, sample_count) return expected_seconds def get_expected_detail(self, estimate_seconds, history_seconds, sample_count): # Returns both expected time and the display explanation used beside Updated. if history_seconds is None: return estimate_seconds, "Method: calculated estimate" if estimate_seconds is None: return history_seconds, "Method: historical median" try: estimate_seconds = float(estimate_seconds) history_seconds = float(history_seconds) sample_count = int(sample_count or 0) except Exception: return history_seconds, "Method: historical median" if sample_count >= 3: return history_seconds, "Method: historical median" if sample_count == 2: if history_seconds > 0: ratio = estimate_seconds / history_seconds if ratio >= 2.0 or ratio <= 0.50: return history_seconds, "Method: guarded median - estimate disagrees" return (estimate_seconds * 0.40) + (history_seconds * 0.60), "Method: 40/60 estimate/median blend" if sample_count == 1: if history_seconds > 0: ratio = estimate_seconds / history_seconds if ratio >= 3.0 or ratio <= 0.33: return history_seconds, "Method: guarded median - estimate disagrees" return (estimate_seconds * 0.70) + (history_seconds * 0.30), "Method: 70/30 estimate/median blend" return estimate_seconds, "Method: calculated estimate" def format_cycle_method_line(self, source_text, history): # Shows mean beside median so outlier influence is visible without changing expected-time logic. method_text = source_text.replace("Method: ", "") stat_parts = [] mean_seconds = history.get("cycle_mean_seconds") median_seconds = history.get("cycle_avg_seconds") if mean_seconds is not None: stat_parts.append("[unused mean: %s]" % self.format_elapsed_seconds(mean_seconds)) if median_seconds is not None: stat_parts.append("[median: %s]" % self.format_elapsed_seconds(median_seconds)) if len(stat_parts) > 0: return "Method: %s %s" % (method_text, " ".join(stat_parts)) return "Method: %s" % method_text def get_cycle_history_detail_lines(self, history): # Shows completed cycle times that feed the median, plus excluded runs for review. detail_lines = [] cycle_rows = history.get("cycle_history_seconds", []) run_count = history.get("run_count", 0) run_word = "run" if run_count == 1 else "runs" detail_lines.append("Median built from %s %s" % (run_count, run_word)) detail_lines.append("") included_rows = [] for row in cycle_rows: excluded = False if len(row) > 3: excluded = row[3] if excluded == False: included_rows.append(row) if len(included_rows) == 0: summary_label_width = len("Expected") detail_lines.append("%*s: none" % (summary_label_width, "Best")) detail_lines.append("%*s: none" % (summary_label_width, "Worst")) return detail_lines cycle_values = [row[1] for row in included_rows] best_seconds = min(cycle_values) worst_seconds = max(cycle_values) best_text = self.format_elapsed_seconds(best_seconds) worst_text = self.format_elapsed_seconds(worst_seconds) spread_text = self.format_elapsed_seconds(worst_seconds - best_seconds) expected_text = "" expected_seconds = history.get("cycle_expected_seconds") if expected_seconds is None: expected_seconds = history.get("cycle_avg_seconds") if expected_seconds is not None: expected_text = self.format_elapsed_seconds(expected_seconds) summary_label_width = len("Expected") summary_value_width = max(len(best_text), len(worst_text), len(spread_text), len(expected_text)) detail_lines.append("%*s: %*s" % (summary_label_width, "Best", summary_value_width, best_text)) detail_lines.append("%*s: %*s" % (summary_label_width, "Worst", summary_value_width, worst_text)) detail_lines.append("%*s: %*s" % (summary_label_width, "Spread", summary_value_width, spread_text)) if expected_text: detail_lines.append("%*s: %*s" % (summary_label_width, "Expected", summary_value_width, expected_text)) detail_lines.append("") sorted_rows = sorted(cycle_rows, key=lambda row: row[0], reverse=True) run_value_width = 0 for row in sorted_rows: elapsed_text = self.format_elapsed_seconds(row[1]) if len(elapsed_text) > run_value_width: run_value_width = len(elapsed_text) for row in sorted_rows: timestamp_text = row[0] elapsed_seconds = row[1] edited = False excluded = False if len(row) > 2: edited = row[2] if len(row) > 3: excluded = row[3] edit_marker = " " if edited: edit_marker = "*" line_text = "%s%s: %*s" % ( edit_marker, self.format_history_timestamp(timestamp_text), run_value_width, self.format_elapsed_seconds(elapsed_seconds)) if excluded: line_text = CYCLE_TIME_HISTORY_EXCLUDED_MARK + line_text detail_lines.append(line_text) return detail_lines def get_file_signature(self, path): try: if os.path.isfile(path): return (os.path.getmtime(path), os.path.getsize(path)) except Exception: pass return (None, None) def copy_history_cache(self, history): return { "run_count": history.get("run_count", 0), "cycle_avg_seconds": history.get("cycle_avg_seconds"), "cycle_mean_seconds": history.get("cycle_mean_seconds"), "cycle_expected_seconds": history.get("cycle_expected_seconds"), "cycle_history_seconds": list(history.get("cycle_history_seconds", [])), "segment_avg_seconds": dict(history.get("segment_avg_seconds", {})), "segment_sample_counts": dict(history.get("segment_sample_counts", {}))} def get_history_averages_for_program(self, program_path): history = { "run_count": 0, "cycle_avg_seconds": None, "cycle_mean_seconds": None, "cycle_expected_seconds": None, "cycle_history_seconds": [], "segment_avg_seconds": {}, "segment_sample_counts": {}} try: program_name = self.get_program_display_name(program_path) cache_signature = ( program_name, self.get_file_signature(CYCLE_RUN_TIMING_LOG), self.get_file_signature(TOOL_SEGMENT_TIMING_LOG), self.get_file_signature(CYCLE_TIME_HISTORY_EDIT_DB)) if self._history_cache_signature == cache_signature and self._history_cache is not None: return self.copy_history_cache(self._history_cache) history_edits = CycleTimeHistoryEdits(CYCLE_TIME_HISTORY_EDIT_DB) history_edits.load() cycle_values = [] cycle_history_rows = [] if os.path.isfile(CYCLE_RUN_TIMING_LOG): with open(CYCLE_RUN_TIMING_LOG, "r") as f: for raw in f: parts = self.split_pipe_log_line(raw) if len(parts) < 7: continue logged_program = parts[1] actual_text = parts[3] result_text = parts[6].strip().lower().replace("[", "").replace("]", "") if logged_program != program_name: continue if result_text != "completed": continue raw_seconds = self.parse_elapsed_text_to_seconds(actual_text) run_id = "" if len(parts) > 7: run_id = parts[7] display_seconds = raw_seconds edited = False excluded = False edit = history_edits.cycle_edits.get(str(run_id)) if edit is not None: if edit.get("excluded"): excluded = True edited_seconds = edit.get("actual_seconds") if edited_seconds is not None: display_seconds = edited_seconds edited = True if excluded == False: cycle_values.append(display_seconds) cycle_history_rows.append((parts[0], display_seconds, edited, excluded)) if len(cycle_values) > 0: history["run_count"] = len(cycle_values) history["cycle_avg_seconds"] = self.median_seconds(cycle_values) history["cycle_mean_seconds"] = self.mean_seconds(cycle_values) history["cycle_history_seconds"] = cycle_history_rows segment_values = {} if os.path.isfile(TOOL_SEGMENT_TIMING_LOG): with open(TOOL_SEGMENT_TIMING_LOG, "r") as f: for raw in f: parts = self.split_pipe_log_line(raw) if len(parts) < 9: continue logged_program = parts[1] segment_text = parts[2] actual_text = parts[5] result_text = parts[8].strip().lower().replace("[", "").replace("]", "") if logged_program != program_name: continue if result_text != "completed": continue match = re.search(r'(\d+)', segment_text) if not match: continue segment_number = int(match.group(1)) raw_seconds = self.parse_elapsed_text_to_seconds(actual_text) run_id = "" if len(parts) > 9: run_id = parts[9] seconds = history_edits.get_segment_seconds(run_id, segment_number, raw_seconds) if seconds is None: continue if segment_number not in segment_values: segment_values[segment_number] = [] segment_values[segment_number].append(seconds) for segment_number in segment_values: values = segment_values[segment_number] if len(values) > 0: history["segment_avg_seconds"][segment_number] = self.median_seconds(values) history["segment_sample_counts"][segment_number] = len(values) self._history_cache_signature = cache_signature self._history_cache = self.copy_history_cache(history) except Exception as e: self.write_status("history average read error: " + str(e), constants.ALARM_LEVEL_LOW) return self.copy_history_cache(history) def apply_history_avg_to_row(self, row, expected_seconds): # Kept method name for small-change compatibility; the displayed value is now Expected. try: if row is not None and expected_seconds is not None: row["history_avg"] = self.format_elapsed_seconds(expected_seconds) except Exception: pass return row def get_segment_expected_seconds(self, segment_number, estimated_seconds, history): # Segment progress should use the same expected value as the displayed segment row. segment_averages = history.get("segment_avg_seconds", {}) segment_counts = history.get("segment_sample_counts", {}) return self.get_expected_seconds( estimated_seconds, segment_averages.get(segment_number), segment_counts.get(segment_number, 0)) def make_segment_timing_row_with_history(self, segment_number, segment_name, estimated_seconds, actual_seconds, result_text, running, history): # Each segment earns trust independently from the full-cycle median. segment_averages = history.get("segment_avg_seconds", {}) segment_counts = history.get("segment_sample_counts", {}) history_seconds = segment_averages.get(segment_number) sample_count = segment_counts.get(segment_number, 0) expected_seconds = self.get_expected_seconds(estimated_seconds, history_seconds, sample_count) row = self.make_timing_row( segment_name, expected_seconds, actual_seconds, result_text, running) row = self.apply_history_avg_to_row(row, expected_seconds) return row def get_finished_progress_color_mark(self, estimated_seconds, actual_seconds): # The status file stays plain text. The tab reader removes this marker and applies GTK color. try: estimated_seconds = float(estimated_seconds) actual_seconds = float(actual_seconds) except Exception: return "" if estimated_seconds <= 0.0: return "" over_percent = ((actual_seconds - estimated_seconds) / estimated_seconds) * 100.0 if over_percent <= CYCLE_TIME_PROGRESS_GREEN_OVER: return CYCLE_TIME_PROGRESS_GREEN_MARK if over_percent <= CYCLE_TIME_PROGRESS_YELLOW_OVER: return CYCLE_TIME_PROGRESS_YELLOW_MARK return CYCLE_TIME_PROGRESS_RED_MARK def make_cycle_time_progress_bar(self, label_text, estimated_seconds, actual_seconds, finished=False): # Shared text progress bar for total cycle and active tool-segment progress. # Finished cycles are pinned to 100% because actual runtime can beat expected. if estimated_seconds is None: return "" try: estimated_seconds = float(estimated_seconds) except Exception: return "" if estimated_seconds <= 0.0: return "" if finished: percent_done = 100.0 else: if actual_seconds is None: actual_seconds = 0.0 try: actual_seconds = float(actual_seconds) except Exception: actual_seconds = 0.0 if actual_seconds < 0.0: actual_seconds = 0.0 percent_done = (actual_seconds / estimated_seconds) * 100.0 if percent_done > 100.0: percent_done = 100.0 if percent_done < 0.0: percent_done = 0.0 blocks = int(round((percent_done / 100.0) * CYCLE_TIME_PROGRESS_WIDTH)) if blocks > CYCLE_TIME_PROGRESS_WIDTH: blocks = CYCLE_TIME_PROGRESS_WIDTH if blocks < 0: blocks = 0 progress_text = u"%s |%s%s| %5.1f%%" % (label_text, CYCLE_TIME_PROGRESS_BLOCK * blocks, u" " * (CYCLE_TIME_PROGRESS_WIDTH - blocks), percent_done) if percent_done >= 100.0: color_mark = self.get_finished_progress_color_mark(estimated_seconds, actual_seconds) if color_mark: progress_text = progress_text + " " + color_mark return progress_text def format_under_over_percent_label(self, estimated_seconds, actual_seconds): # Appends compact percent detail to completed rows. # Uses raw seconds, so the percent can show small differences hidden by rounded display seconds. try: estimated_seconds = float(estimated_seconds) actual_seconds = float(actual_seconds) except Exception: return "" if estimated_seconds <= 0.0: return "" delta_percent = ((actual_seconds - estimated_seconds) / estimated_seconds) * 100.0 rounded_percent = int(round(abs(delta_percent))) if delta_percent < 0: return "%d%% under" % rounded_percent if delta_percent > 0: return "%d%% over" % rounded_percent return "0%% on" def make_timing_row(self, label_text, estimated_seconds, actual_seconds=None, result_text="", running=False): if estimated_seconds is None: estimated_text = "unknown" else: estimated_text = self.format_elapsed_seconds(estimated_seconds) # percent/speed fields are retained for possible future display columns. history_avg_text = "" if running: if actual_seconds is None: actual_text = "running" amount_text = "unknown" timing_text = "remaining" else: actual_text = self.format_elapsed_seconds(actual_seconds) amount_text, timing_text = self.get_live_remaining_label(estimated_seconds, actual_seconds) return { "label": label_text, "actual": actual_text, "estimated": estimated_text, "amount": amount_text, "timing": timing_text, "percent": "", "speed": "", "result": "%s" % result_text, "history_avg": history_avg_text,} if actual_seconds is None: return { "label": label_text, "actual": "pending", "estimated": estimated_text, "amount": "", "timing": "estimated", "percent": "", "speed": "", "result": "pending", "history_avg": history_avg_text,} actual_text = self.format_elapsed_seconds(actual_seconds) if estimated_seconds is None: return { "label": label_text, "actual": actual_text, "estimated": "unknown", "amount": "", "timing": "actual", "percent": "", "speed": "", "result": "%s" % result_text, "history_avg": history_avg_text} delta_seconds = actual_seconds - estimated_seconds display_delta_seconds = int(round(delta_seconds)) abs_delta_text = self.format_elapsed_seconds(abs(display_delta_seconds)) percent_text = self.format_delta_percent(estimated_seconds, actual_seconds).replace("+", "") if display_delta_seconds < 0: timing_text = "under estimate" speed_text = "faster than expected" elif display_delta_seconds > 0: timing_text = "over estimate" speed_text = "slower than expected" else: abs_delta_text = "" timing_text = "bullseye" speed_text = "as expected" percent_label = self.format_under_over_percent_label(estimated_seconds, actual_seconds) display_result_text = "%s" % result_text if percent_label: display_result_text = "%s [%s]" % (display_result_text, percent_label) return { "label": label_text, "actual": actual_text, "estimated": estimated_text, "amount": abs_delta_text, "timing": timing_text, "percent": percent_text, "speed": speed_text, "result": display_result_text, "history_avg": history_avg_text,} def format_timing_rows(self, rows): if len(rows) == 0: return [] timing_aliases = { "estimated": "est", "remaining": "rem", "under estimate": "under est", "over estimate": "over est", "on estimate": "on est", "actual": "act"} for row in rows: if row["timing"] in timing_aliases: row["timing"] = timing_aliases[row["timing"]] label_width = max([len(row["label"]) for row in rows]) actual_width = max([len(row["actual"]) for row in rows]) estimate_width = max([len(row["estimated"]) for row in rows]) amount_width = max([len(row["amount"]) for row in rows] + [1]) timing_width = max([len(row["timing"]) for row in rows] + [1]) result_width = max([len(row["result"]) for row in rows]) output = [] pipe_pad = " | " for row in rows: line_text = "%-*s %*s of %*s%s%*s %-*s%s%-*s" % ( label_width, row["label"], actual_width, row["actual"], estimate_width, row["estimated"], pipe_pad, amount_width, row["amount"], timing_width, row["timing"], pipe_pad, result_width, row["result"]) if row.get("active_segment"): line_text = CYCLE_TIME_ACTIVE_SEGMENT_MARK + line_text output.append(line_text) return output def format_delta_percent(self, estimated_seconds, actual_seconds): if estimated_seconds is not None and estimated_seconds > 0: delta_seconds = actual_seconds - estimated_seconds return "%+.1f%%" % ((delta_seconds / estimated_seconds) * 100.0) return "" def ensure_cycle_run_timing_log_header(self): try: if os.path.exists(CYCLE_RUN_TIMING_LOG) and os.path.getsize(CYCLE_RUN_TIMING_LOG) > 0: return log_dir = os.path.dirname(CYCLE_RUN_TIMING_LOG) if log_dir and os.path.isdir(log_dir) == False: os.makedirs(log_dir) with open(CYCLE_RUN_TIMING_LOG, "a") as f: f.write("%-19s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %s\n" % ( "Timestamp", TOOL_LOG_PROGRAM_WIDTH, "Program", TOOL_LOG_TIME_WIDTH, "Estimated", TOOL_LOG_TIME_WIDTH, "Actual", TOOL_LOG_DELTA_WIDTH, "Delta", TOOL_LOG_PERCENT_WIDTH, "Delta %", TOOL_LOG_RESULT_WIDTH, "Result", TOOL_LOG_RUN_WIDTH, "Run ID", "Detail")) f.write("%s | %s | %s | %s | %s | %s | %s | %s | %s\n" % ( "-" * 19, "-" * TOOL_LOG_PROGRAM_WIDTH, "-" * TOOL_LOG_TIME_WIDTH, "-" * TOOL_LOG_TIME_WIDTH, "-" * TOOL_LOG_DELTA_WIDTH, "-" * TOOL_LOG_PERCENT_WIDTH, "-" * TOOL_LOG_RESULT_WIDTH, "-" * TOOL_LOG_RUN_WIDTH, "-" * 40)) except Exception as e: self.write_status("cycle run timing header error: " + str(e), constants.ALARM_LEVEL_LOW) def write_cycle_run_timing_log(self, actual_seconds, result_text, detail_text): try: program_path = self.active_program_path if not program_path and self.ui: program_path = self.get_current_program_path(self.ui) filename = self.get_program_display_name(program_path) estimated_seconds = self.active_total_estimate_seconds estimated_text = "" delta_text = "" delta_percent_text = "" if estimated_seconds is not None: estimated_text = self.format_elapsed_seconds(estimated_seconds) delta_text = self.format_signed_seconds(actual_seconds - estimated_seconds) delta_percent_text = self.format_delta_percent(estimated_seconds, actual_seconds) timestamp = time.strftime("%Y-%m-%d %H:%M:%S") self.ensure_cycle_run_timing_log_header() with open(CYCLE_RUN_TIMING_LOG, "a") as f: f.write("%s | %-*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %s\n" % ( timestamp, TOOL_LOG_PROGRAM_WIDTH, self.clean_log_field(filename), TOOL_LOG_TIME_WIDTH, TOOL_LOG_TIME_WIDTH, self.clean_log_field(estimated_text), TOOL_LOG_TIME_WIDTH, TOOL_LOG_TIME_WIDTH, self.clean_log_field(self.format_elapsed_seconds(actual_seconds)), TOOL_LOG_DELTA_WIDTH, TOOL_LOG_DELTA_WIDTH, self.clean_log_field(delta_text), TOOL_LOG_PERCENT_WIDTH, TOOL_LOG_PERCENT_WIDTH, self.clean_log_field(delta_percent_text), TOOL_LOG_RESULT_WIDTH, TOOL_LOG_RESULT_WIDTH, self.clean_log_field(result_text), TOOL_LOG_RUN_WIDTH, TOOL_LOG_RUN_WIDTH, self.clean_log_field(self.active_run_id), self.clean_log_field(detail_text))) except Exception as e: self.write_status("cycle run timing log write error: " + str(e), constants.ALARM_LEVEL_LOW) def ensure_tool_segment_timing_log_header(self): try: if os.path.exists(TOOL_SEGMENT_TIMING_LOG) and os.path.getsize(TOOL_SEGMENT_TIMING_LOG) > 0: return log_dir = os.path.dirname(TOOL_SEGMENT_TIMING_LOG) if log_dir and os.path.isdir(log_dir) == False: os.makedirs(log_dir) with open(TOOL_SEGMENT_TIMING_LOG, "a") as f: f.write("%-19s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %-*s | %s\n" % ( "Timestamp", TOOL_LOG_PROGRAM_WIDTH, "Program", TOOL_LOG_SEGMENT_WIDTH, "Segment", TOOL_LOG_TOOL_WIDTH, "Tool", TOOL_LOG_TIME_WIDTH, "Estimated", TOOL_LOG_TIME_WIDTH, "Actual", TOOL_LOG_DELTA_WIDTH, "Delta", TOOL_LOG_PERCENT_WIDTH, "Delta %", TOOL_LOG_RESULT_WIDTH, "Result", TOOL_LOG_RUN_WIDTH, "Run ID", "Detail")) f.write("%s | %s | %s | %s | %s | %s | %s | %s | %s | %s | %s\n" % ( "-" * 19, "-" * TOOL_LOG_PROGRAM_WIDTH, "-" * TOOL_LOG_SEGMENT_WIDTH, "-" * TOOL_LOG_TOOL_WIDTH, "-" * TOOL_LOG_TIME_WIDTH, "-" * TOOL_LOG_TIME_WIDTH, "-" * TOOL_LOG_DELTA_WIDTH, "-" * TOOL_LOG_PERCENT_WIDTH, "-" * TOOL_LOG_RESULT_WIDTH, "-" * TOOL_LOG_RUN_WIDTH, "-" * 40)) except Exception as e: self.write_status("tool segment timing header error: " + str(e), constants.ALARM_LEVEL_LOW) def write_tool_segment_timing_log(self, segment_number, tool_text, actual_seconds, result_text, detail_text): try: program_path = self.active_program_path if not program_path and self.ui: program_path = self.get_current_program_path(self.ui) filename = self.get_program_display_name(program_path) estimated_seconds = self.segment_estimates.get(segment_number) estimated_text = "" delta_text = "" delta_percent_text = "" if estimated_seconds is not None: estimated_text = self.format_elapsed_seconds(estimated_seconds) delta_text = self.format_signed_seconds(actual_seconds - estimated_seconds) delta_percent_text = self.format_delta_percent(estimated_seconds, actual_seconds) segment_text = "Segment %d" % segment_number timestamp = time.strftime("%Y-%m-%d %H:%M:%S") self.ensure_tool_segment_timing_log_header() with open(TOOL_SEGMENT_TIMING_LOG, "a") as f: f.write("%s | %-*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %-*.*s | %s\n" % ( timestamp, TOOL_LOG_PROGRAM_WIDTH, self.clean_log_field(filename), TOOL_LOG_SEGMENT_WIDTH, TOOL_LOG_SEGMENT_WIDTH, segment_text, TOOL_LOG_TOOL_WIDTH, TOOL_LOG_TOOL_WIDTH, self.clean_log_field(tool_text), TOOL_LOG_TIME_WIDTH, TOOL_LOG_TIME_WIDTH, self.clean_log_field(estimated_text), TOOL_LOG_TIME_WIDTH, TOOL_LOG_TIME_WIDTH, self.clean_log_field(self.format_elapsed_seconds(actual_seconds)), TOOL_LOG_DELTA_WIDTH, TOOL_LOG_DELTA_WIDTH, self.clean_log_field(delta_text), TOOL_LOG_PERCENT_WIDTH, TOOL_LOG_PERCENT_WIDTH, self.clean_log_field(delta_percent_text), TOOL_LOG_RESULT_WIDTH, TOOL_LOG_RESULT_WIDTH, self.clean_log_field(result_text), TOOL_LOG_RUN_WIDTH, TOOL_LOG_RUN_WIDTH, self.clean_log_field(self.active_run_id), self.clean_log_field(detail_text))) except Exception as e: self.write_status("tool segment timing log write error: " + str(e), constants.ALARM_LEVEL_LOW) def get_actual_log_estimate_rows(self): # The live logger starts Segment 1 at Cycle Start using the first program tool. # Time before the first M6 is folded into Segment 1 so estimates and actuals line up. rows = [] pre_tool_seconds = 0.0 segment_number = 1 for tool_num, segment_seconds in self.cycle_time.tool_segments: if tool_num is None: pre_tool_seconds += segment_seconds else: row_seconds = pre_tool_seconds + segment_seconds rows.append((segment_number, "T%d" % tool_num, row_seconds)) pre_tool_seconds = 0.0 segment_number += 1 if len(rows) == 0 and pre_tool_seconds > 0.0: rows.append((1, "", pre_tool_seconds)) return rows def set_loaded_estimates_from_current_file(self, program_path): self.loaded_estimate_rows = self.get_actual_log_estimate_rows() self.loaded_total_estimate_seconds = 0.0 for segment_number, tool_text, estimated_seconds in self.loaded_estimate_rows: try: self.loaded_total_estimate_seconds += float(estimated_seconds) except Exception: pass self.loaded_estimate_program_path = os.path.normpath(program_path) self.loaded_estimate_program_signature = self.get_program_signature(program_path) def loaded_estimates_match_program(self, program_path): if program_path and self.loaded_estimate_program_path: normalized_program_path = os.path.normpath(program_path) if normalized_program_path == os.path.normpath(self.loaded_estimate_program_path): return self.get_program_signature(program_path) == self.loaded_estimate_program_signature return False def ensure_estimates_for_program(self, program_path): # Normal path: estimates were calculated on file load. # Fallback path: recalculate at cycle start if the load callback was missed. if self.loaded_estimates_match_program(program_path): return if program_path and os.path.isfile(program_path): self.cycle_time.estimate_runtime(program_path) self.set_loaded_estimates_from_current_file(program_path) def get_loaded_estimate_map(self): estimate_map = {} for segment_number, tool_text, estimated_seconds in self.loaded_estimate_rows: estimate_map[segment_number] = estimated_seconds return estimate_map def format_raw_estimate_line(self, total_estimate_seconds): parts = [] if total_estimate_seconds is not None: parts.append("Raw Estimate: %s" % self.format_elapsed_seconds(total_estimate_seconds)) else: parts.append("Raw Estimate: unknown") for segment_number, tool_text, estimated_seconds in self.loaded_estimate_rows: if tool_text: parts.append("%s: %s" % (tool_text, self.format_elapsed_seconds(estimated_seconds))) else: parts.append("Seg %d: %s" % (segment_number, self.format_elapsed_seconds(estimated_seconds))) return " | ".join(parts) def _get_last_non_whitespace_line(self, full_path): last_line = None with open(full_path, 'r') as f: for raw in f: stripped = raw.strip() if stripped: last_line = stripped return last_line def _last_line_has_end_code(self, full_path): try: last_line = self._get_last_non_whitespace_line(full_path) if last_line: return re.search(r'^\s*(?:M30|M2)\s*(?:\(.*\)|;.*)?\s*$', last_line, re.IGNORECASE) is not None return False except Exception as e: self.write_status("M30 check error: " + str(e), constants.ALARM_LEVEL_LOW) return False def _is_nc_file(self, full_path): return os.path.splitext(full_path)[1].lower() == '.nc' def _show_m30_popup_async(self, full_path): if M30_CHECKER: pass else: return if self._is_nc_file(full_path): pass else: return with self._m30_prompt_lock: if self._m30_prompt_active.get(full_path): return self._m30_prompt_active[full_path] = True def _run_popup(): try: time.sleep(M30_POPUP_DELAY) if os.path.isfile(full_path): pass else: return if self._last_line_has_end_code(full_path): return subprocess.Popen([sys.executable, os.path.abspath(__file__), '--m30-popup', full_path], close_fds=True) except Exception as e: self.write_status("M30 popup launch error: " + str(e), constants.ALARM_LEVEL_LOW) finally: with self._m30_prompt_lock: if full_path in self._m30_prompt_active: del self._m30_prompt_active[full_path] t = threading.Thread(target=_run_popup) t.daemon = True t.start() def write_completed_cycle_time_snapshot(self, program_path=""): # Saves the completed tab .txt as an individual timestamped file. # Filename format: Completed_filename.nc.cycle_time_YYYY-MM-DD_HH-MM-SS_.txt try: if not os.path.isfile(CYCLE_TIME_TAB_STATUS): return if not os.path.isdir(CYCLE_TIME_TAB_SNAPSHOT_DIR): os.makedirs(CYCLE_TIME_TAB_SNAPSHOT_DIR) if program_path: program_name = os.path.basename(program_path) else: program_name = os.path.basename(self.active_program_path or self.loaded_estimate_program_path or "unknown_program.nc") safe_program_name = self.clean_snapshot_filename(program_name) timestamp_text = time.strftime("%Y-%m-%d_%H-%M-%S") snapshot_name = "Completed_%s.cycle_time_%s_.txt" % (safe_program_name, timestamp_text) snapshot_path = os.path.join(CYCLE_TIME_TAB_SNAPSHOT_DIR, snapshot_name) source_file = open(CYCLE_TIME_TAB_STATUS, "r") try: snapshot_text = source_file.read() finally: source_file.close() output_file = open(snapshot_path, "w") try: output_file.write(snapshot_text) if snapshot_text.endswith("\n") == False: output_file.write("\n") finally: output_file.close() except Exception as e: self.write_status("completed cycle time snapshot write error: " + str(e), constants.ALARM_LEVEL_LOW) def clean_snapshot_filename(self, filename): # Keep snapshot names readable while removing characters unsafe for filenames. text = str(filename or "unknown_program.nc") text = text.replace("/", "_") text = text.replace("\\", "_") text = text.replace(":", "-") text = text.replace("|", "_") text = text.replace("\t", " ") text = text.replace("\r", " ") text = text.replace("\n", " ") return text.strip() def format_tools_on_deck_seconds(self, seconds): # Compact overlay format. Keep it short because this is shown in the main UI overlay. try: seconds = int(round(float(seconds))) except Exception: seconds = 0 if seconds < 0: seconds = 0 hours = seconds // 3600 minutes = (seconds % 3600) // 60 seconds = seconds % 60 if hours > 0: if minutes > 0: return "%dh %dm" % (hours, minutes) return "%dh" % hours if minutes > 0: if seconds > 0: return "%dm %ds" % (minutes, seconds) return "%dm" % minutes return "%ds" % seconds def write_tools_on_deck_status_file(self, program_path, history, force_estimate_only=False): # Publishes a tiny, stable file for Tools On Deck. # Tools On Deck remains independent and ignores this file if the program/signature does not match. try: if program_path and os.path.isfile(program_path): pass else: return directory = os.path.dirname(CYCLE_TIME_TOOLS_ON_DECK_STATUS) if directory and not os.path.isdir(directory): os.makedirs(directory) program_signature = self.get_program_signature(program_path) active_segment_number = 0 if self.active_segment_number > 0: active_segment_number = self.active_segment_number lines = [] lines.append("program=%s" % os.path.normpath(program_path)) lines.append("signature=%s|%s" % (program_signature[0], program_signature[1])) lines.append("updated=%s" % time.strftime("%Y-%m-%d %H:%M:%S")) total_estimate_seconds = None if self.loaded_total_estimate_seconds is not None: total_estimate_seconds = self.loaded_total_estimate_seconds elif self.active_total_estimate_seconds is not None: total_estimate_seconds = self.active_total_estimate_seconds total_expected_seconds = None if total_estimate_seconds is not None: total_expected_seconds = self.get_expected_seconds( total_estimate_seconds, history.get("cycle_avg_seconds"), history.get("run_count", 0)) if total_expected_seconds is not None: if force_estimate_only == False and self.active_start_time is not None: elapsed_seconds = self.get_running_elapsed_seconds(self.active_start_time) remaining_seconds = self.get_remaining_seconds(total_expected_seconds, elapsed_seconds) if remaining_seconds is not None: if remaining_seconds < 0: lines.append("summary|%d|%d|over|%d" % ( int(round(float(elapsed_seconds))), int(round(float(total_expected_seconds))), int(round(abs(float(remaining_seconds)))))) else: lines.append("summary|%d|%d|rem|%d" % ( int(round(float(elapsed_seconds))), int(round(float(total_expected_seconds))), int(round(float(remaining_seconds))))) else: lines.append("summary|0|%d|est|%d" % ( int(round(float(total_expected_seconds))), int(round(float(total_expected_seconds))))) for segment_number, tool_text, estimated_seconds in self.loaded_estimate_rows: if tool_text: expected_seconds = self.get_segment_expected_seconds(segment_number, estimated_seconds, history) label_mode = "est" display_seconds = expected_seconds # Completed segments should keep their final rem/over result while the cycle is still active. # Without this, any non-active segment falls back to est as soon as the next tool starts. actual_segment = self.cycle_time_tab_actual_segments.get(segment_number) if force_estimate_only == False and self.active_start_time is not None and actual_segment is not None: actual_tool_text, actual_seconds, actual_result = actual_segment remaining_seconds = self.get_remaining_seconds(expected_seconds, actual_seconds) if remaining_seconds is not None: if remaining_seconds < 0: label_mode = "over" display_seconds = abs(remaining_seconds) else: label_mode = "rem" display_seconds = remaining_seconds # Active segment still gets live countdown / over-estimate behavior. elif force_estimate_only == False and self.active_start_time is not None and self.active_segment_start_time is not None and segment_number == active_segment_number: elapsed_seconds = self.get_running_elapsed_seconds(self.active_segment_start_time) remaining_seconds = self.get_remaining_seconds(expected_seconds, elapsed_seconds) if remaining_seconds is not None: if remaining_seconds < 0: label_mode = "over" display_seconds = abs(remaining_seconds) else: label_mode = "rem" display_seconds = remaining_seconds if display_seconds is not None: lines.append("%s|%s|%d" % (tool_text, label_mode, int(round(float(display_seconds))))) tmp_path = CYCLE_TIME_TOOLS_ON_DECK_STATUS + ".tmp" handle = open(tmp_path, "w") try: handle.write("\n".join(lines)) handle.write("\n") finally: handle.close() os.rename(tmp_path, CYCLE_TIME_TOOLS_ON_DECK_STATUS) except Exception as e: self.write_status("tools-on-deck status write error: " + str(e), constants.ALARM_LEVEL_LOW) def reset_tools_on_deck_status_file(self, program_path): # After a stop or completion, clear stale live rem/over labels from TOD. # The tab snapshot can keep the stopped/completed result, but TOD should return to estimates. try: if program_path and os.path.isfile(program_path): history = self.get_history_averages_for_program(program_path) self.write_tools_on_deck_status_file(program_path, history, True) except Exception as e: self.write_status("tools-on-deck reset error: " + str(e), constants.ALARM_LEVEL_LOW) def write_cycle_time_tab_snapshot_file(self, lines, history_lines=None): try: directory = os.path.dirname(CYCLE_TIME_TAB_STATUS) if directory and not os.path.isdir(directory): os.makedirs(directory) tmp_path = CYCLE_TIME_TAB_STATUS + ".tmp" f = open(tmp_path, "w") try: f.write("\n".join(lines)) f.write("\n") finally: f.close() os.rename(tmp_path, CYCLE_TIME_TAB_STATUS) if history_lines is not None: tmp_history_path = CYCLE_TIME_TAB_HISTORY + ".tmp" f = open(tmp_history_path, "w") try: f.write("\n".join(history_lines)) f.write("\n") finally: f.close() os.rename(tmp_history_path, CYCLE_TIME_TAB_HISTORY) except Exception as e: self.write_status("cycle time tab snapshot write error: " + str(e), constants.ALARM_LEVEL_LOW) def get_running_elapsed_seconds(self, start_time): if start_time is None: return None try: return max(0.0, time.time() - start_time) except Exception: return None def get_live_remaining_label(self, estimated_seconds, elapsed_seconds): # Used while a cycle or segment is still running. # Before expected time: "xs rem". After expected time: "xs over est". remaining_seconds = self.get_remaining_seconds(estimated_seconds, elapsed_seconds) if remaining_seconds is None: return "unknown", "remaining" if remaining_seconds < 0: return self.format_elapsed_seconds(abs(remaining_seconds)), "over estimate" return self.format_elapsed_seconds(remaining_seconds), "remaining" def get_remaining_seconds(self, estimated_seconds, elapsed_seconds): # Positive means time remaining. Negative means the live run is already over estimate. if estimated_seconds is None or elapsed_seconds is None: return None try: return float(estimated_seconds) - float(elapsed_seconds) except Exception: return None def program_is_paused(self): try: ui = self.ui or getattr(singletons, "g_Machine", None) if not ui: return False if getattr(ui.status, "interp_state", None) == linuxcnc.INTERP_PAUSED: return True if getattr(ui.status, "paused", False): return True except Exception: pass return False def get_live_status_text(self): if self.program_is_paused(): return "paused" if self.cycle_time_tab_waiting_for_tool_change: return "waiting" return "running" def history_edit_db_changed(self): # Lets the tab refresh after editing history, even while no job is running. current_signature = self.get_file_signature(CYCLE_TIME_HISTORY_EDIT_DB) if self._history_edit_db_signature is None: self._history_edit_db_signature = current_signature return False if current_signature != self._history_edit_db_signature: self._history_edit_db_signature = current_signature return True return False def refresh_cycle_time_tab_snapshot(self): try: if self.active_start_time is not None: self.update_tool_change_wait_state() self.publish_cycle_time_tab_snapshot() elif self.history_edit_db_changed(): program_path = self.loaded_estimate_program_path or self.active_program_path if program_path: self.publish_cycle_time_tab_snapshot(program_path) except Exception as e: self.write_status("cycle time tab live refresh error: " + str(e), constants.ALARM_LEVEL_LOW) return True def publish_cycle_time_tab_snapshot(self, program_path=""): try: if not program_path: program_path = self.active_program_path or self.loaded_estimate_program_path program_display = program_path if program_path else self.get_program_display_name(program_path) history = self.get_history_averages_for_program(program_path) history_edits = CycleTimeHistoryEdits(CYCLE_TIME_HISTORY_EDIT_DB) history_edits.load() display_run_id = self.active_run_id if display_run_id == "": display_run_id = self.cycle_time_tab_run_id program_line = "%s" % program_display total_estimate_seconds = None if self.loaded_total_estimate_seconds is not None: total_estimate_seconds = self.loaded_total_estimate_seconds elif self.active_total_estimate_seconds is not None: total_estimate_seconds = self.active_total_estimate_seconds total_expected_seconds, expected_source_text = self.get_expected_detail( total_estimate_seconds, history.get("cycle_avg_seconds"), history.get("run_count", 0)) history["cycle_expected_seconds"] = total_expected_seconds lines = [] lines.append(program_line) lines.append(" Updated: %s" % time.strftime("%Y-%m-%d %H:%M:%S")) # Original method line. Uncomment this and comment the next line to hide mean/median reference stats. # Current method line. Shows mean beside median so outlier influence is visible. # lines.append(" Method: %s" % expected_source_text.replace("Method: ", "")) lines.append(" %s" % self.format_cycle_method_line(expected_source_text, history)) lines.append(self.format_raw_estimate_line(total_estimate_seconds)) lines.append("") history_detail_lines = self.get_cycle_history_detail_lines(history) total_row = None total_progress_actual_seconds = None total_progress_result_text = "" segment_progress_bar = "" segment_progress_number = 1 if self.active_segment_number > 0: segment_progress_number = self.active_segment_number if self.cycle_time_tab_cycle_actual_seconds is not None: total_actual_seconds = self.cycle_time_tab_cycle_actual_seconds total_result_text = self.cycle_time_tab_cycle_result edited_total_seconds = history_edits.get_cycle_seconds(display_run_id, total_actual_seconds) if edited_total_seconds is None: total_result_text = "excluded" else: total_actual_seconds = edited_total_seconds total_progress_actual_seconds = total_actual_seconds total_progress_result_text = total_result_text total_row = self.make_timing_row( "Total:", total_expected_seconds, total_actual_seconds, total_result_text, False) elif self.active_start_time is not None: elapsed_seconds = self.get_running_elapsed_seconds(self.active_start_time) total_status_text = self.get_live_status_text() total_progress_actual_seconds = elapsed_seconds total_progress_result_text = total_status_text total_row = self.make_timing_row( "Total:", total_expected_seconds, elapsed_seconds, total_status_text, True) else: total_row = self.make_timing_row( "Total:", total_expected_seconds, None, "", False) total_row = self.apply_history_avg_to_row( total_row, total_expected_seconds) segment_rows = [] if len(self.loaded_estimate_rows) > 0: for segment_number, tool_text, estimated_seconds in self.loaded_estimate_rows: if tool_text: segment_name = "Seg %d / %s:" % (segment_number, tool_text) else: segment_name = "Seg %d:" % segment_number actual_data = self.cycle_time_tab_actual_segments.get(segment_number) if actual_data: actual_tool, actual_seconds, result_text = actual_data edited_actual_seconds = history_edits.get_segment_seconds(display_run_id, segment_number, actual_seconds) if edited_actual_seconds is None: result_text = "excluded" else: actual_seconds = edited_actual_seconds if segment_number == segment_progress_number: segment_expected_seconds = self.get_segment_expected_seconds(segment_number, estimated_seconds, history) segment_progress_bar = self.make_cycle_time_progress_bar("Progress:", segment_expected_seconds, actual_seconds, result_text == "completed") row = self.make_segment_timing_row_with_history( segment_number, segment_name, estimated_seconds, actual_seconds, result_text, False, history) segment_rows.append(row) else: if ( self.active_start_time is not None and self.active_segment_start_time is not None and segment_number == segment_progress_number ): elapsed_seconds = self.get_running_elapsed_seconds(self.active_segment_start_time) segment_status_text = self.get_live_status_text() segment_expected_seconds = self.get_segment_expected_seconds(segment_number, estimated_seconds, history) segment_progress_bar = self.make_cycle_time_progress_bar("Progress:", segment_expected_seconds, elapsed_seconds, False) row = self.make_segment_timing_row_with_history( segment_number, segment_name, estimated_seconds, elapsed_seconds, segment_status_text, True, history) if len(self.loaded_estimate_rows) > 1: row["active_segment"] = True segment_rows.append(row) else: row = self.make_segment_timing_row_with_history( segment_number, segment_name, estimated_seconds, None, "", False, history) if segment_number == segment_progress_number: segment_expected_seconds = self.get_segment_expected_seconds(segment_number, estimated_seconds, history) segment_progress_bar = self.make_cycle_time_progress_bar("Progress:", segment_expected_seconds, None, False) segment_rows.append(row) all_rows = [] if total_row is not None: all_rows.append(total_row) all_rows.extend(segment_rows) formatted_rows = self.format_timing_rows(all_rows) lines.append("===== Total Cycle Run =====") total_progress_bar = self.make_cycle_time_progress_bar("Progress:", total_expected_seconds, total_progress_actual_seconds, total_progress_result_text == "completed") if total_progress_bar: lines.append(total_progress_bar) if len(formatted_rows) > 0: lines.append(formatted_rows[0]) else: lines.append("No total cycle estimate is available.") lines.append("") lines.append("===== Tool Segments =====") if segment_progress_bar: lines.append(segment_progress_bar) if len(segment_rows) == 0: lines.append("No tool segment estimate is available.") else: for row_text in formatted_rows[1:]: lines.append(row_text) self.write_tools_on_deck_status_file(program_path, history) self.write_cycle_time_tab_snapshot_file(lines, history_detail_lines) except Exception as e: self.write_status("cycle time tab snapshot error: " + str(e), constants.ALARM_LEVEL_LOW) def format_status_seconds(self, seconds): # Status window uses long units so calculated and median times align cleanly. try: total_seconds = int(round(float(seconds))) except Exception: total_seconds = 0 minutes = total_seconds // 60 seconds = total_seconds % 60 if minutes > 0: return "%d min %2d sec" % (minutes, seconds) return "%2d sec" % seconds def write_estimate_status_block(self, full_path, runtime_min, units): try: calculated_seconds = float(runtime_min) * 60.0 calculated_text = self.format_status_seconds(calculated_seconds) history = self.get_history_averages_for_program(full_path) run_count, last_run_dt, last_run_type = self.last_run.get_run_history(full_path) output_lines = [] output_lines.append("Cycle Estimation:") summary_label_width = len("Calculated run time") cycle_median_seconds = history.get("cycle_avg_seconds") if cycle_median_seconds is not None: cycle_median_text = self.format_status_seconds(cycle_median_seconds) else: cycle_median_text = "no completed runs" summary_value_width = max(len(calculated_text), len(cycle_median_text)) output_lines.append(" %*s: %*s" % (summary_label_width, "Calculated run time", summary_value_width, calculated_text)) output_lines.append(" %*s: %*s" % (summary_label_width, "Median run time", summary_value_width, cycle_median_text)) if len(self.loaded_estimate_rows) > 0: output_lines.append(" ") output_lines.append(" Calculated time by tool segment:") segment_averages = history.get("segment_avg_seconds", {}) segment_value_texts = [] segment_name_width = 0 calculated_width = 0 median_width = 0 for segment_number, tool_text, segment_seconds in self.loaded_estimate_rows: if tool_text: tool_label_width = 4 segment_name = "Segment %d / %-*s:" % (segment_number, tool_label_width, tool_text) else: segment_name = "Segment %d:" % segment_number calculated_segment_text = self.format_status_seconds(segment_seconds) if segment_number in segment_averages: median_segment_text = self.format_status_seconds(segment_averages[segment_number]) else: median_segment_text = "no median" segment_value_texts.append((segment_name, calculated_segment_text, median_segment_text)) if len(segment_name) > segment_name_width: segment_name_width = len(segment_name) if len(calculated_segment_text) > calculated_width: calculated_width = len(calculated_segment_text) if len(median_segment_text) > median_width: median_width = len(median_segment_text) for segment_name, calculated_segment_text, median_segment_text in segment_value_texts: output_lines.append(" %-*s %*s (%*s median)" % ( segment_name_width, segment_name, calculated_width, calculated_segment_text, median_width, median_segment_text)) output_lines.append(" ") output_lines.append(" Run history: %s total runs (partial + complete)" % run_count) if last_run_dt is None: output_lines.append(" Last run: no run history found in archive log.") else: now = datetime.now() elapsed = now - last_run_dt run_type_text = last_run_type if last_run_type else "unknown" output_lines.append(" Last run: %s (%s) [%s]" % (last_run_dt.strftime('%Y-%m-%d %H:%M:%S'), self.last_run.format_elapsed_since(elapsed), run_type_text)) self._write_status_block(output_lines) except Exception as e: self.write_status("estimate status block error: " + str(e), constants.ALARM_LEVEL_LOW) def _write_status_block(self, lines): try: block_lines = [] block_lines.append("#######################################") block_lines.extend(lines) block_lines.append("#######################################") block_lines.append(" ") for line in reversed(block_lines): self.error_handler.write(line, constants.ALARM_LEVEL_QUIET) except Exception as e: self.write_status("Cycle Time status write error: " + str(e), constants.ALARM_LEVEL_LOW) def handle_file_loaded(self, loaded_path): # One plugin owns the file-load event, so estimates cannot get out of sync with actual tracking. loaded_path = os.path.normpath(loaded_path) self.write_file_loaded_log(loaded_path) self.reset_program_tracking("new file loaded", loaded_path) if os.path.isfile(loaded_path): pass else: self.write_status("Cycle Time file not found: " + loaded_path, constants.ALARM_LEVEL_LOW) return self._show_m30_popup_async(loaded_path) runtime_min, units = self.cycle_time.estimate_runtime(loaded_path) self.set_loaded_estimates_from_current_file(loaded_path) self.write_estimate_status_block(loaded_path, runtime_min, units) self.publish_cycle_time_tab_snapshot(loaded_path) def write_stopped_segment_time(self, end_time): if self.active_segment_tool and self.active_segment_start_time is not None: elapsed_seconds = end_time - self.active_segment_start_time elapsed_text = self.format_elapsed_seconds(elapsed_seconds) self.write_segment_log("Segment stopped prematurely", self.active_segment_number, self.active_segment_tool, elapsed_text, "Program stopped before normal completion") self.write_tool_segment_timing_log(self.active_segment_number, self.active_segment_tool, elapsed_seconds, "stopped", "Program stopped before normal completion") self.cycle_time_tab_actual_segments[self.active_segment_number] = (self.active_segment_tool, elapsed_seconds, "stopped") self.publish_cycle_time_tab_snapshot() self.write_status(" ") self.write_status("========================================") self.write_status("Segment %d / %s was stopped after %s." % (self.active_segment_number, self.active_segment_tool, elapsed_text), constants.ALARM_LEVEL_QUIET) self.write_status("========================================") self.write_status(" ") def write_completed_segment_time(self, end_time): if self.active_segment_tool and self.active_segment_start_time is not None: elapsed_seconds = end_time - self.active_segment_start_time elapsed_text = self.format_elapsed_seconds(elapsed_seconds) self.write_segment_log("Segment completed", self.active_segment_number, self.active_segment_tool, elapsed_text, "Tool segment ended normally") self.write_tool_segment_timing_log(self.active_segment_number, self.active_segment_tool, elapsed_seconds, "completed", "Tool segment ended normally") self.cycle_time_tab_actual_segments[self.active_segment_number] = (self.active_segment_tool, elapsed_seconds, "completed") self.publish_cycle_time_tab_snapshot() self.write_status(" ") self.write_status("========================================") self.write_status("Segment %d / %s took %s to complete." % (self.active_segment_number, self.active_segment_tool, elapsed_text), constants.ALARM_LEVEL_QUIET) self.write_status("========================================") self.write_status(" ") def reset_program_tracking(self, reason="", program_path=""): # Clears incomplete run state. This is intentionally not a completed segment log. self.active_program_path = program_path if program_path else None self.active_start_time = None self.active_segment_tool = "" self.active_segment_start_time = None self.active_segment_number = 0 self.active_run_id = "" self.active_total_estimate_seconds = None self.segment_estimates = {} self.cycle_time_tab_actual_segments = {} self.cycle_time_tab_cycle_actual_seconds = None self.cycle_time_tab_cycle_result = "" self.cycle_time_tab_run_id = "" self.cycle_time_tab_waiting_for_tool_change = False self.cycle_time_tab_waiting_tool_text = "" self.cycle_time_tab_waiting_since = None self.last_tool_change_key = "" self.last_tool_change_time = 0.0 if reason: self.write_status("segment timing reset: " + reason, constants.ALARM_LEVEL_QUIET) def format_history_timestamp(self, timestamp_text): # Converts log timestamps to compact aligned tab text, for example: " 6/07 9:54". try: dt_val = datetime.strptime(str(timestamp_text).strip(), "%Y-%m-%d %H:%M:%S") return "%2d/%02d %2d:%02d" % (dt_val.month, dt_val.day, dt_val.hour, dt_val.minute) except Exception: return str(timestamp_text).strip() def format_elapsed_seconds(self, elapsed_seconds): try: elapsed_seconds = int(round(elapsed_seconds)) except Exception: elapsed_seconds = 0 if elapsed_seconds < 0: elapsed_seconds = 0 hours = elapsed_seconds // 3600 minutes = (elapsed_seconds % 3600) // 60 seconds = elapsed_seconds % 60 if hours > 0: return "%dh %02dm %02ds" % (hours, minutes, seconds) if minutes > 0: return "%dm %02ds" % (minutes, seconds) return "%ds" % seconds def apply_email_tokens(self, text, event_text, program_path, elapsed_text, tool_text, message_text): filename = self.get_program_display_name(program_path) text = text.replace("{event}", event_text) text = text.replace("{filename}", filename) text = text.replace("{path}", program_path if program_path else "Unknown path") text = text.replace("{elapsed}", elapsed_text) text = text.replace("{tool}", tool_text if tool_text else "N/A") text = text.replace("{message}", message_text if message_text else "") return text def send_email_async(self, event_text, program_path="", elapsed_seconds=None, tool_text="", message_text=""): t = threading.Thread( target=self.send_email, args=(event_text, program_path, elapsed_seconds, tool_text, message_text) ) t.daemon = True t.start() def send_operation_complete_email_async(self, program_path="", elapsed_seconds=None): self.send_email_async( "Operation Complete", program_path, elapsed_seconds, "", "" ) def send_tool_change_email_async(self, tool_text, message_text=""): program_path = "" try: ui = self.ui or getattr(singletons, "g_Machine", None) if ui: program_path = self.get_current_program_path(ui) except Exception: pass self.send_email_async( "Tool Change Requested", program_path, None, tool_text, message_text ) def send_email(self, event_text, program_path="", elapsed_seconds=None, tool_text="", message_text=""): try: config = self._load_config() sender = config.get("email", "sender") recipient = config.get("email", "recipient") subject = config.get("email", "subject") body = config.get("email", "body").replace("\\n", "\n") smtp_server = config.get("email", "smtp_server") smtp_port = config.getint("email", "smtp_port") smtp_username = config.get("email", "smtp_username") smtp_password = config.get("email", "smtp_password") if elapsed_seconds is None: elapsed_text = "Unknown" else: elapsed_text = self.format_elapsed_seconds(elapsed_seconds) body_had_tokens = ( "{event}" in body or "{filename}" in body or "{path}" in body or "{elapsed}" in body or "{tool}" in body or "{message}" in body ) subject = self.apply_email_tokens(subject, event_text, program_path, elapsed_text, tool_text, message_text) body = self.apply_email_tokens(body, event_text, program_path, elapsed_text, tool_text, message_text) if body_had_tokens == False: body = ( body + "\n\nEvent: " + event_text + "\nProgram: " + self.get_program_display_name(program_path) + "\nElapsed: " + elapsed_text + "\nTool: " + (tool_text if tool_text else "N/A") ) if message_text: body = body + "\nMessage: " + message_text msg = MIMEText(body) msg["From"] = sender msg["To"] = recipient msg["Subject"] = subject smtp = smtplib.SMTP_SSL(smtp_server, smtp_port, timeout=15) smtp.login(smtp_username, smtp_password) smtp.sendmail(sender, [recipient], msg.as_string()) smtp.quit() if tool_text: self.write_status("email sent for " + event_text + " " + tool_text) else: self.write_status("email sent for " + self.get_program_display_name(program_path)) self.write_status(" ") except Exception as e: self.write_status("email send failed: " + str(e), constants.ALARM_LEVEL_QUIET) def setup_key_handler(self): main_ui = getattr(singletons, "g_Machine", None) if main_ui and hasattr(main_ui, "window"): main_ui.window.add_events(gtk.gdk.KEY_PRESS_MASK) main_ui.window.connect("key-press-event", self.on_global_key_press) self.write_status("Ctrl+H bound to mode cycle") return False def on_global_key_press(self, widget, event): if (event.state & gtk.gdk.CONTROL_MASK and event.keyval == gtk.keysyms.h): self.cycle_notify_mode() return True return False def setup_hook(self): while True: try: if hasattr(singletons, "g_Machine") and singletons.g_Machine: self.ui = singletons.g_Machine err_handler = getattr(self.ui, "error_handler", None) if err_handler: self.error_handler = err_handler else: raise Exception("No error_handler found") self.cycle_time.error_handler = self.error_handler self.last_run.error_handler = self.error_handler if self.original_interp_task_status_change is None: self.original_interp_task_status_change = self.ui.handle_interp_task_status_change self.ui.handle_interp_task_status_change = types.MethodType( self.handle_interp_task_status_change_wrapper, self.ui ) self.write_status("completion hook installed") if self.original_set_message_line_text is None and hasattr(self.ui, "set_message_line_text"): self.original_set_message_line_text = self.ui.set_message_line_text self.ui.set_message_line_text = types.MethodType( self.set_message_line_text_wrapper, self.ui ) self.write_status("tool-change message hook installed") if self.original_load_gcode_file is None and hasattr(self.ui, "load_gcode_file"): if getattr(self.ui, "_cycle_time_monitor_load_gcode_wrapped", False) == False: self.original_load_gcode_file = self.ui.load_gcode_file self.ui.load_gcode_file = types.MethodType( self.load_gcode_file_wrapper, self.ui ) self.ui._cycle_time_monitor_load_gcode_wrapped = True self.write_status("file-load segment reset hook installed") return except Exception as e: self.write_status("hook setup error: " + str(e), constants.ALARM_LEVEL_QUIET) time.sleep(0.5) def load_gcode_file_wrapper(self, ui, path, *args, **kwargs): result = self.original_load_gcode_file(path, *args, **kwargs) try: loaded_path = self.get_current_program_path(ui) if loaded_path and path: if os.path.normpath(loaded_path) == os.path.normpath(path): self.handle_file_loaded(loaded_path) except Exception as e: self.write_status("file-load monitor error: " + str(e), constants.ALARM_LEVEL_QUIET) return result def handle_interp_task_status_change_wrapper(self, ui): prev_interp = ui.prev_lcnc_interp_state if self.original_interp_task_status_change: self.original_interp_task_status_change() if self.program_just_completed(ui, prev_interp): if self.program_ended_cleanly(ui): self.handle_program_completed(ui) else: self.handle_program_stopped(ui) return if self.program_just_started(ui, prev_interp): self.begin_new_program_run(ui) return if self.program_is_running(ui): self.update_tool_change_wait_state() self.capture_program_start(ui) return def set_message_line_text_wrapper(self, ui, message): try: self.handle_possible_tool_change_message(message) except Exception as e: self.write_status("tool-change detect error: " + str(e), constants.ALARM_LEVEL_QUIET) return self.original_set_message_line_text(message) def normalize_prompt_message(self, message): try: if message is None: return "" text = str(message) text = text.replace("*", " ") text = text.replace("$$REPLY_TEXT$$", "") text = re.sub(r"\s+", " ", text) return text.strip() except Exception: return "" def extract_tool_from_message(self, message): text = self.normalize_prompt_message(message) patterns = [ r"\bInsert\s+T\s*([0-9]+)\b", r"\bInsert\s+tool\s*([0-9]+)\b", r"\bT\s*([0-9]+)\b" ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return "T" + match.group(1), text return "", text def handle_possible_tool_change_message(self, message): tool_text, clean_message = self.extract_tool_from_message(message) if not tool_text: return now = time.time() dedupe_key = tool_text + "|" + clean_message if ( dedupe_key == self.last_tool_change_key and (now - self.last_tool_change_time) < TOOL_CHANGE_SECS ): return self.last_tool_change_key = dedupe_key self.last_tool_change_time = now if self.active_segment_start_time is None: self.active_segment_number = 1 self.active_segment_tool = tool_text self.active_segment_start_time = now elif tool_text != self.active_segment_tool: self.write_completed_segment_time(now) self.active_segment_number += 1 self.active_segment_tool = tool_text self.active_segment_start_time = now self.cycle_time_tab_waiting_for_tool_change = True self.cycle_time_tab_waiting_tool_text = tool_text self.cycle_time_tab_waiting_since = now self.publish_cycle_time_tab_snapshot() self.write_status("detected tool change request: " + tool_text) self.write_segment_log("Tool change requested", self.active_segment_number, tool_text, "", clean_message) if self.notify_tool_change_enabled(): self.send_tool_change_email_async(tool_text, clean_message) def program_just_started(self, ui, prev_interp): return ( self.program_is_running(ui) and prev_interp == linuxcnc.INTERP_SEEKING ) def program_is_running(self, ui): return ( ui.status.interp_state not in ( linuxcnc.INTERP_IDLE, linuxcnc.INTERP_SEEKING ) and ui.status.task_mode == linuxcnc.MODE_AUTO and ui.status.task_state == linuxcnc.STATE_ON ) def program_ended_cleanly(self, ui): return ( getattr(ui.status, "program_ended", False) or getattr(ui.status, "program_ended_and_reset", False) ) def program_just_completed(self, ui, prev_interp): # ESC can switch task_mode back to MODE_MANUAL before interp_state reaches IDLE. # Use the interpreter transition as the source of truth for end/stop detection. return ( ui.status.interp_state == linuxcnc.INTERP_IDLE and prev_interp not in ( linuxcnc.INTERP_IDLE, linuxcnc.INTERP_SEEKING ) and ui.status.task_state == linuxcnc.STATE_ON ) def begin_new_program_run(self, ui): # New cycle starts must not inherit segment timing from an aborted prior run. now = time.time() program_path = self.get_current_program_path(ui) self.ensure_estimates_for_program(program_path) self.active_program_path = program_path self.active_start_time = now self.active_run_id = self.make_run_id(now) self.cycle_time_tab_run_id = self.active_run_id self.active_total_estimate_seconds = self.loaded_total_estimate_seconds self.segment_estimates = self.get_loaded_estimate_map() self.cycle_time_tab_actual_segments = {} self.cycle_time_tab_cycle_actual_seconds = None self.cycle_time_tab_cycle_result = "" self.cycle_time_tab_waiting_for_tool_change = False self.cycle_time_tab_waiting_tool_text = "" self.cycle_time_tab_waiting_since = None self.active_segment_number = 1 self.active_segment_tool = self.get_initial_segment_tool(ui, program_path) self.active_segment_start_time = now self.last_tool_change_key = "" self.last_tool_change_time = 0.0 self.write_segment_log("Cycle started", self.active_segment_number, self.active_segment_tool, "", "New Cycle Start detected") self.publish_cycle_time_tab_snapshot(program_path) def capture_program_start(self, ui): program_path = self.get_current_program_path(ui) if self.active_start_time is None: self.begin_new_program_run(ui) return if program_path and self.active_program_path: if os.path.normpath(program_path) != os.path.normpath(self.active_program_path): self.begin_new_program_run(ui) return def handle_program_stopped(self, ui): program_path = self.active_program_path if not program_path: program_path = self.get_current_program_path(ui) filename = self.get_program_display_name(program_path) end_time = time.time() self.write_stopped_segment_time(end_time) if self.active_start_time is not None: elapsed_seconds = end_time - self.active_start_time self.write_cycle_run_timing_log(elapsed_seconds, "stopped", "Program stopped before normal completion") self.cycle_time_tab_cycle_actual_seconds = elapsed_seconds self.cycle_time_tab_cycle_result = "stopped" self.publish_cycle_time_tab_snapshot(program_path) self.write_status(" ") self.write_status("detected program stop: " + filename) self.write_segment_log("Program stopped", 0, "", "", filename) self.reset_program_tracking("program stopped", program_path) self.reset_tools_on_deck_status_file(program_path) def handle_program_completed(self, ui): program_path = self.active_program_path if not program_path: program_path = self.get_current_program_path(ui) end_time = time.time() if self.active_start_time is None: elapsed_seconds = None else: elapsed_seconds = end_time - self.active_start_time filename = self.get_program_display_name(program_path) self.write_completed_segment_time(end_time) if elapsed_seconds is not None: self.write_cycle_run_timing_log(elapsed_seconds, "completed", "Program completed normally") self.cycle_time_tab_cycle_actual_seconds = elapsed_seconds self.cycle_time_tab_cycle_result = "completed" self.publish_cycle_time_tab_snapshot(program_path) self.write_completed_cycle_time_snapshot(program_path) self.write_status(" ") self.write_status("detected program end: " + filename) self.write_segment_log("Program completed", 0, "", self.format_elapsed_seconds(elapsed_seconds) if elapsed_seconds is not None else "Unknown", filename) if self.notify_completion_enabled(): self.send_operation_complete_email_async(program_path, elapsed_seconds) self.active_program_path = None self.active_start_time = None self.active_segment_tool = "" self.active_segment_start_time = None self.active_segment_number = 0 self.active_run_id = "" self.active_total_estimate_seconds = None self.segment_estimates = {} self.reset_tools_on_deck_status_file(program_path) class FakeConfigFile(object): def __init__(self, text): self.lines = text.splitlines(True) def readline(self): if self.lines: return self.lines.pop(0) return "" if __name__ == '__main__': if len(sys.argv) >= 3 and sys.argv[1] == '--m30-popup': sys.exit(M30PopupProcess(sys.argv[2]).run()) DESCRIPTION_LONG = """Cycle Time Monitor estimates cycle time when a program is loaded, logs actual elapsed time by tool segment while the program runs, and can optionally send email notifications for tool-change prompts and program completion.

"""