From dd6410fd5a897f8563c10102614ffcc372620bfc Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Sun, 29 Mar 2026 07:16:16 -0500 Subject: [PATCH] support dot-notation for nested keys to avoid returning large objects when only specific subfields are needed --- frigate/stats/emitter.py | 51 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 48 insertions(+), 3 deletions(-) diff --git a/frigate/stats/emitter.py b/frigate/stats/emitter.py index 42d4c16a8..b9ae1d6b3 100644 --- a/frigate/stats/emitter.py +++ b/frigate/stats/emitter.py @@ -52,18 +52,63 @@ class StatsEmitter(threading.Thread): def get_stats_history( self, keys: Optional[list[str]] = None ) -> list[dict[str, Any]]: - """Get stats history.""" + """Get stats history. + + Supports dot-notation for nested keys to avoid returning large objects + when only specific subfields are needed. Handles two patterns: + + - Flat dict: "service.last_updated" returns {"service": {"last_updated": ...}} + - Dict-of-dicts: "cameras.camera_fps" returns each camera entry filtered + to only include "camera_fps" + """ if not keys: return self.stats_history + # Pre-parse keys into top-level keys and dot-notation fields + top_level_keys: list[str] = [] + nested_keys: dict[str, list[str]] = {} + + for k in keys: + if "." in k: + parent_key, child_key = k.split(".", 1) + nested_keys.setdefault(parent_key, []).append(child_key) + else: + top_level_keys.append(k) + selected_stats: list[dict[str, Any]] = [] for s in self.stats_history: - selected = {} + selected: dict[str, Any] = {} - for k in keys: + for k in top_level_keys: selected[k] = s.get(k) + for parent_key, child_keys in nested_keys.items(): + parent = s.get(parent_key) + + if not isinstance(parent, dict): + selected[parent_key] = parent + continue + + # Check if values are dicts (dict-of-dicts like cameras/detectors) + first_value = next(iter(parent.values()), None) + + if isinstance(first_value, dict): + # Filter each nested entry to only requested fields + selected[parent_key] = { + entry_key: { + field: entry.get(field) for field in child_keys + } + for entry_key, entry in parent.items() + } + else: + # Flat dict (like service) - pick individual fields + if parent_key not in selected: + selected[parent_key] = {} + + for child_key in child_keys: + selected[parent_key][child_key] = parent.get(child_key) + selected_stats.append(selected) return selected_stats