Python API

Auto-generated Python API documentation for OctoPrint-TempETA.

Main Plugin Class

Bases: PrinterCallbackBase, StartupPluginBase, TemplatePluginBase, SettingsPluginBase, AssetPluginBase, EventHandlerPluginBase, SimpleApiPluginBase

Main plugin implementation for Temperature ETA.

Implements OctoPrint Issue #469: Show estimated time remaining for printer heating (bed, hotend, chamber).

Initialize plugin with temperature history tracking.

Source code in octoprint_temp_eta/__init__.py

def __init__(self):
    """Initialize plugin with temperature history tracking."""
    super().__init__()
    self._lock = threading.Lock()
    # Serializes profile-switch sequences against each other. Acquired
    # around the persist -> id-swap -> history-replace flow performed
    # inside the switch handler so two concurrent switches (and the
    # switch's own internal persist step) cannot interleave. Persists
    # triggered elsewhere are not serialized by this lock.
    self._profile_switch_lock = threading.Lock()

    self._debug_logging_enabled = False
    self._last_debug_log_time = 0.0
    self._last_heater_support_decision = {}
    self._heater_supported_cache: dict[str, bool] = {}

    # MQTT publisher (initialized in on_after_startup when logger is
    # available; delegates publishing to the OctoPrint-MQTT plugin)
    self._mqtt_publisher: Optional[MqttPublisher] = None

    # Number of temperature samples to keep per heater.
    # This is configurable via settings (history_size). We cache the active
    # value here for fast access in the 2Hz callback.
    self._default_history_size = 60
    self._history_maxlen = self._default_history_size

    # Persist history per printer profile (file per profile id).
    # Note: ETA uses only the most recent seconds of history, so we keep
    # persistence bounded by both maxlen and a max-age filter on load.
    self._active_profile_id: Optional[str] = None
    # Persistence backoff to reduce SD card wear in long-running phases
    # where the target is never reached.
    # Backoff sequence after phase reset: 30s -> 60s -> 120s -> 240s -> 300s (cap).
    self._persist_backoff_reset_s = 30.0
    self._persist_backoff_initial_s = 60.0
    self._persist_backoff_max_s = 300.0
    self._persist_backoff_current_s = self._persist_backoff_initial_s
    self._next_persist_time = 0.0
    self._persist_phase_active = False
    # Guards the persist-backoff triplet (_persist_phase_active,
    # _persist_backoff_current_s, _next_persist_time) which is mutated from
    # the temperature callback, profile switches, and disconnect events.
    # Dedicated lock (not self._lock) so these short sections never nest
    # with the history lock taken by _persist_current_profile_history.
    self._persist_state_lock = threading.Lock()
    self._last_persist_size_warning_time = 0.0
    self._persist_max_json_bytes = 256 * 1024
    self._persist_max_age_seconds = 180.0
    self._history_dirty = False
    # Monotonic counter bumped whenever new history data is recorded.
    # A persist captures this under the lock and only clears the dirty
    # flag if it is unchanged after the write (no concurrent update).
    self._history_dirty_epoch = 0

    self._temp_history = {
        "bed": deque(maxlen=self._history_maxlen),
        "tool0": deque(maxlen=self._history_maxlen),
        "chamber": deque(maxlen=self._history_maxlen),
    }

    # Cooldown history (target==0). Kept separate from heating history so
    # the heat-up ETA fit doesn't get polluted by cooldown samples.
    self._cooldown_history = {
        "bed": deque(maxlen=self._history_maxlen),
        "tool0": deque(maxlen=self._history_maxlen),
        "chamber": deque(maxlen=self._history_maxlen),
    }

    # Ambient baseline per heater for ambient-mode cooldown.
    # If the user doesn't provide an ambient temperature, we rely on the
    # lowest temperature observed while the heater target is OFF (within a
    # reasonable range) as a proxy for ambient.
    self._cooldown_ambient_baseline: dict[str, Optional[float]] = {
        "bed": None,
        "tool0": None,
        "chamber": None,
    }
    self._last_update_time = 0.0

    # Cache hot-path settings values for fast access in the ~2Hz callback.
    # Refreshed on startup and on settings save.
    self._threshold_start_c = 5.0
    self._update_interval_s = 1.0
    self._last_runtime_cache_refresh_time = 0.0

    # Track last seen target per heater so we can detect transitions like
    # heating -> off (cooldown start).
    self._last_target_by_heater: dict[str, float] = {}

    # When enabled, suppress ETA updates while a print job is active.
    # This keeps the UI focused on the pre-print heat-up phase.
    self._suppressing_due_to_print = False
    # Guards _suppressing_due_to_print so the one-shot frontend clear is
    # not triggered twice by concurrent temperature callbacks and events.
    self._suppress_lock = threading.Lock()

    self._last_settings_snapshot_log_time = 0.0

on_after_startup

on_after_startup()

Called after OctoPrint startup, register for temperature updates.

Source code in octoprint_temp_eta/__init__.py

def on_after_startup(self):
    """Called after OctoPrint startup, register for temperature updates."""
    self._logger.info("Temperature ETA Plugin started")
    self._printer.register_callback(self)

    self._refresh_debug_logging_flag()
    self._debug_log("Debug logging enabled")

    self._refresh_runtime_caches()

    # Apply configured history size now that settings are available.
    self._set_history_maxlen(self._read_history_maxlen_setting())

    # Load persisted history for the active printer profile.
    self._switch_active_profile_if_needed(force=True)

    # Initialize MQTT publishing via the OctoPrint-MQTT plugin helper
    self._mqtt_publisher = MqttPublisher(
        self._logger, plugin_version=str(getattr(self, "_plugin_version", ""))
    )
    self._mqtt_publisher.initialize(self._plugin_manager)
    self._configure_mqtt_publisher()
    self._mqtt_publisher.publish_discovery(self._get_discovery_heaters())

on_printer_add_temperature

on_printer_add_temperature(data)

Called when new temperature data is available (~2Hz).

Parameters:

Name	Type	Description	Default
data	dict	Temperature data from OctoPrint { "bed": {"actual": float, "target": float}, "tool0": {"actual": float, "target": float}, ... }	required

Source code in octoprint_temp_eta/__init__.py

