API Referenz

Apps

api

map

ranking

Project: MyCyclingCity Generation: AI-based

Ranking application for MyCyclingCity.

leaderboard

Project: MyCyclingCity Generation: AI-based

Leaderboard application for MyCyclingCity.

kiosk

iot

game

mgmt

Utilities

api.utils

Project: MyCyclingCity Generation: AI-based

Shared utility functions for MyCyclingCity applications.

format_km_de

format_km_de(
    value: Optional[float],
    decimals: int = 3,
    language_code: Optional[str] = None,
) -> str

Format number based on language: German format (dot thousands, comma decimal) or English format (comma thousands, dot decimal).

Parameters:

Name	Type	Description	Default
value	Optional[float]	The numeric value to format. Can be None.	required
decimals	int	Number of decimal places (default: 3).	3
language_code	Optional[str]	Optional language code override. If None, uses current language.	None

Returns:

Type	Description
str	Formatted string with appropriate thousands and decimal separators.

Examples:

>>> format_km_de(1234.567, 3, 'de')
'1.234,567'
>>> format_km_de(1234.567, 3, 'en')
'1,234.567'

Source code in api/utils.py

def format_km_de(value: Optional[float], decimals: int = 3, language_code: Optional[str] = None) -> str:
    """
    Format number based on language: German format (dot thousands, comma decimal) 
    or English format (comma thousands, dot decimal).

    Args:
        value: The numeric value to format. Can be None.
        decimals: Number of decimal places (default: 3).
        language_code: Optional language code override. If None, uses current language.

    Returns:
        Formatted string with appropriate thousands and decimal separators.

    Examples:
        >>> format_km_de(1234.567, 3, 'de')
        '1.234,567'
        >>> format_km_de(1234.567, 3, 'en')
        '1,234.567'
    """
    if value is None:
        value = 0

    try:
        num = float(value)
        fixed = f"{num:.{decimals}f}"
        parts = fixed.split('.')
        integer_str = parts[0]
        decimal_part = parts[1] if len(parts) > 1 else ''

        # Get current language
        if language_code is None:
            current_language = translation.get_language()
            if not current_language:
                current_language = getattr(settings, 'LANGUAGE_CODE', 'de')
        else:
            current_language = language_code

        # Normalize language code (e.g., 'en-us' -> 'en', 'de-de' -> 'de')
        if current_language:
            lang_code = str(current_language).split('-')[0].lower()
        else:
            lang_code = 'de'
        is_german = (lang_code == 'de')

        if is_german:
            # German format: dot for thousands, comma for decimal
            integer_with_separators = ''
            for i, char in enumerate(reversed(integer_str)):
                if i > 0 and i % 3 == 0:
                    integer_with_separators = '.' + integer_with_separators
                integer_with_separators = char + integer_with_separators
            return integer_with_separators + (',' + decimal_part if decimal_part else '')
        else:
            # English format: comma for thousands, dot for decimal
            integer_with_separators = ''
            for i, char in enumerate(reversed(integer_str)):
                if i > 0 and i % 3 == 0:
                    integer_with_separators = ',' + integer_with_separators
                integer_with_separators = char + integer_with_separators
            return integer_with_separators + ('.' + decimal_part if decimal_part else '')
    except (ValueError, TypeError):
        if language_code is None:
            current_language = translation.get_language()
            if not current_language:
                current_language = getattr(settings, 'LANGUAGE_CODE', 'de')
        else:
            current_language = language_code
        lang_code = str(current_language).split('-')[0].lower() if current_language else 'de'
        is_german = lang_code == 'de'
        decimal_sep = ',' if is_german else '.'
        return '0' + (decimal_sep + '0' * decimals if decimals > 0 else '')

api.helpers

Project: MyCyclingCity Generation: AI-based

Shared helper functions for building group hierarchies and data structures. Used by map, ranking, and leaderboard apps.

are_all_parents_visible

are_all_parents_visible(group: Group) -> bool

Check if all parent groups in the hierarchy are visible.

