?HAMA

Edit File Size: 1.70 KB

/opt/cloudlinux/venv/lib64/python3.11/site-packages/lvestats/plugins/generic/burster/storage/utils.py

Text Base64

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import typing
import itertools
from typing import Collection, Iterator

from lvestats.orm import BurstingEventType
from .base import InBurstingEventRow, Timestamp
from ..common import LveId

def lves_historical_events_iter(
    lves_ids: Collection[LveId],
    events_ts_and_type_iter: Iterator[tuple[Timestamp, BurstingEventType]],
):
    for ts, event_type in events_ts_and_type_iter:
        for lve_id in lves_ids:
            yield InBurstingEventRow(
                lve_id=lve_id,
                timestamp=ts,
                event_type=event_type,
            )

def lve_historical_events_ts_and_type_iter(
    start_stop_pairs_iter: Iterator[tuple[Timestamp, Timestamp]],
) -> Iterator[tuple[Timestamp, BurstingEventType]]:
    timestemps_iter = itertools.chain.from_iterable(start_stop_pairs_iter)
    event_types_iter = itertools.cycle([BurstingEventType.STARTED, BurstingEventType.STOPPED])
    for timestamp, event_type in zip(timestemps_iter, event_types_iter):
        yield timestamp, event_type

def lve_historical_intervals_iter(
    overusing_intervals: int,
    interval_duration: int,
    cooldown: int = 1,
) -> Iterator[tuple[Timestamp, Timestamp]]:
    assert overusing_intervals >= 0
    assert interval_duration >= 1
    assert cooldown >= 1
    now = Timestamp(0)
    for _ in range(overusing_intervals):
        started_at, stopped_at = now, now + interval_duration
        yield typing.cast(tuple[Timestamp, Timestamp], (started_at, stopped_at))
        now = typing.cast(Timestamp, stopped_at + cooldown)

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-06-03 07:00:56	Edit Download recursive recursive
base.py	1.27 KB	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive
cleanup.py	2.04 KB	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive
load.py	4.10 KB	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive
save.py	2.67 KB	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive
utils.py	1.70 KB	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive
__init__.py	545 B	lrw-r--r--	2026-05-20 17:45:52	Edit Download recursive recursive

Directory Contents