def on_printer_add_temperature(self, data):
    """Called when new temperature data is available (~2Hz).

    Args:
        data (dict): Temperature data from OctoPrint
            {
                "bed": {"actual": float, "target": float},
                "tool0": {"actual": float, "target": float},
                ...
            }
    """
    if not self._settings.get_boolean(["enabled"]):
        return

    # Ensure we are tracking the right profile's history.
    self._switch_active_profile_if_needed()

    # If enabled, suppress ETA completely while OctoPrint considers a print job active.
    if self._suppress_while_printing_enabled() and self._is_print_job_active():
        # Clear once when suppression starts to avoid stale countdowns.
        # Check-and-set under the lock so the one-shot clear cannot be
        # triggered twice by concurrent callbacks/events.
        with self._suppress_lock:
            just_started = not self._suppressing_due_to_print
            self._suppressing_due_to_print = True
        if just_started:
            self._clear_all_heaters_frontend()
        return

    # If we were suppressing but the condition no longer applies (e.g. warm-up resumed or print ended), re-enable.
    with self._suppress_lock:
        self._suppressing_due_to_print = False

    current_time = time.time()

    # Keep cached settings reasonably fresh even if settings change outside
    # of the normal save flow (e.g. tests or edge-case integrations).
    if (current_time - self._last_runtime_cache_refresh_time) >= 5.0:
        self._last_runtime_cache_refresh_time = current_time
        self._refresh_runtime_caches()

    # Periodic settings snapshot for debugging.
    self._debug_log_settings_snapshot(current_time)

    threshold = float(getattr(self, "_threshold_start_c", 5.0))
    update_interval = float(getattr(self, "_update_interval_s", 1.0))
    heating_enabled = self._heating_enabled()
    cooldown_enabled = self._cooldown_enabled()

    # Avoid eager f-string formatting in the hot path.
    try:
        is_logger_debug = bool(getattr(self._logger, "isEnabledFor")(10))  # type: ignore[attr-defined]
    except _EXPECTED_ERRORS:
        is_logger_debug = False
    if is_logger_debug:
        heaters_in_data = [k for k, v in data.items() if isinstance(v, Mapping)]
        if heaters_in_data:
            self._logger.debug(
                "Received temperature data for heaters: %s", heaters_in_data
            )

    epsilon_hold = 0.2

    # Track whether we are in an active heating phase (at least one heater
    # is above the threshold window and not holding). This drives persistence
    # backoff scheduling.
    phase_active_now = False
    target_changed_in_active_phase = False

    recorded_count = 0
    recorded_cooldown_count = 0
    heaters_seen = 0
    with self._lock:
        # Protect shared history state (OctoPrint may call callbacks from worker threads).
        for heater, temps in data.items():
            # OctoPrint wraps heater entries in frozendict when
            # devel.useFrozenDictForPrinterState is enabled, which is not a
            # dict subclass; accept any Mapping so ETA still works there.
            if not isinstance(temps, Mapping):
                continue
            heater_key = str(heater)
            heaters_seen += 1

            # Record samples only while ETA could be shown (active target and above threshold).
            target_raw = temps.get("target", 0)
            actual_raw = temps.get("actual")
            if actual_raw is None:
                continue
            try:
                actual = float(actual_raw)
            except _EXPECTED_ERRORS:
                continue

            try:
                target = float(target_raw or 0)
            except _EXPECTED_ERRORS:
                # Some firmwares/virtual printer formats may provide a non-numeric
                # target (e.g. "off"). Treat that as OFF for cooldown tracking.
                target = 0.0
                self._debug_log_throttled(
                    current_time,
                    30.0,
                    "Non-numeric target treated as OFF heater=%s target_raw=%r",
                    str(heater),
                    target_raw,
                )

            prev_target = self._last_target_by_heater.get(str(heater))
            self._last_target_by_heater[heater_key] = float(target)

            if target <= 0:
                # Cooldown tracking (target==0).
                if cooldown_enabled:
                    if heater_key not in self._cooldown_history:
                        self._cooldown_history[heater_key] = deque(
                            maxlen=self._history_maxlen
                        )

                    # If we just transitioned from heating to OFF, start a fresh
                    # cooldown history so our linear cooldown fit doesn't include
                    # old OFF samples from before the heat-up phase.
                    if prev_target is not None and prev_target > 0:
                        self._cooldown_history[heater_key].clear()
                        self._debug_log_throttled(
                            current_time,
                            10.0,
                            "Cooldown start detected, cleared history heater=%s prev_target=%.1f actual=%.1f",
                            str(heater),
                            float(prev_target),
                            float(actual),
                        )

                    self._cooldown_history[heater_key].append(
                        (current_time, actual)
                    )
                    recorded_cooldown_count += 1

                    # Track a baseline ambient temp while OFF.
                    # We only learn baseline values in a sane range to
                    # avoid "ambient" being polluted by still-hot cooldown.
                    if math.isfinite(actual) and actual < 120.0:
                        prev = self._cooldown_ambient_baseline.get(heater_key)
                        if prev is None or actual < prev:
                            self._cooldown_ambient_baseline[heater_key] = actual
                continue

            if not heating_enabled:
                continue

            remaining = target - actual
            # Avoid recording while holding near target to prevent constant churn.
            if remaining <= epsilon_hold:
                continue
            if remaining < threshold:
                continue

            # This heater is actively heating within the ETA window.
            phase_active_now = True
            if (
                prev_target is not None
                and math.isfinite(prev_target)
                and math.isfinite(target)
                and prev_target > 0.0
                and target > 0.0
                and abs(prev_target - target) >= 0.5
            ):
                target_changed_in_active_phase = True

            if heater_key not in self._temp_history:
                self._temp_history[heater_key] = deque(maxlen=self._history_maxlen)

            self._temp_history[heater_key].append((current_time, actual, target))
            self._history_dirty = True
            self._history_dirty_epoch += 1
            recorded_count += 1

    # Avoid flooding logs while idle/holding temperature: log far less often
    # when we didn't record any samples.
    debug_interval = 5.0 if recorded_count > 0 else 60.0
    self._debug_log_throttled(
        current_time,
        debug_interval,
        "Temp callback profile=%s heaters=%d recorded=%d cooldown_recorded=%d threshold=%.2f",
        str(self._active_profile_id),
        int(heaters_seen),
        recorded_count,
        recorded_cooldown_count,
        float(threshold),
    )

    # Atomic check-and-set of the broadcast/persist cadence gate so two
    # concurrent callbacks cannot both pass and double-broadcast.
    with self._persist_state_lock:
        due = (current_time - self._last_update_time) >= update_interval
        if due:
            self._last_update_time = current_time
        phase_active_prev = self._persist_phase_active

    if due:
        self._calculate_and_broadcast_eta(data)

        # Update persistence phase state based on transitions. Decide using
        # the phase snapshot taken above so the branch is consistent.
        if target_changed_in_active_phase:
            self._enter_persist_phase(current_time, "target_change")
        elif phase_active_prev and (not phase_active_now):
            # Active -> inactive transition: schedule a shorter persist.
            self._reset_persist_backoff(current_time, "phase_end")
        elif (not phase_active_prev) and phase_active_now:
            # Inactive -> active transition: start backoff sequence.
            self._enter_persist_phase(current_time, "phase_start")

        self._maybe_persist_history(current_time)

