335 lines
11 KiB
Python
335 lines
11 KiB
Python
"""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,
|
|
)
|