MCPcopy Create free account
hub / github.com/Nicxe/f1_sensor / append

Method append

custom_components/f1_sensor/__init__.py:623–647  ·  view source on GitHub ↗
(
        self, payload: dict[str, Any], *, received_at: str | None = None
    )

Source from the content-addressed store, hash-verified

621 return [dict(item) for item in reversed(self._items[-max_items:])]
622
623 def append(
624 self, payload: dict[str, Any], *, received_at: str | None = None
625 ) -> dict[str, Any] | None:
626 if not isinstance(payload, dict):
627 return None
628 self._sync_session_context(force=True)
629 item = _normalize_race_control_log_item(
630 payload,
631 received_at=received_at,
632 sequence=self._next_sequence,
633 )
634 event_id = item.get("event_id")
635 if event_id and event_id in self._event_ids:
636 return None
637 self._items.append(item)
638 if event_id:
639 self._event_ids.add(str(event_id))
640 while len(self._items) > RACE_CONTROL_LOG_MAX_ITEMS:
641 old = self._items.pop(0)
642 old_event_id = old.get("event_id")
643 if old_event_id:
644 self._event_ids.discard(str(old_event_id))
645 self._next_sequence += 1
646 self._schedule_save()
647 return dict(item)
648
649 async def async_clear(self, *, reason: str = "manual") -> None:
650 self._clear(reason=reason)

Calls 4

_sync_session_contextMethod · 0.95
_schedule_saveMethod · 0.95
getMethod · 0.45

Tested by 15

subscribeMethod · 0.64
inject_messageMethod · 0.64
getMethod · 0.64
getMethod · 0.64
_fake_fetchFunction · 0.64
getMethod · 0.64
headMethod · 0.64
getMethod · 0.64
add_listenerMethod · 0.64
add_listenerMethod · 0.64