on_printer_send_current_data

on_printer_send_current_data(_data)

Stub: Called when current printer data is sent (required by callback interface).

Source code in octoprint_temp_eta/__init__.py

def on_printer_send_current_data(self, _data):
    """Stub: Called when current printer data is sent (required by callback interface)."""
    return None

on_printer_add_log

on_printer_add_log(_data)

Stub: Called when log entry is added (required by callback interface).

Source code in octoprint_temp_eta/__init__.py

def on_printer_add_log(self, _data):
    """Stub: Called when log entry is added (required by callback interface)."""
    return None

on_printer_add_message

on_printer_add_message(_data)

Stub: Called when printer message is added (required by callback interface).

Source code in octoprint_temp_eta/__init__.py

def on_printer_add_message(self, _data):
    """Stub: Called when printer message is added (required by callback interface)."""
    return None

on_event

on_event(event, _payload)

Handle OctoPrint events to keep UI state consistent.

Clears all ETAs immediately on disconnect or printer errors so the navbar/tab do not keep showing stale countdowns.

Parameters:

Name	Type	Description	Default
event	str	OctoPrint event name	required
payload	dict	Event payload	required

Source code in octoprint_temp_eta/__init__.py

def on_event(self, event, _payload):
    """Handle OctoPrint events to keep UI state consistent.

    Clears all ETAs immediately on disconnect or printer errors so the navbar/tab
    do not keep showing stale countdowns.

    Args:
        event (str): OctoPrint event name
        payload (dict): Event payload
    """
    if event in (
        "Disconnected",
        "Error",
        "Shutdown",
    ):  # clear UI on connection loss
        # Persist what we have before clearing.
        self._persist_current_profile_history()
        self._reset_persist_backoff(time.time(), "disconnect_or_error")
        with self._lock:
            heaters = list(self._temp_history.keys())
            for h in heaters:
                self._temp_history[h].clear()

            for h in list(self._cooldown_history.keys()):
                self._cooldown_history[h].clear()

        self._send_clear_messages(heaters)

    # Reset suppression flag on job lifecycle changes; actual suppression is decided in the temperature callback.
    if event in (
        "PrintStarted",
        "PrintResumed",
        "PrintDone",
        "PrintFailed",
        "PrintCancelled",
    ):
        with self._suppress_lock:
            self._suppressing_due_to_print = False

get_settings_defaults

get_settings_defaults()

Return the default settings for the plugin.

Returns:

Name	Type	Description
dict		Dictionary containing default plugin settings.

Source code in octoprint_temp_eta/__init__.py

def get_settings_defaults(self):
    """Return the default settings for the plugin.

    Returns:
        dict: Dictionary containing default plugin settings.
    """
    return {
        "enabled": True,
        "enable_heating_eta": True,
        "suppress_while_printing": False,
        "show_in_sidebar": True,
        "show_in_navbar": True,
        "show_in_tab": True,
        "show_progress_bars": True,
        "show_historical_graph": True,
        "historical_graph_window_seconds": 180,
        "temp_display": "octoprint",
        "threshold_unit": "octoprint",
        "debug_logging": False,
        "threshold_start": 5.0,
        "algorithm": "linear",
        "update_interval": 1.0,
        "history_size": 60,
        # Persistence (advanced; protects SD cards on long-running phases)
        "persist_backoff_reset_s": 30.0,
        "persist_backoff_initial_s": 60.0,
        "persist_backoff_max_s": 300.0,
        "persist_max_json_bytes": 256 * 1024,
        # Cool Down ETA
        "enable_cooldown_eta": True,
        "cooldown_mode": "threshold",
        "cooldown_target_tool0": 50.0,
        "cooldown_target_bed": 40.0,
        "cooldown_target_chamber": 30.0,
        "cooldown_ambient_temp": None,
        "cooldown_hysteresis_c": 1.0,
        "cooldown_fit_window_seconds": 120,
        # Extended settings: status colors
        "color_mode": "bands",
        "color_heating": "#5cb85c",
        "color_cooling": "#337ab7",
        "color_idle": "#777777",
        # Extended settings: sound alerts
        "sound_enabled": False,
        "sound_target_reached": False,
        "sound_cooldown_finished": False,
        "sound_volume": 0.5,
        "sound_min_interval_s": 10.0,
        # Extended settings: browser notifications (toast)
        "notification_enabled": False,
        "notification_target_reached": False,
        "notification_cooldown_finished": False,
        "notification_timeout_s": 6.0,
        "notification_min_interval_s": 10.0,
        # MQTT settings (publishing via the OctoPrint-MQTT plugin;
        # broker connection settings live in that plugin)
        "mqtt_enabled": False,
        "mqtt_base_topic": "octoprint/temp_eta",
        "mqtt_use_appearance_name": True,
        "mqtt_custom_identifier": "",
        "mqtt_qos": 0,
        "mqtt_retain": False,
        "mqtt_publish_interval": 1.0,
        "mqtt_discovery_enabled": False,
        "mqtt_discovery_prefix": "homeassistant",
    }

get_settings_version

get_settings_version()

Return the settings format version for migration handling.

Source code in octoprint_temp_eta/__init__.py

def get_settings_version(self) -> int:
    """Return the settings format version for migration handling."""
    return 2

on_settings_migrate

on_settings_migrate(target, current)

Migrate stored plugin settings to the current format.

Version 2 delegates the MQTT broker connection to the OctoPrint-MQTT plugin; the plugin's own broker settings (including the stored password) are removed from config.yaml.

Source code in octoprint_temp_eta/__init__.py

def on_settings_migrate(self, target: int, current: Optional[int]) -> None:
    """Migrate stored plugin settings to the current format.

    Version 2 delegates the MQTT broker connection to the OctoPrint-MQTT
    plugin; the plugin's own broker settings (including the stored
    password) are removed from config.yaml.
    """
    if current is None or current < 2:
        remove = getattr(self._settings, "remove", None)
        for key in (
            "mqtt_broker_host",
            "mqtt_broker_port",
            "mqtt_username",
            "mqtt_password",
            "mqtt_use_tls",
            "mqtt_tls_insecure",
        ):
            if callable(remove):
                try:
                    remove([key])
                    continue
                except _EXPECTED_ERRORS:
                    pass
            self._settings.set([key], None)
        self._logger.info(
            "Migrated MQTT settings: broker configuration is now handled "
            "by the OctoPrint-MQTT plugin"
        )