Parameters:

Name	Type	Description	Default
group	Group	The group to check.	required

Returns:

Type	Description
bool	True if the group and all its parents are visible, False otherwise.

Source code in api/helpers.py

def are_all_parents_visible(group: Group) -> bool:
    """
    Check if all parent groups in the hierarchy are visible.

    Args:
        group: The group to check.

    Returns:
        True if the group and all its parents are visible, False otherwise.
    """
    visited = set()
    current = group

    # First check the group itself
    if not current.is_visible:
        return False

    # Then check all parent groups recursively
    while current and current.parent_id:
        if current.id in visited:
            # Circular reference detected, break to avoid infinite loop
            break
        visited.add(current.id)

        # Load parent if not already loaded (select_related only loads direct parent)
        if not hasattr(current, 'parent') or current.parent is None:
            try:
                current.parent = Group.objects.get(id=current.parent_id)
            except Group.DoesNotExist:
                break

        # Move to parent and check if it's visible
        current = current.parent
        if current:
            if current.id in visited:
                break
            visited.add(current.id)
            if not current.is_visible:
                return False

    return True

build_cyclist_velos_api_fields

build_cyclist_velos_api_fields(
    cyclist: Cyclist,
    *,
    include_session: bool = False,
    include_daily: bool = False,
    period_start: Optional[datetime] = None,
    period_end: Optional[datetime] = None
) -> Dict[str, int]

Standard Velos fields for cyclist API responses.

Source code in api/helpers.py

def build_cyclist_velos_api_fields(
    cyclist: Cyclist,
    *,
    include_session: bool = False,
    include_daily: bool = False,
    period_start: Optional[timezone.datetime] = None,
    period_end: Optional[timezone.datetime] = None,
) -> Dict[str, int]:
    """Standard Velos fields for cyclist API responses."""
    fields: Dict[str, int] = {
        'velos_balance': _get_cyclist_velos_balance(cyclist),
        'velos_total': get_cyclist_velos_total(cyclist),
    }
    if include_session:
        fields['session_velos'] = get_cyclist_session_velos(cyclist)
    if include_daily:
        fields['velos_daily'] = get_cyclist_velos_daily(cyclist)
    if period_start and period_end:
        fields['velos_period'] = get_cyclist_velos_period(cyclist, period_start, period_end)
    return fields

build_events_data

build_events_data(
    kiosk: bool = False,
) -> List[Dict[str, Any]]

Build event data structure for display.

Parameters:

Name	Type	Description	Default
kiosk	bool	Whether in kiosk mode (filters groups with distance > 0).	False

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries containing event data.

Source code in api/helpers.py

def build_events_data(kiosk: bool = False) -> List[Dict[str, Any]]:
    """
    Build event data structure for display.

    Args:
        kiosk: Whether in kiosk mode (filters groups with distance > 0).

    Returns:
        List of dictionaries containing event data.
    """
    from django.utils import timezone
    # Event is already imported at top of file from eventboard.models

    now = timezone.now()
    active_events = Event.objects.filter(is_active=True, is_visible_on_map=True)
    # Filter by should_be_displayed() instead of is_currently_active()
    # This allows showing events after end_time until hide_after_date
    active_events = [e for e in active_events if e.should_be_displayed()]
    events_data = []

    for event in active_events:
        # Get all groups participating in this event
        event_groups = []
        for status in event.group_statuses.select_related('group').all():
            # In kiosk mode, only show groups with current_velos > 0
            if kiosk and int(status.current_velos) <= 0:
                continue
            event_groups.append({
                'name': status.group.name,
                'velos': int(status.current_velos),
                'group_id': status.group.id
            })
        # In kiosk mode, only add event if it has groups with Velos > 0
        if event_groups and (not kiosk or len(event_groups) > 0):
            event_groups_sorted = sorted(event_groups, key=lambda x: x['velos'], reverse=True)[:10]
            total_velos = sum(g['velos'] for g in event_groups)
            is_ended = event.end_time and now > event.end_time
            events_data.append({
                'id': event.id,
                'name': event.name,
                'event_type': event.get_event_type_display(),
                'description': event.description or '',
                'start_time': event.start_time,
                'end_time': event.end_time,
                'total_velos': total_velos,
                'is_ended': is_ended,
                'groups': event_groups_sorted
            })

    return events_data

