step 2: add gauge state sizing and animation model
This commit is contained in:
305
src/dpg_gauges/state.py
Normal file
305
src/dpg_gauges/state.py
Normal file
@@ -0,0 +1,305 @@
|
||||
"""Global configuration, gauge registry, and runtime state."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
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 GaugeConfig
|
||||
from .scales import clamp_value, 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 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 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 = 0.0) -> None:
|
||||
state = get_gauge_state(tag)
|
||||
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:
|
||||
state.displayed_value = clamped
|
||||
state.smoothing_state.value = clamped
|
||||
state.dirty |= DirtyFlags.VALUE | DirtyFlags.DYNAMIC | DirtyFlags.TEXT
|
||||
|
||||
|
||||
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 ("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
|
||||
|
||||
|
||||
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:
|
||||
get_gauge_state(tag)
|
||||
del _registry[tag]
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
Reference in New Issue
Block a user