is_template_autoescaped

is_template_autoescaped()

Enable autoescaping for all plugin templates.

Opt-in to OctoPrint's template autoescaping (OctoPrint 1.11+) to reduce XSS risk from unescaped injected variables.

Returns:

Name	Type	Description
bool	bool	True to enable autoescaping.

Source code in octoprint_temp_eta/__init__.py

def is_template_autoescaped(self) -> bool:  # pyright: ignore
    """Enable autoescaping for all plugin templates.

    Opt-in to OctoPrint's template autoescaping (OctoPrint 1.11+) to reduce
    XSS risk from unescaped injected variables.

    Returns:
        bool: True to enable autoescaping.
    """
    return True

get_template_configs

get_template_configs()

Configure which templates to use and how to bind them.

Returns:

Name	Type	Description
list		List of template configuration dictionaries.

Source code in octoprint_temp_eta/__init__.py

def get_template_configs(self):
    """Configure which templates to use and how to bind them.

    Returns:
        list: List of template configuration dictionaries.
    """
    return [
        {"type": "navbar", "custom_bindings": True},
        {
            "type": "sidebar",
            "custom_bindings": True,
            "name": gettext("Temperature ETA"),
            "icon": "fa fa-clock",
        },
        # Use OctoPrint's default settingsViewModel binding for settings UI.
        {"type": "settings", "custom_bindings": True},
        {"type": "tab", "custom_bindings": False},
    ]

on_settings_save

on_settings_save(data)

Persist settings and clear UI/state when disabling the plugin.

OctoPrint applies settings changes only on save. When the plugin is disabled, we actively clear any previously shown countdowns in navbar/tab so the UI does not keep showing stale values.

Parameters:

Name	Type	Description	Default
data	dict	Settings data posted from the UI.	required

Source code in octoprint_temp_eta/__init__.py

def on_settings_save(self, data: dict[str, Any]) -> dict[str, Any]:
    """Persist settings and clear UI/state when disabling the plugin.

    OctoPrint applies settings changes only on save. When the plugin is disabled,
    we actively clear any previously shown countdowns in navbar/tab so the UI
    does not keep showing stale values.

    Args:
        data (dict): Settings data posted from the UI.
    """
    if not getattr(self, "_settings", None):
        return {}

    self._sanitize_settings_payload(data)

    was_enabled = bool(self._settings.get_boolean(["enabled"]))
    old_debug = bool(getattr(self, "_debug_logging_enabled", False))
    old_history_maxlen = self._read_history_maxlen_setting()
    old_mqtt_identity = self._read_mqtt_identity()
    saved = octoprint.plugin.SettingsPlugin.on_settings_save(self, data)
    is_enabled = bool(self._settings.get_boolean(["enabled"]))

    self._refresh_debug_logging_flag()
    self._refresh_runtime_caches()
    if old_debug != bool(self._debug_logging_enabled):
        self._logger.info(
            "Debug logging %s",
            "enabled" if self._debug_logging_enabled else "disabled",
        )

    new_history_maxlen = self._read_history_maxlen_setting()
    if new_history_maxlen != old_history_maxlen:
        self._set_history_maxlen(new_history_maxlen)

    if was_enabled and not is_enabled:
        self._clear_all_heaters_frontend()

    # Reconfigure the MQTT publisher with new settings
    self._configure_mqtt_publisher()
    self._sync_mqtt_discovery(old_mqtt_identity)

    return saved if isinstance(saved, dict) else {}

get_assets

get_assets()

Return static assets (JS, CSS, LESS) to be included.

Returns:

Name	Type	Description
dict		Dictionary with asset types and their file paths.

Source code in octoprint_temp_eta/__init__.py

def get_assets(self):
    """Return static assets (JS, CSS, LESS) to be included.

    Returns:
        dict: Dictionary with asset types and their file paths.
    """
    return {
        "js": ["js/temp_eta.js"],
        # Ship pre-compiled CSS rather than LESS: OctoPrint only compiles
        # LESS when a server-side compiler (lesscpy / Node lessc) is present,
        # which is not guaranteed on a pip-installed plugin. Without it the
        # LESS is silently dropped from the bundle, so none of the plugin's
        # styling (graph colours included) reaches the page. The .less stays
        # in the source tree as the editable origin; regenerate temp_eta.css
        # with `npx less less/temp_eta.less css/temp_eta.css` after editing.
        "css": ["css/temp_eta.css"],
    }

is_api_protected

is_api_protected()

Whether the Simple API requires an authenticated user.

OctoPrint's default for this will switch from False to True in the future. We explicitly opt in to avoid relying on defaults.

Source code in octoprint_temp_eta/__init__.py

def is_api_protected(self) -> bool:  # type: ignore[override]
    """Whether the Simple API requires an authenticated user.

    OctoPrint's default for this will switch from False to True in the
    future. We explicitly opt in to avoid relying on defaults.
    """
    return True

is_api_adminonly

is_api_adminonly()

Whether the Simple API is restricted to admin users.

Source code in octoprint_temp_eta/__init__.py

def is_api_adminonly(self) -> bool:  # type: ignore[override]
    """Whether the Simple API is restricted to admin users."""
    return True

get_api_commands

get_api_commands()

Return supported Simple API commands for the plugin.

Source code in octoprint_temp_eta/__init__.py

def get_api_commands(self) -> dict[str, list]:  # type: ignore[override]
    """Return supported Simple API commands for the plugin."""
    return {
        "reset_profile_history": [],
        "reset_settings_defaults": [],
    }

on_api_get

on_api_get(_request)

Return plugin status for the Simple API GET endpoint.

Exposes whether the OctoPrint-MQTT plugin (the delegation target for all MQTT publishing) is available and whether MQTT publishing is enabled, so the settings UI can warn when the MQTT plugin is missing. The broker connection itself is owned by the OctoPrint-MQTT plugin.

Source code in octoprint_temp_eta/__init__.py

def on_api_get(self, _request: Any):  # type: ignore[override]
    """Return plugin status for the Simple API GET endpoint.

    Exposes whether the OctoPrint-MQTT plugin (the delegation target for
    all MQTT publishing) is available and whether MQTT publishing is
    enabled, so the settings UI can warn when the MQTT plugin is missing.
    The broker connection itself is owned by the OctoPrint-MQTT plugin.
    """
    publisher = self._mqtt_publisher
    return jsonify(
        {
            "mqtt_available": bool(
                publisher is not None and publisher.is_mqtt_plugin_available()
            ),
            "mqtt_enabled": bool(self._settings.get_boolean(["mqtt_enabled"])),
        }
    )