build_group_hierarchy

build_group_hierarchy(
    target_group: Optional[Group] = None,
    kiosk: bool = False,
    show_cyclists: bool = True,
) -> List[Dict[str, Any]]

Build a hierarchical data structure of groups with their members and subgroups.

Parameters:

Name	Type	Description	Default
target_group	Optional[Group]	Optional specific group to filter by.	None
kiosk	bool	Whether in kiosk mode (hides groups with zero metric Velos).	False
show_cyclists	bool	Whether to include cyclist data in the hierarchy.	True

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries containing group hierarchy data.

Source code in api/helpers.py

def build_group_hierarchy(
    target_group: Optional[Group] = None,
    kiosk: bool = False,
    show_cyclists: bool = True
) -> List[Dict[str, Any]]:
    """
    Build a hierarchical data structure of groups with their members and subgroups.

    Args:
        target_group: Optional specific group to filter by.
        kiosk: Whether in kiosk mode (hides groups with zero metric Velos).
        show_cyclists: Whether to include cyclist data in the hierarchy.

    Returns:
        List of dictionaries containing group hierarchy data.
    """
    group_filter = {'is_visible': True}

    if target_group:
        parent_groups = Group.objects.filter(id=target_group.id, **group_filter).order_by('name')
    else:
        parent_groups = Group.objects.filter(parent__isnull=True, **group_filter).order_by('name')

    return build_hierarchy_from_parent_groups(
        parent_groups,
        kiosk=kiosk,
        show_cyclists=show_cyclists,
    )

build_group_velos_api_fields

build_group_velos_api_fields(
    group: Group,
) -> Dict[str, int]

Ledger Velos fields for group API responses.

Source code in api/helpers.py

def build_group_velos_api_fields(group: Group) -> Dict[str, int]:
    """Ledger Velos fields for group API responses."""
    return {
        'velos_total': int(group.velos_total or 0),
        'velos_spendable': int(group.velos_spendable or 0),
    }

build_hierarchy_from_parent_groups

build_hierarchy_from_parent_groups(
    parent_groups,
    kiosk: bool = False,
    show_cyclists: bool = True,
) -> List[Dict[str, Any]]

Build hierarchy data from a parent-group queryset (map/ranking shared helper).

All Velos totals come from HourlyMetric (synced by mcc_worker for active sessions). Parent groups: sum of child group metric totals.

Source code in api/helpers.py

