Source code for gyoza.resources

"""Cooperative resource reservation updates from inside an operation."""

import json
import logging

from gyoza.models.resources import validate_reservation




class GyozaResources:
    """Emit reservation updates through the existing Gyoza event stream."""



    def set_reservation(
        self, *, ram_mb: int | None = None, vram_mb: int | None = None
    ) -> None:
        """Declare remaining reservations after physically freeing memory.

        Omitted fields are unchanged. The master currently rejects increases.
        This emits an asynchronous event; returning does not confirm acceptance.
        """
        reservation = {
            name: value
            for name, value in (("ram_mb", ram_mb), ("vram_mb", vram_mb))
            if value is not None
        }
        validate_reservation(reservation)
        logging.getLogger("gyoza").info("[RESERVATION] %s", json.dumps(reservation))




gyoza_resources = GyozaResources()