on_api_command

on_api_command(command, _data)

Handle Simple API commands.

Source code in octoprint_temp_eta/__init__.py

def on_api_command(self, command: str, _data: dict[str, Any]):  # type: ignore[override]
    """Handle Simple API commands."""
    if command == "reset_profile_history":
        deleted_count = self._reset_all_profile_histories()
        profile_id = self._get_current_profile_id()
        logger = getattr(self, "_logger", None)
        if logger is not None:
            logger.info(
                "Reset persisted history for all profiles (trigger_profile=%s deleted_files=%d)",
                str(profile_id),
                int(deleted_count),
            )
        return jsonify(
            {
                "success": True,
                "profile_id": profile_id,
                "deleted_files": deleted_count,
            }
        )

    if command == "reset_settings_defaults":
        self._reset_user_settings_to_defaults()
        logger = getattr(self, "_logger", None)
        if logger is not None:
            logger.info("Restored plugin settings defaults (user-editable keys)")

        # Notify all connected clients so the settings UI can refresh via requestData().
        if getattr(self, "_plugin_manager", None):
            try:
                self._plugin_manager.send_plugin_message(
                    self._identifier, {"type": "settings_reset"}
                )
            except _EXPECTED_ERRORS:
                pass

        return jsonify({"success": True, "message": gettext("Defaults restored.")})

    return jsonify({"success": False, "error": "unknown_command"})

get_update_information

get_update_information()

Provide update information for the Software Update plugin.

Returns:

Name	Type	Description
dict		Update configuration for the plugin.

Source code in octoprint_temp_eta/__init__.py

def get_update_information(self):
    """Provide update information for the Software Update plugin.

    Returns:
        dict: Update configuration for the plugin.
    """
    return {
        "temp_eta": {
            "displayName": "Temperature ETA Plugin",
            "displayVersion": self._plugin_version,
            # version check: github repository
            "type": "github_release",
            "user": "Ajimaru",
            "repo": "OctoPrint-TempETA",
            "current": self._plugin_version,
            # update method: pip
            "pip": "https://github.com/Ajimaru/OctoPrint-TempETA/archive/{target_version}.zip",
        }
    }

Calculator Module

Temperature ETA calculation algorithms.

This module provides pure calculation logic for estimating time remaining until a printer heater reaches its target temperature or cools down.

All functions are stateless and independent of OctoPrint plugin mechanics, making them easy to test and maintain.

calculate_linear_eta

calculate_linear_eta(history, target, window_seconds=10.0)

Calculate ETA assuming constant heating rate.

Estimates the heating rate from the first and last sample within the recent time window and extrapolates linearly to the target.

Parameters:

Name	Type	Description	Default
history	deque	Deque of (timestamp, actual_temp, target_temp) tuples	required
target	float	Target temperature in degrees	required
window_seconds	float	Time window for rate calculation (default: 10s)	10.0

Returns:

Type	Description
Optional[float]	Estimated seconds to target, or None if insufficient data
Optional[float]	or the heater is not heating within the window

Source code in octoprint_temp_eta/calculator.py

def calculate_linear_eta(
    history: deque, target: float, window_seconds: float = 10.0
) -> Optional[float]:
    """
    Calculate ETA assuming constant heating rate.

    Estimates the heating rate from the first and last sample within the
    recent time window and extrapolates linearly to the target.

    Args:
        history: Deque of (timestamp, actual_temp, target_temp) tuples
        target: Target temperature in degrees
        window_seconds: Time window for rate calculation (default: 10s)

    Returns:
        Estimated seconds to target, or None if insufficient data
        or the heater is not heating within the window
    """
    if not _validate_scalar(target) or not _validate_window(window_seconds):
        return None
    if not history or len(history) < 2:
        return None

    last_ts = _find_last_ts(history)
    if last_ts is None:
        return None

    recent = _filter_recent(history, last_ts - window_seconds)
    if len(recent) < 2:
        return None

    t0, temp0 = recent[0][0], recent[0][1]
    t1, temp1 = recent[-1][0], recent[-1][1]
    time_diff = t1 - t0
    temp_diff = temp1 - temp0
    if time_diff <= 0 or temp_diff <= 0:
        return None

    remaining = target - temp1
    if remaining <= 0:
        return None

    return max(0.0, remaining / (temp_diff / time_diff))

calculate_exponential_eta

calculate_exponential_eta(history, target, window_seconds=30.0)

Calculate ETA accounting for thermal asymptotic behavior.

Uses exponential model: T(t) = T_final - (T_final - T_0) * e^(-t/tau) Falls back to linear estimation when insufficient data or poor fit.

Parameters:

Name	Type	Description	Default
history	deque	Deque of (timestamp, actual_temp, target_temp) tuples	required
target	float	Target temperature in degrees	required
window_seconds	float	Time window for exponential fit (default: 30s)	30.0

Returns:

Type	Description
Optional[float]	Estimated seconds to target, or None if insufficient data

Source code in octoprint_temp_eta/calculator.py

def calculate_exponential_eta(
    history: deque, target: float, window_seconds: float = 30.0
) -> Optional[float]:
    """
    Calculate ETA accounting for thermal asymptotic behavior.

    Uses exponential model: T(t) = T_final - (T_final - T_0) * e^(-t/tau)
    Falls back to linear estimation when insufficient data or poor fit.

    Args:
        history: Deque of (timestamp, actual_temp, target_temp) tuples
        target: Target temperature in degrees
        window_seconds: Time window for exponential fit (default: 30s)

    Returns:
        Estimated seconds to target, or None if insufficient data
    """
    if not _validate_scalar(target) or not _validate_window(window_seconds):
        return None
    if not history or len(history) < 3:
        return None

    last_ts = _find_last_ts(history)
    if last_ts is None:
        return None

    recent = _dedupe_by_ts(_filter_recent(history, last_ts - window_seconds))
    if len(recent) < 6:
        return calculate_linear_eta(history, target)

    temp_now = recent[-1][1]
    remaining_now = target - temp_now
    if remaining_now <= 0:
        return None

    # Guard: window must show meaningful heating; otherwise no ETA is possible.
    if (recent[-1][1] - recent[0][1]) <= 0.2:
        return None

    epsilon_c = 0.5
    if remaining_now <= epsilon_c:
        return 0.0

    try:
        eta = _exponential_fit(recent, target, epsilon_c)
    except (ValueError, ArithmeticError) as exc:
        _LOG.debug("Exponential ETA math error: %s", exc)
        return calculate_linear_eta(history, target)

    if eta is None:
        return calculate_linear_eta(history, target)

    linear_eta = calculate_linear_eta(history, target)
    if linear_eta is not None and eta > linear_eta * 5:
        return linear_eta

    return eta