def build_hierarchy_from_parent_groups(
    parent_groups,
    kiosk: bool = False,
    show_cyclists: bool = True,
) -> List[Dict[str, Any]]:
    """
    Build hierarchy data from a parent-group queryset (map/ranking shared helper).

    All Velos totals come from HourlyMetric (synced by mcc_worker for active sessions).
    Parent groups: sum of child group metric totals.
    """
    group_filter = {'is_visible': True}
    member_filter = {'is_visible': True}
    if kiosk:
        member_filter['distance_total__gt'] = 0

    hierarchy = []
    groups_for_metrics: List[Group] = []
    for p_group in parent_groups:
        groups_for_metrics.append(p_group)
        groups_for_metrics.extend(list(p_group.children.filter(**group_filter)))
    group_velos_by_id = _calculate_group_velos_from_metrics(groups_for_metrics, use_cache=True)
    group_km_by_id = _calculate_group_totals_from_metrics(groups_for_metrics, use_cache=True)

    for p_group in parent_groups:
        direct_members = _members_for_group(p_group, member_filter, show_cyclists)

        subgroups_qs = p_group.children.filter(**group_filter).order_by('name')

        subgroups_data = []
        for sub in subgroups_qs:
            sub_member_data = _members_for_group(sub, member_filter, show_cyclists)
            sub_velos = _group_velos_for_ranking(
                sub,
                sub_member_data,
                group_metric_velos=group_velos_by_id.get(sub.id, 0),
            )
            sub_km = _group_km_for_ranking(
                sub,
                sub_member_data,
                group_metric_km=group_km_by_id.get(sub.id, 0.0),
            )
            if not kiosk or (sub_velos > 0 or sub_member_data):
                subgroups_data.append({
                    'id': sub.id,
                    'name': sub.name,
                    'km': round(float(sub_km), 3),
                    'velos': sub_velos,
                    'members': sub_member_data,
                })

        subgroups_data = sorted(subgroups_data, key=lambda x: x['velos'], reverse=True)
        p_velos = _group_velos_for_ranking(
            p_group,
            direct_members,
            child_entries=subgroups_data,
            group_metric_velos=group_velos_by_id.get(p_group.id, 0),
        )
        p_km = _group_km_for_ranking(
            p_group,
            direct_members,
            child_entries=subgroups_data,
            group_metric_km=group_km_by_id.get(p_group.id, 0.0),
        )
        if not kiosk or (p_velos > 0 or subgroups_data or direct_members):
            hierarchy.append({
                'id': p_group.id,
                'name': p_group.name,
                'km': round(float(p_km), 3),
                'velos': p_velos,
                'direct_members': direct_members,
                'subgroups': subgroups_data,
            })

    return sorted(hierarchy, key=lambda x: x['velos'], reverse=True)

filter_cyclist_metrics_by_snapshot

filter_cyclist_metrics_by_snapshot(
    queryset: QuerySet, cyclist_ids: List[int]
) -> QuerySet

Filter HourlyMetric queryset for cyclists, excluding metrics before latest snapshot date.

This function determines snapshot dates based on the groups each cyclist belongs to, then filters metrics accordingly.

Parameters:

Name	Type	Description	Default
queryset	QuerySet	HourlyMetric queryset to filter (should already be filtered by cyclist)	required
cyclist_ids	List[int]	List of cyclist IDs	required

Returns:

Type	Description
QuerySet	Filtered queryset with snapshot-aware filtering applied

Source code in api/helpers.py

def filter_cyclist_metrics_by_snapshot(queryset: QuerySet, cyclist_ids: List[int]) -> QuerySet:
    """
    Filter HourlyMetric queryset for cyclists, excluding metrics before latest snapshot date.

    This function determines snapshot dates based on the groups each cyclist belongs to,
    then filters metrics accordingly.

    Args:
        queryset: HourlyMetric queryset to filter (should already be filtered by cyclist)
        cyclist_ids: List of cyclist IDs

    Returns:
        Filtered queryset with snapshot-aware filtering applied
    """
    if not cyclist_ids:
        return queryset

    # Get groups for each cyclist
    cyclist_groups_map: Dict[int, List[int]] = {}
    for cyclist in Cyclist.objects.filter(id__in=cyclist_ids):
        cyclist_groups_map[cyclist.id] = list(cyclist.groups.values_list('id', flat=True))

    # Get snapshot dates for all groups
    all_group_ids = set()
    for group_ids in cyclist_groups_map.values():
        all_group_ids.update(group_ids)

    snapshot_dates = _get_latest_snapshot_date_for_groups(list(all_group_ids))

    # For each cyclist, find the latest snapshot date from their groups
    cyclist_snapshot_dates: Dict[int, Optional[timezone.datetime]] = {}
    for cyclist_id, group_ids in cyclist_groups_map.items():
        latest_date = None
        for group_id in group_ids:
            group_date = snapshot_dates.get(group_id)
            if group_date and (latest_date is None or group_date > latest_date):
                latest_date = group_date
        cyclist_snapshot_dates[cyclist_id] = latest_date

    # Build Q objects for filtering
    q_objects = []
    for cyclist_id in cyclist_ids:
        snapshot_date = cyclist_snapshot_dates.get(cyclist_id)
        if snapshot_date:
            # Only include metrics after snapshot date for this cyclist
            q_objects.append(
                Q(cyclist_id=cyclist_id, timestamp__gt=snapshot_date)
            )
        else:
            # No snapshot, include all metrics for this cyclist
            q_objects.append(Q(cyclist_id=cyclist_id))

    if q_objects:
        return queryset.filter(reduce(operator.or_, q_objects))

    return queryset

