"""Global configuration, gauge registry, and runtime state.""" from __future__ import annotations import math import time from dataclasses import dataclass, field, replace from enum import IntFlag, auto from typing import Any from .animation import AnimationState, start_animation from .exceptions import GaugeConfigError, GaugeNotFoundError from .gauges import CompassGaugeConfig, GaugeConfig from .scales import clamp_value, compass_heading, validate_markers, validate_thresholds from .sizing import RequestedSize, SizeState from .smoothing import SmoothingState from .themes import GaugeTheme, get_theme from .types import AnimationConfig, GaugeMarker, GaugeValue, SmoothingConfig, Tag, ThresholdBand class DirtyFlags(IntFlag): NONE = 0 VALUE = auto() STATIC = auto() DYNAMIC = auto() TEXT = auto() SIZE = auto() STYLE = auto() DEBUG = auto() FULL = VALUE | STATIC | DYNAMIC | TEXT | SIZE | STYLE | DEBUG @dataclass class DpgGaugesConfig: default_theme: str | GaugeTheme | None = None animation: AnimationConfig | bool | None = None smoothing: SmoothingConfig | bool | None = None default_width: int | None = None default_height: int | None = None frame_rate_limit: float | None = None debug: bool = False def __post_init__(self) -> None: if isinstance(self.animation, bool): self.animation = AnimationConfig(enabled=self.animation) if isinstance(self.smoothing, bool): self.smoothing = SmoothingConfig(enabled=self.smoothing) if self.default_theme is not None: self.default_theme = get_theme(self.default_theme) if self.default_width is not None and self.default_width < -1: msg = "default_width must be -1, 0, or positive" raise GaugeConfigError(msg) if self.default_height is not None and self.default_height < -1: msg = "default_height must be -1, 0, or positive" raise GaugeConfigError(msg) if self.frame_rate_limit is not None and self.frame_rate_limit <= 0: msg = "frame_rate_limit must be greater than zero" raise GaugeConfigError(msg) @dataclass class GaugeState: tag: Tag gauge_type: str container_tag: Tag drawlist_tag: Tag | None text_tags: dict[str, Tag] config: GaugeConfig target_value: GaugeValue clamped_value: float | None displayed_value: float | None previous_displayed_value: float | None value_revision: int = 0 config_revision: int = 0 measured_width: int = 0 measured_height: int = 0 last_nonzero_width: int = 0 last_nonzero_height: int = 0 dirty: DirtyFlags = DirtyFlags.FULL renderer: Any = None animation_state: AnimationState = field(default_factory=AnimationState) smoothing_state: SmoothingState = field(default_factory=SmoothingState) invalid_value: bool = False _config = DpgGaugesConfig() _registry: dict[Tag, GaugeState] = {} _next_auto_tag = 1 def configure_gauges(**config: Any) -> DpgGaugesConfig: global _config _config = DpgGaugesConfig(**config) return _config def get_config() -> DpgGaugesConfig: return _config def clear_registry() -> None: global _next_auto_tag _registry.clear() _next_auto_tag = 1 def registry_size() -> int: return len(_registry) def resolve_tag(tag: Tag | None) -> Tag: global _next_auto_tag if tag is not None: return tag tag = f"dpg_gauge_{_next_auto_tag}" _next_auto_tag += 1 return tag def apply_global_defaults(config: GaugeConfig) -> GaugeConfig: updates: dict[str, Any] = {} if config.theme is None and _config.default_theme is not None: updates["theme"] = _config.default_theme if config.animation is None and _config.animation is not None: updates["animation"] = _config.animation if config.smoothing is None and _config.smoothing is not None: updates["smoothing"] = _config.smoothing if config.width == 0 and _config.default_width is not None: updates["width"] = _config.default_width if config.height == 0 and _config.default_height is not None: updates["height"] = _config.default_height return replace(config, **updates) if updates else config def coerce_numeric_value(value: GaugeValue) -> float | None: if isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(value): return float(value) return None def animation_config(config: GaugeConfig) -> AnimationConfig | None: animation = config.animation if isinstance(animation, bool): return AnimationConfig(enabled=animation) return animation def smoothing_config(config: GaugeConfig) -> SmoothingConfig | None: smoothing = config.smoothing if isinstance(smoothing, bool): return SmoothingConfig(enabled=smoothing) return smoothing def clamp_runtime_value(value: GaugeValue, config: GaugeConfig) -> tuple[float | None, bool]: numeric = coerce_numeric_value(value) if numeric is None: return None, value is not None if not math.isfinite(numeric): return None, True if isinstance(config, CompassGaugeConfig): numeric = compass_heading(numeric) if config.clamp: return clamp_value(numeric, config.min_value, config.max_value), False return numeric, False def register_gauge( gauge_type: str, config: GaugeConfig, *, container_tag: Tag | None = None, drawlist_tag: Tag | None = None, text_tags: dict[str, Tag] | None = None, ) -> GaugeState: config = apply_global_defaults(config) tag = resolve_tag(config.tag) if tag in _registry: msg = f"gauge already exists: {tag!r}" raise GaugeConfigError(msg) if config.tag != tag: config = replace(config, tag=tag) clamped, invalid = clamp_runtime_value(config.value, config) state = GaugeState( tag=tag, gauge_type=gauge_type, container_tag=container_tag or tag, drawlist_tag=drawlist_tag, text_tags=text_tags or {}, config=config, target_value=config.value, clamped_value=clamped, displayed_value=clamped, previous_displayed_value=clamped, measured_width=config.width if config.width > 0 else 0, measured_height=config.height if config.height > 0 else 0, last_nonzero_width=config.width if config.width > 0 else 0, last_nonzero_height=config.height if config.height > 0 else 0, invalid_value=invalid, ) state.smoothing_state.value = clamped _registry[tag] = state return state def get_gauge_state(tag: Tag) -> GaugeState: try: return _registry[tag] except KeyError as exc: msg = f"gauge not found: {tag!r}" raise GaugeNotFoundError(msg) from exc def set_runtime_value(tag: Tag, value: GaugeValue, *, now: float | None = None) -> None: state = get_gauge_state(tag) now = time.monotonic() if now is None else now clamped, invalid = clamp_runtime_value(value, state.config) state.target_value = value state.clamped_value = clamped state.invalid_value = invalid state.value_revision += 1 state.previous_displayed_value = state.displayed_value state.animation_state = start_animation( state.displayed_value, clamped, animation_config(state.config), now ) if not state.animation_state.active: smoothing = smoothing_config(state.config) if smoothing is not None and smoothing.enabled and clamped is not None and not invalid: if state.smoothing_state.value is None: state.smoothing_state.value = state.previous_displayed_value state.smoothing_state.active = state.smoothing_state.value != clamped else: state.displayed_value = clamped state.smoothing_state.value = clamped state.smoothing_state.active = False state.dirty |= DirtyFlags.VALUE | DirtyFlags.DYNAMIC | DirtyFlags.TEXT render_runtime_gauge(state) def get_runtime_value(tag: Tag) -> GaugeValue: return get_gauge_state(tag).target_value def update_runtime_gauge(tag: Tag, **config: Any) -> None: state = get_gauge_state(tag) value_marker = object() value = config.pop("value", value_marker) dirty = DirtyFlags.NONE if config: old_config = state.config new_config = replace(old_config, **config) state.config = new_config state.config_revision += 1 if any(name in config for name in ("min_value", "max_value", "clamp")): clamped, invalid = clamp_runtime_value(state.target_value, new_config) state.clamped_value = clamped state.invalid_value = invalid state.displayed_value = clamped state.smoothing_state.value = clamped dirty |= DirtyFlags.STATIC | DirtyFlags.DYNAMIC | DirtyFlags.TEXT if any(name in config for name in ("thresholds", "markers")): dirty |= DirtyFlags.STATIC | DirtyFlags.DYNAMIC if any(name in config for name in ("state", "states")): dirty |= DirtyFlags.DYNAMIC | DirtyFlags.TEXT if any( name in config for name in ("label", "unit", "precision", "show_value", "show_unit") ): dirty |= DirtyFlags.TEXT if any(name in config for name in ("width", "height", "autosize_x", "autosize_y", "show")): dirty |= DirtyFlags.SIZE | DirtyFlags.STATIC | DirtyFlags.DYNAMIC | DirtyFlags.TEXT if any(name in config for name in ("theme", "style")): dirty |= DirtyFlags.STYLE | DirtyFlags.STATIC | DirtyFlags.DYNAMIC | DirtyFlags.TEXT if value is not value_marker: set_runtime_value(tag, value) dirty |= DirtyFlags.VALUE | DirtyFlags.DYNAMIC | DirtyFlags.TEXT state.dirty |= dirty render_runtime_gauge(state) def set_runtime_thresholds( tag: Tag, thresholds: list[ThresholdBand] | tuple[ThresholdBand, ...] ) -> None: state = get_gauge_state(tag) validate_thresholds(thresholds, state.config.min_value, state.config.max_value) update_runtime_gauge(tag, thresholds=tuple(thresholds)) def set_runtime_markers(tag: Tag, markers: list[GaugeMarker] | tuple[GaugeMarker, ...]) -> None: state = get_gauge_state(tag) validate_markers(markers, state.config.min_value, state.config.max_value) update_runtime_gauge(tag, markers=tuple(markers)) def set_runtime_show(tag: Tag, show: bool) -> None: update_runtime_gauge(tag, show=show) def delete_runtime_gauge(tag: Tag) -> None: state = get_gauge_state(tag) if state.renderer is not None: state.renderer.delete() del _registry[tag] def render_runtime_gauge(state: GaugeState) -> None: if state.renderer is not None: state.renderer.render() def update_state_size(state: GaugeState, size: SizeState) -> None: state.measured_width = size.measured_width state.measured_height = size.measured_height state.last_nonzero_width = size.last_nonzero_width state.last_nonzero_height = size.last_nonzero_height state.dirty |= DirtyFlags.SIZE | DirtyFlags.STATIC | DirtyFlags.DYNAMIC | DirtyFlags.TEXT def size_state_for_gauge(state: GaugeState) -> SizeState: return SizeState( requested=RequestedSize( state.config.width, state.config.height, state.config.autosize_x, state.config.autosize_y, ), measured_width=state.measured_width, measured_height=state.measured_height, last_nonzero_width=state.last_nonzero_width, last_nonzero_height=state.last_nonzero_height, )