calculate_cooldown_linear_eta

calculate_cooldown_linear_eta(cooldown_history, goal_c, window_seconds=60.0)

Linear cooldown ETA from recent slope.

Estimates the cooling rate from the first and last sample within the recent time window and extrapolates linearly to the goal. The result is capped at 24 hours.

Parameters:

Name	Type	Description	Default
cooldown_history	deque	Deque of (timestamp, temp) tuples	required
goal_c	float	Target cooldown temperature in degrees	required
window_seconds	float	Time window for fit (default: 60s)	60.0

Returns:

Type	Description
Optional[float]	Estimated seconds to goal, or None if insufficient data
Optional[float]	or the heater is not cooling within the window

Source code in octoprint_temp_eta/calculator.py

def calculate_cooldown_linear_eta(
    cooldown_history: deque, goal_c: float, window_seconds: float = 60.0
) -> Optional[float]:
    """
    Linear cooldown ETA from recent slope.

    Estimates the cooling rate from the first and last sample within the
    recent time window and extrapolates linearly to the goal. The result
    is capped at 24 hours.

    Args:
        cooldown_history: Deque of (timestamp, temp) tuples
        goal_c: Target cooldown temperature in degrees
        window_seconds: Time window for fit (default: 60s)

    Returns:
        Estimated seconds to goal, or None if insufficient data
        or the heater is not cooling within the window
    """
    if not _validate_scalar(goal_c) or not _validate_window(window_seconds):
        return None
    if not cooldown_history:
        return None

    last_ts = _find_last_ts(cooldown_history)
    if last_ts is None:
        return None

    recent = _filter_recent(cooldown_history, last_ts - window_seconds)
    if len(recent) < 2:
        return None

    t0, temp0 = recent[0]
    t1, temp1 = recent[-1]
    dt = t1 - t0
    if dt <= 0:
        return None

    slope = (temp1 - temp0) / dt
    if slope >= -1e-3:
        return None

    remaining = temp1 - goal_c
    if remaining <= 0:
        return None

    eta = remaining / (-slope)
    if not math.isfinite(eta) or eta < 0:
        return None

    return float(min(eta, 24 * 3600))

calculate_cooldown_exponential_eta

calculate_cooldown_exponential_eta(cooldown_history, ambient_c, goal_c, window_seconds=60.0)