filter_metrics_by_snapshot

filter_metrics_by_snapshot(
    queryset: QuerySet,
    group_ids: List[int],
    field_name: str = "group_at_time_id",
) -> QuerySet

Filter HourlyMetric queryset to exclude metrics before latest snapshot date.

This function applies snapshot filtering to a queryset by checking each group's latest snapshot date and only including metrics after that date.

Parameters:

Name	Type	Description	Default
queryset	QuerySet	HourlyMetric queryset to filter	required
group_ids	List[int]	List of group IDs to check for snapshots	required
field_name	str	Field name to use for group filtering (default: 'group_at_time_id') Use 'group_at_time_id' for group metrics, 'cyclist__groups__id' for cyclist metrics	'group_at_time_id'

Returns:

Type	Description
QuerySet	Filtered queryset with snapshot-aware filtering applied

Source code in api/helpers.py

def filter_metrics_by_snapshot(queryset: QuerySet, group_ids: List[int], field_name: str = 'group_at_time_id') -> QuerySet:
    """
    Filter HourlyMetric queryset to exclude metrics before latest snapshot date.

    This function applies snapshot filtering to a queryset by checking each group's
    latest snapshot date and only including metrics after that date.

    Args:
        queryset: HourlyMetric queryset to filter
        group_ids: List of group IDs to check for snapshots
        field_name: Field name to use for group filtering (default: 'group_at_time_id')
                   Use 'group_at_time_id' for group metrics, 'cyclist__groups__id' for cyclist metrics

    Returns:
        Filtered queryset with snapshot-aware filtering applied
    """
    if not group_ids:
        return queryset

    # Get snapshot dates for all groups
    snapshot_dates = _get_latest_snapshot_date_for_groups(group_ids)

    # Build Q objects for filtering
    # For each group, if it has a snapshot, only include metrics after that date
    q_objects = []

    for group_id in group_ids:
        snapshot_date = snapshot_dates.get(group_id)
        if snapshot_date:
            # Only include metrics after snapshot date for this group
            q_objects.append(
                Q(**{field_name: group_id, 'timestamp__gt': snapshot_date})
            )
        else:
            # No snapshot, include all metrics for this group
            q_objects.append(Q(**{field_name: group_id}))

    if q_objects:
        # Combine all Q objects with OR (metrics matching any group condition)
        return queryset.filter(reduce(operator.or_, q_objects))

    return queryset

get_cyclist_by_identifier

get_cyclist_by_identifier(
    identifier: str,
) -> Optional[Cyclist]

Resolve a cyclist by user_id or id_tag (case-insensitive).

Source code in api/helpers.py

def get_cyclist_by_identifier(identifier: str) -> Optional[Cyclist]:
    """Resolve a cyclist by user_id or id_tag (case-insensitive)."""
    from django.db.models import Q

    try:
        return Cyclist.objects.get(
            Q(user_id__iexact=identifier) | Q(id_tag__iexact=identifier)
        )
    except Cyclist.DoesNotExist:
        return None

get_cyclist_session_km

get_cyclist_session_km(cyclist: Cyclist)

Return cumulative session distance in km for the cyclist's active device session.

Source code in api/helpers.py

def get_cyclist_session_km(cyclist: Cyclist):
    """Return cumulative session distance in km for the cyclist's active device session."""
    from decimal import Decimal

    try:
        session = cyclist.cyclistdevicecurrentmileage
    except CyclistDeviceCurrentMileage.DoesNotExist:
        return Decimal('0.00000')
    return session.cumulative_mileage or Decimal('0.00000')

