protocols

class StatusProtocol(*args, **kwargs)

Bases: Protocol

state: str

category: str

status: str

reason: str | None

code: int

color: str

set(state: str | None = None, category: str | None = None, status: str | None = None, reason: str | None = None, code: int | None = None) → None

display_name(**kwargs: Any) → str

is_terminal() → bool

class JobProtocol(*args, **kwargs)

Bases: Protocol

cpus: int

cpu_ids: list[int]

dependencies: list[JobProtocol]

exclusive: bool

gpus: int

gpu_ids: list[str]

id: str

mask: str

measurements: Measurements

status: StatusProtocol

runtime: float

timeout: float

timekeeper: Timekeeper

display_name(**kwargs: Any) → str

total_timeout() → float

set_status(state: str | None = None, category: str | None = None, status: str | None = None, reason: str | None = None, code: int | None = None) → None

refresh() → None

save() → None

size() → float

finish() → None

getstate() → dict[str, Any]

setstate(state: dict[str, Any]) → None

required_resources() → list[dict[str, Any]]

Returns a list of resource

required[i] == [{“type”: type, “slots”: slots}, …]

one entry per resource. For a test requiring 1 slot from 2 cpus:

required[i] = [{“type”: “cpus”, “slots”: 1}, {“type”: “cpus”: “slots”: 1}]

This general way of describing resources allows for oversubscribing resources. Each test requires 1 slot per required cpu, but the machine config can specify multiple slots per cpu available

property resources: list[dict[str, list[dict]]]

resources is of the form:

resources[i] = {str: [{"id": str, "slots": int}]}

If the test required 2 cpus and 2 gpus, resources would look like:

resources = [
  {"cpus": [{"id": "1", "slots": 1}, {"id": "2", "slots": 1}]},
  {"gpus": [{"id": "1", "slots": 1}, {"id": "2", "slots": 1}]},
]

assign_resources(arg: dict[str, list[dict]]) → None

free_resources() → dict[str, list[dict]]

cost() → float

class Measurements(data: dict[str, typing.Any]=<factory>)

Bases: object

data: dict[str, Any]

add_measurement(name: str, value: Any) → None

update(measurements: dict) → None

asdict() → dict[str, Any]

items() → Generator[tuple[str, Any], None, None]