Exponential cooldown ETA (Newton's law of cooling).

Models cooldown as: T(t) = T_ambient + (T_0 - T_ambient) * e^(-t/tau)

Source code in octoprint_temp_eta/calculator.py

def calculate_cooldown_exponential_eta(
    cooldown_history: deque,
    ambient_c: float,
    goal_c: float,
    window_seconds: float = 60.0,
) -> Optional[float]:
    """
    Exponential cooldown ETA (Newton's law of cooling).

    Models cooldown as: T(t) = T_ambient + (T_0 - T_ambient) * e^(-t/tau)
    """
    if not (_validate_scalar(ambient_c) and _validate_scalar(goal_c)):
        return None
    if not _validate_window(window_seconds):
        return None
    if goal_c <= ambient_c:
        return None
    if not cooldown_history or len(cooldown_history) < 4:
        return None

    last_ts = _find_last_ts(cooldown_history)
    if last_ts is None:
        return None

    recent = _filter_recent(cooldown_history, last_ts - window_seconds)
    if len(recent) < 6:
        return calculate_cooldown_linear_eta(cooldown_history, goal_c, window_seconds)

    if recent[-1][1] <= goal_c:
        return None

    try:
        return _cooldown_exponential_fit(recent, ambient_c, goal_c, epsilon=0.5)
    except (ValueError, ArithmeticError) as exc:
        _LOG.debug("Exponential ETA math error: %s", exc)
        return None

MQTT Publisher Module

Thread-safe MQTT publisher backed by the OctoPrint-MQTT plugin helper.

Keeps the exact topic scheme and payload format of the previous built-in client ({base}/{heater}/eta and {base}/{heater}/state_change), including per-heater publish throttling and state transition events.

Initialize the publisher.

Parameters:

Name	Type	Description	Default
logger	Any	Logger instance for debug/info messages	required
plugin_version	str	Plugin version, used in discovery device info	''

Source code in octoprint_temp_eta/mqtt_publisher.py

def __init__(self, logger: Any, plugin_version: str = ""):
    """Initialize the publisher.

    Args:
        logger: Logger instance for debug/info messages
        plugin_version: Plugin version, used in discovery device info
    """
    self._logger = logger
    self._plugin_version = plugin_version
    self._lock = threading.Lock()

    # mqtt_publish helper from the OctoPrint-MQTT plugin, acquired in
    # initialize(); None while the MQTT plugin is not available.
    self._mqtt_publish: Optional[Callable[..., Any]] = None

    self._enabled = False

    # Publishing settings
    self._base_topic = "octoprint/temp_eta"
    self._qos = 0
    self._retain = False
    self._publish_interval = 1.0

    # Home Assistant autodiscovery settings
    self._discovery_enabled = False
    self._discovery_prefix = "homeassistant"
    self._instance_slug = ""

    # State tracking for state transition events.
    # Publish throttling is tracked per heater: a shared timestamp would let
    # the first heater in every broadcast batch consume the publish slot and
    # permanently starve the others.
    self._last_published_time: dict[str, float] = {}
    self._last_heater_state: dict[str, Optional[str]] = {}

    # Identity of the last published discovery configs, so retained
    # topics can be cleared even after the identity changed:
    # (discovery_prefix, instance_slug, heaters tuple)
    self._last_discovery_identity: Optional[tuple[str, str, tuple[str, ...]]] = None

initialize

initialize(plugin_manager)

Acquire the mqtt_publish helper from the OctoPrint-MQTT plugin.

Call from on_after_startup (all plugins are loaded by then).

Parameters:

Name	Type	Description	Default
plugin_manager	Any	OctoPrint plugin manager instance	required

Source code in octoprint_temp_eta/mqtt_publisher.py

def initialize(self, plugin_manager: Any) -> None:
    """Acquire the mqtt_publish helper from the OctoPrint-MQTT plugin.

    Call from ``on_after_startup`` (all plugins are loaded by then).

    Args:
        plugin_manager: OctoPrint plugin manager instance
    """
    helper = self._acquire_helper(plugin_manager)
    with self._lock:
        self._mqtt_publish = helper
    if helper is not None:
        self._logger.info("MQTT plugin helper found, MQTT publishing available")
    else:
        self._logger.info(
            "OctoPrint-MQTT plugin not available, MQTT publishing disabled. "
            "Install and enable the OctoPrint-MQTT plugin to use MQTT features."
        )

is_mqtt_plugin_available

is_mqtt_plugin_available()

Check whether the OctoPrint-MQTT publish helper was acquired.

Returns:

Name	Type	Description
bool	bool	True if the OctoPrint-MQTT plugin is installed and enabled

Source code in octoprint_temp_eta/mqtt_publisher.py

def is_mqtt_plugin_available(self) -> bool:
    """Check whether the OctoPrint-MQTT publish helper was acquired.

    Returns:
        bool: True if the OctoPrint-MQTT plugin is installed and enabled
    """
    with self._lock:
        return self._mqtt_publish is not None

is_operational

is_operational()

Check whether publishing is enabled and the helper is available.

Source code in octoprint_temp_eta/mqtt_publisher.py

def is_operational(self) -> bool:
    """Check whether publishing is enabled and the helper is available."""
    with self._lock:
        return self._mqtt_publish is not None and self._enabled

configure

configure(settings)

Update MQTT configuration from plugin settings.

Parameters:

Name	Type	Description	Default
settings	dict[str, Any]	Dictionary with MQTT configuration keys	required

Source code in octoprint_temp_eta/mqtt_publisher.py

def configure(self, settings: dict[str, Any]) -> None:
    """Update MQTT configuration from plugin settings.

    Args:
        settings: Dictionary with MQTT configuration keys
    """
    with self._lock:
        self._enabled = bool(settings.get("mqtt_enabled", False))

        base_topic = str(
            settings.get("mqtt_base_topic", "octoprint/temp_eta")
        ).strip()
        use_appearance_name = bool(settings.get("mqtt_use_appearance_name", True))
        appearance_name = str(settings.get("mqtt_appearance_name") or "").strip()
        custom_identifier = str(settings.get("mqtt_custom_identifier", "")).strip()

        # Build final topic with optional identifier suffix
        self._base_topic = self._build_final_topic(
            base_topic, use_appearance_name, appearance_name, custom_identifier
        )
        self._instance_slug = self._build_instance_slug(
            use_appearance_name, appearance_name, custom_identifier
        )

        self._qos = self._coerce_int(
            settings.get("mqtt_qos"), default=0, lo=0, hi=2
        )
        self._retain = bool(settings.get("mqtt_retain", False))
        self._publish_interval = self._coerce_float(
            settings.get("mqtt_publish_interval"), default=1.0, lo=0.0, hi=3600.0
        )

        self._discovery_enabled = bool(
            settings.get("mqtt_discovery_enabled", False)
        )
        self._discovery_prefix = (
            str(settings.get("mqtt_discovery_prefix") or "homeassistant")
            .strip()
            .strip("/")
        ) or "homeassistant"

publish_eta_update

publish_eta_update(heater, eta, eta_kind, target, actual, cooldown_target=None)

Publish ETA update for a heater.

Parameters:

Name	Type	Description	Default
heater	str	Heater name (bed, tool0, chamber)	required
eta	Optional[float]	ETA in seconds, or None	required
eta_kind	Optional[str]	"heating", "cooling", or None	required
target	Optional[float]	Target temperature	required
actual	Optional[float]	Actual temperature	required
cooldown_target	Optional[float]	Cooldown target temperature (if applicable)	None

Source code in octoprint_temp_eta/mqtt_publisher.py

def publish_eta_update(
    self,
    heater: str,
    eta: Optional[float],
    eta_kind: Optional[str],
    target: Optional[float],
    actual: Optional[float],
    cooldown_target: Optional[float] = None,
) -> None:
    """Publish ETA update for a heater.

    Args:
        heater: Heater name (bed, tool0, chamber)
        eta: ETA in seconds, or None
        eta_kind: "heating", "cooling", or None
        target: Target temperature
        actual: Actual temperature
        cooldown_target: Cooldown target temperature (if applicable)
    """
    with self._lock:
        if not self._enabled or self._mqtt_publish is None:
            return

        # Check if we should publish based on the per-heater interval
        now = time.time()
        if (
            now - self._last_published_time.get(heater, 0.0)
        ) < self._publish_interval:
            return

        self._last_published_time[heater] = now

        # Determine state for transition detection
        current_state = None
        if eta_kind == "heating" and eta is not None:
            current_state = "heating"
        elif eta_kind == "cooling" and eta is not None:
            current_state = "cooling"
        elif target is not None and actual is not None:
            if abs(target - actual) <= 1.0:
                current_state = "at_target"
            elif cooldown_target is not None and actual is not None:
                if abs(cooldown_target - actual) <= 1.0:
                    current_state = "cooled_down"

        # Detect state transitions
        last_state = self._last_heater_state.get(heater)
        state_changed = last_state != current_state
        self._last_heater_state[heater] = current_state

        # Build payload
        payload = {
            "heater": heater,
            "eta_seconds": eta,
            "eta_kind": eta_kind,
            "target": target,
            "actual": actual,
            "cooldown_target": cooldown_target,
            "timestamp": now,
            "state": current_state,
        }

        # Publish ETA data
        topic = f"{self._base_topic}/{heater}/eta"
        self._publish_message(topic, payload)

        # Publish state transition event if state changed
        if state_changed and current_state is not None:
            event_payload = {
                "heater": heater,
                "state": current_state,
                "previous_state": last_state,
                "timestamp": now,
                "actual": actual,
                "target": target,
            }
            event_topic = f"{self._base_topic}/{heater}/state_change"
            self._publish_message(event_topic, event_payload)

            self._logger.info(
                "MQTT: %s state changed from %s to %s",
                heater,
                last_state or "unknown",
                current_state,
            )

publish_discovery

publish_discovery(heaters)

Publish retained Home Assistant autodiscovery configs.

Parameters:

Name	Type	Description	Default
heaters	list[str]	Heater names to publish sensors for (e.g. tool0, bed)	required

Source code in octoprint_temp_eta/mqtt_publisher.py

def publish_discovery(self, heaters: list[str]) -> None:
    """Publish retained Home Assistant autodiscovery configs.

    Args:
        heaters: Heater names to publish sensors for (e.g. tool0, bed)
    """
    with self._lock:
        if (
            self._mqtt_publish is None
            or not self._enabled
            or not self._discovery_enabled
            or not heaters
        ):
            return

        for heater in heaters:
            for object_id in self.SENSOR_DEFINITIONS:
                topic = self._discovery_topic(
                    self._discovery_prefix, self._instance_slug, heater, object_id
                )
                payload = self._build_discovery_payload(heater, object_id)
                # Discovery configs are always retained, otherwise Home
                # Assistant loses them on restart.
                self._publish_message(topic, payload, retained=True)

        self._last_discovery_identity = (
            self._discovery_prefix,
            self._instance_slug,
            tuple(heaters),
        )
        self._logger.info(
            "Published MQTT discovery configs for %d heater(s) as instance '%s'",
            len(heaters),
            self._instance_slug,
        )

clear_retained_topics

clear_retained_topics()

Clear previously published retained discovery configs.

Publishes empty retained payloads (Home Assistant convention for removing discovered entities). Uses the identity of the last published discovery configs so the old topics are cleared even when prefix/identifier changed in the same settings save.

Source code in octoprint_temp_eta/mqtt_publisher.py

def clear_retained_topics(self) -> None:
    """Clear previously published retained discovery configs.

    Publishes empty retained payloads (Home Assistant convention for
    removing discovered entities). Uses the identity of the last
    published discovery configs so the old topics are cleared even when
    prefix/identifier changed in the same settings save.
    """
    with self._lock:
        if self._mqtt_publish is None or self._last_discovery_identity is None:
            return

        prefix, instance_slug, heaters = self._last_discovery_identity
        for heater in heaters:
            for object_id in self.SENSOR_DEFINITIONS:
                topic = self._discovery_topic(
                    prefix, instance_slug, heater, object_id
                )
                self._publish_message(topic, "", retained=True)

        self._last_discovery_identity = None
        self._logger.info(
            "Cleared retained MQTT discovery topics for instance '%s'",
            instance_slug,
        )

Usage Examples

Using the Calculator

from octoprint_temp_eta.calculator import calculate_linear_eta, calculate_exponential_eta
from collections import deque
import time

# Create temperature history
history = deque()
for i in range(10):
    timestamp = time.time() + i
    temperature = 25 + i * 0.2  # small ramp
    target = 200.0
    history.append((timestamp, temperature, target))

# Calculate ETA using linear estimator
eta_seconds = calculate_linear_eta(history, target)
print(f"Linear ETA: {eta_seconds}")

# Calculate ETA using exponential estimator (fallbacks to linear if needed)
eta_exp = calculate_exponential_eta(history, target)
print(f"Exponential ETA: {eta_exp}")

Using the MQTT Publisher

The publisher does not open its own broker connection — it delegates every publish to the OctoPrint-MQTT plugin's mqtt_publish helper:

from octoprint_temp_eta.mqtt_publisher import MqttPublisher
import logging

# Create logger
logger = logging.getLogger(__name__)

# Instantiate publisher (logger plus plugin version for HA discovery info)
publisher = MqttPublisher(logger, plugin_version="1.0.0")

# Acquire the mqtt_publish helper from the OctoPrint-MQTT plugin
# (inside a plugin this is self._plugin_manager, in on_after_startup)
publisher.initialize(plugin_manager)

# Configure via settings-like dict
publisher.configure({
    "mqtt_enabled": True,
    "mqtt_base_topic": "octoprint/temp_eta",
    "mqtt_qos": 0,
    "mqtt_retain": False,
    "mqtt_publish_interval": 1.0,
    "mqtt_discovery_enabled": False,
    "mqtt_discovery_prefix": "homeassistant",
})

# Publish a sample ETA update (heater name, eta seconds, eta kind, target, actual)
publisher.publish_eta_update(
    heater="tool0",
    eta=120.0,
    eta_kind="heating",
    target=200.0,
    actual=50.0,
)

# Optional: publish/clear retained Home Assistant discovery configs
publisher.publish_discovery(["tool0", "bed"])
publisher.clear_retained_topics()

Consuming ETA data from another plugin or client

The plugin does not expose a Python getter for current ETAs. Consume the data through one of its public channels instead:

• Plugin messages: every update is broadcast via send_plugin_message("temp_eta", payload); frontend components can listen in onDataUpdaterPluginMessage (see the JavaScript API).
• MQTT: enable the MQTT integration and subscribe to {active_topic}/{heater}/eta and {active_topic}/{heater}/state_change (see the README for payload formats).
• Simple API: GET /api/plugin/temp_eta returns the MQTT integration status (mqtt_available = OctoPrint-MQTT plugin present, mqtt_enabled); POST supports the reset_profile_history and reset_settings_defaults commands (admin only).

Threading Considerations

The calculator functions are pure and stateless — they read the history they are given and share no module state, so they are safe to call from any thread as long as the caller does not mutate the passed history concurrently. The plugin itself guards its history deques with an internal lock, and MqttPublisher serializes all its state behind its own lock.

Error Handling

The estimators signal "no trustworthy estimate" by returning None rather than raising:

eta = calculate_linear_eta(history, target)
if eta is None:
    print("Insufficient data for ETA calculation")
else:
    print(f"ETA: {eta:.1f}s")

Invalid inputs (NaN/infinite targets, non-positive windows, malformed samples) are filtered or rejected the same way; internal math errors in the exponential fit degrade to the linear estimate instead of propagating.

Type Hints

The public estimator signatures (from octoprint_temp_eta/calculator.py):

from collections import deque
from typing import Optional

def calculate_linear_eta(
    history: deque, target: float, window_seconds: float = 10.0
) -> Optional[float]: ...

def calculate_exponential_eta(
    history: deque, target: float, window_seconds: float = 30.0
) -> Optional[float]: ...

def calculate_cooldown_linear_eta(
    cooldown_history: deque, goal_c: float, window_seconds: float = 60.0
) -> Optional[float]: ...

def calculate_cooldown_exponential_eta(
    cooldown_history: deque,
    ambient_c: float,
    goal_c: float,
    window_seconds: float = 60.0,
) -> Optional[float]: ...

Heating histories hold (timestamp, actual_temp, target_temp) tuples; cooldown histories hold (timestamp, actual_temp) tuples.

Logging

The plugin logs under the octoprint_temp_eta logger. Enable the Debug logging setting to get verbose [debug]-prefixed messages (emitted at info level so they show up without reconfiguring OctoPrint's logging).

Next Steps

• JavaScript API - Frontend API reference
• Algorithms - ETA calculation details
• Testing - Unit tests and examples