get_cyclist_session_velos

get_cyclist_session_velos(cyclist: Cyclist) -> int

Velos earned in the cyclist's active device session (not yet redeemed).

Source code in api/helpers.py

def get_cyclist_session_velos(cyclist: Cyclist) -> int:
    """Velos earned in the cyclist's active device session (not yet redeemed)."""
    from api.velos import calculate_session_velos

    try:
        session = cyclist.cyclistdevicecurrentmileage
    except CyclistDeviceCurrentMileage.DoesNotExist:
        return 0
    if not session or not session.cumulative_mileage:
        return 0
    return calculate_session_velos(session.cumulative_mileage, session.device)

get_cyclist_velos_daily

get_cyclist_velos_daily(
    cyclist: Cyclist, now: Optional[datetime] = None
) -> int

Velos earned today (metrics + active session).

Source code in api/helpers.py

def get_cyclist_velos_daily(cyclist: Cyclist, now: Optional[timezone.datetime] = None) -> int:
    """Velos earned today (metrics + active session)."""
    if now is None:
        now = timezone.now()
    today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    snapshot_date = _cyclist_effective_snapshot_start(cyclist)
    effective_start = max(today_start, snapshot_date) if snapshot_date else today_start

    daily_velos = HourlyMetric.objects.filter(
        cyclist=cyclist,
        timestamp__gte=effective_start,
        group_at_time__isnull=False,
    ).aggregate(total=Sum('velos'))['total'] or 0

    try:
        session = cyclist.cyclistdevicecurrentmileage
    except CyclistDeviceCurrentMileage.DoesNotExist:
        return int(daily_velos)

    if (
        session
        and session.cumulative_mileage
        and session.last_activity
        and session.last_activity >= today_start
    ):
        daily_velos += get_cyclist_session_velos(cyclist)

    return int(daily_velos)

get_cyclist_velos_period

get_cyclist_velos_period(
    cyclist: Cyclist, start_dt: datetime, end_dt: datetime
) -> int

Sum HourlyMetric.velos for a cyclist in a date range plus overlapping session.

Source code in api/helpers.py

def get_cyclist_velos_period(
    cyclist: Cyclist,
    start_dt: timezone.datetime,
    end_dt: timezone.datetime,
) -> int:
    """Sum HourlyMetric.velos for a cyclist in a date range plus overlapping session."""
    period_velos = HourlyMetric.objects.filter(
        cyclist=cyclist,
        timestamp__gte=start_dt,
        timestamp__lte=end_dt,
        group_at_time__isnull=False,
    ).aggregate(total=Sum('velos'))['total'] or 0

    try:
        session = cyclist.cyclistdevicecurrentmileage
    except CyclistDeviceCurrentMileage.DoesNotExist:
        return int(period_velos)

    if (
        session
        and session.cumulative_mileage
        and session.last_activity
        and start_dt <= session.last_activity <= end_dt
        and (not session.start_time or session.start_time <= end_dt)
    ):
        period_velos += get_cyclist_session_velos(cyclist)

    return int(period_velos)

get_cyclist_velos_total

get_cyclist_velos_total(
    cyclist: Cyclist, use_cache: bool = False
) -> int

Lifetime Velos from HourlyMetric (respecting year-end snapshots).

Source code in api/helpers.py

def get_cyclist_velos_total(cyclist: Cyclist, use_cache: bool = False) -> int:
    """Lifetime Velos from HourlyMetric (respecting year-end snapshots)."""
    totals = _calculate_cyclist_velos_from_metrics([cyclist], use_cache=use_cache)
    return int(totals.get(cyclist.id, 0))

get_external_display_settings_context

get_external_display_settings_context() -> Dict[str, Any]

Return admin-controlled km display flags for external GUIs.

Source code in api/helpers.py

def get_external_display_settings_context() -> Dict[str, Any]:
    """Return admin-controlled km display flags for external GUIs."""
    from api.models import ExternalDisplaySettings

    settings_obj = ExternalDisplaySettings.get_settings()
    return {
        'show_km_in_leaderboard_footer': settings_obj.show_km_in_leaderboard_footer,
        'show_km_in_ranking_headers': settings_obj.show_km_in_ranking_headers,
        'km_display_decimals': settings_obj.km_display_decimals,
    }

get_group_velos_ledger

get_group_velos_ledger(
    groups: List[Group],
) -> Dict[int, int]

Return permanent group Velos ledger totals (leaderboard ranking source).

Uses Group.velos_total, not the sum of member velos_balance values.

Source code in api/helpers.py

def get_group_velos_ledger(groups: List[Group]) -> Dict[int, int]:
    """
    Return permanent group Velos ledger totals (leaderboard ranking source).

    Uses Group.velos_total, not the sum of member velos_balance values.
    """
    return {group.id: int(group.velos_total or 0) for group in groups}

get_leaderboard_footer_period_context

get_leaderboard_footer_period_context(
    groups_data: List[Dict[str, Any]],
) -> Dict[str, Any]

Describe the time window used for footer total Velos/km (HourlyMetric sums).

Totals count metrics after each group's latest YearEndSnapshot, or all time when no snapshot exists. Returns template-friendly mode + dates.

Source code in api/helpers.py

def get_leaderboard_footer_period_context(
    groups_data: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """
    Describe the time window used for footer total Velos/km (HourlyMetric sums).

    Totals count metrics after each group's latest YearEndSnapshot, or all time
    when no snapshot exists. Returns template-friendly mode + dates.
    """
    from datetime import date, timedelta

    group_ids = [int(g['id']) for g in groups_data if g.get('id')]
    if not group_ids:
        return {
            'footer_totals_period_mode': 'none',
            'footer_totals_period_since': None,
            'footer_totals_period_dates': [],
        }

    snapshot_dates = _get_latest_snapshot_date_for_groups(group_ids)
    unique_calendar_dates = sorted({
        snapshot_date.date()
        for snapshot_date in snapshot_dates.values()
        if snapshot_date is not None
    })

    if not unique_calendar_dates:
        return {
            'footer_totals_period_mode': 'all_time',
            'footer_totals_period_since': None,
            'footer_totals_period_dates': [],
        }

    if len(unique_calendar_dates) == 1:
        # HourlyMetric uses timestamp__gt snapshot_date → effective start is next day
        effective_start: date = unique_calendar_dates[0] + timedelta(days=1)
        return {
            'footer_totals_period_mode': 'single',
            'footer_totals_period_since': effective_start,
            'footer_totals_period_dates': [],
        }

    effective_starts = [d + timedelta(days=1) for d in unique_calendar_dates]
    return {
        'footer_totals_period_mode': 'multiple',
        'footer_totals_period_since': None,
        'footer_totals_period_dates': effective_starts,
    }

invalidate_cache_for_top_group

invalidate_cache_for_top_group(top_group: Group)

Invalidate all cache entries related to a TOP group and its subgroups.

This function clears cache for: - Group totals (leaderboard) - Cyclist totals (ranking) - Device totals (ranking) - All subgroups and their descendants

Parameters:

Name	Type	Description	Default
top_group	Group	The TOP group for which to invalidate cache	required

Source code in api/helpers.py

def invalidate_cache_for_top_group(top_group: Group):
    """
    Invalidate all cache entries related to a TOP group and its subgroups.

    This function clears cache for:
    - Group totals (leaderboard)
    - Cyclist totals (ranking)
    - Device totals (ranking)
    - All subgroups and their descendants

    Args:
        top_group: The TOP group for which to invalidate cache
    """
    from django.core.cache import cache
    from eventboard.utils import get_all_subgroup_ids

    # Get all subgroup IDs (including TOP group itself)
    all_subgroup_ids = get_all_subgroup_ids(top_group)
    all_subgroup_ids.append(top_group.id)

    # Get all cyclists in these groups
    all_cyclist_ids = list(Cyclist.objects.filter(groups__id__in=all_subgroup_ids).values_list('id', flat=True).distinct())

    # Get all devices in these groups
    all_device_ids = list(Device.objects.filter(group__id__in=all_subgroup_ids).values_list('id', flat=True))

    # Invalidate group totals cache (leaderboard)
    # Cache keys follow pattern: 'leaderboard_group_totals_{group_ids}_{timestamp}_desc{flag}'
    # We need to clear all possible combinations, so we'll use a pattern-based approach
    # Since we can't easily list all cache keys, we'll clear by pattern matching
    # For simplicity, we'll clear all leaderboard caches (they'll be regenerated on next request)
    cache_patterns = [
        'leaderboard_group_totals_*',
        'ranking_cyclist_totals_*',
        'ranking_device_totals_*',
        'ranking_group_totals_*',
    ]

    # Django cache doesn't support pattern deletion directly, so we'll use a workaround:
    # Set a version number that changes, or clear specific known keys
    # For now, we'll clear the most common patterns by trying to delete known key formats

    # Clear cache for specific group IDs (most efficient approach)
    group_ids_str = "-".join(map(str, sorted(all_subgroup_ids)))
    now = timezone.now()

    # Try to clear common cache key patterns
    for hour_offset in range(24):  # Clear last 24 hours of cache
        timestamp = (now - timedelta(hours=hour_offset)).strftime("%Y%m%d%H")
        for desc_flag in [0, 1]:
            cache_key = f'leaderboard_group_totals_{group_ids_str}_{timestamp}_desc{desc_flag}'
            cache.delete(cache_key)

    # Clear cyclist totals cache
    if all_cyclist_ids:
        cyclist_ids_str = "-".join(map(str, sorted(all_cyclist_ids)))
        for hour_offset in range(24):
            timestamp = (now - timedelta(hours=hour_offset)).strftime("%Y%m%d%H")
            cache_key = f'ranking_cyclist_totals_{cyclist_ids_str}_{timestamp}'
            cache.delete(cache_key)

    # Clear device totals cache
    if all_device_ids:
        device_ids_str = "-".join(map(str, sorted(all_device_ids)))
        for hour_offset in range(24):
            timestamp = (now - timedelta(hours=hour_offset)).strftime("%Y%m%d%H")
            cache_key = f'ranking_device_totals_{device_ids_str}_{timestamp}'
            cache.delete(cache_key)

    # Clear group totals cache (from helpers.py)
    for hour_offset in range(24):
        timestamp = (now - timedelta(hours=hour_offset)).strftime("%Y%m%d%H")
        cache_key = f'ranking_group_totals_{group_ids_str}_{timestamp}'
        cache.delete(cache_key)

    logger.info(f"Invalidated cache for TOP group '{top_group.name}' (ID: {top_group.id}) and {len(all_subgroup_ids)} subgroups")

snapshot_session_velos

snapshot_session_velos(cyclist: Cyclist) -> int

Snapshot session Velos before ending a device session (e.g. game round stop).

Source code in api/helpers.py

def snapshot_session_velos(cyclist: Cyclist) -> int:
    """Snapshot session Velos before ending a device session (e.g. game round stop)."""
    return get_cyclist_session_velos(cyclist)

sum_display_totals_from_groups_data

sum_display_totals_from_groups_data(
    groups_data: List[Dict[str, Any]],
) -> Dict[str, Any]

Sum Velos and km from leaderboard-style group dicts (filtered view).

Uses the same per-group values already prepared for the UI (HourlyMetric-based where applicable).

Source code in api/helpers.py

def sum_display_totals_from_groups_data(groups_data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Sum Velos and km from leaderboard-style group dicts (filtered view).

    Uses the same per-group values already prepared for the UI (HourlyMetric-based
    where applicable).
    """
    return {
        'total_velos': sum(int(g.get('velos_total', 0) or 0) for g in groups_data),
        'total_km': sum(float(g.get('distance_total', 0) or 0) for g in groups_data),
    }