wehub-resource-sync
52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from livekit.protocol import agent
|
|
from livekit.protocol.agent_pb import agent_dev
|
|
|
|
from ..job import JobAcceptArguments, RunningJobInfo
|
|
|
|
|
|
@dataclass
|
|
class CliArgs:
|
|
log_level: str
|
|
url: str | None = None
|
|
api_key: str | None = field(repr=False, default=None)
|
|
api_secret: str | None = field(repr=False, default=None)
|
|
# address of the driving `lk` CLI's dev channel (hot-reload IPC + ServerInfo).
|
|
cli_addr: str | None = None
|
|
log_format: str = "json"
|
|
dev: bool = False
|
|
# set by `lk simulate` when launching the agent under test; disables the
|
|
# worker load limit so simulation runs can saturate the agent
|
|
simulation: bool = False
|
|
|
|
|
|
def running_job_to_proto(info: RunningJobInfo) -> agent_dev.RunningAgentJobInfo:
|
|
return agent_dev.RunningAgentJobInfo(
|
|
job=info.job.SerializeToString(),
|
|
accept_name=info.accept_arguments.name,
|
|
accept_identity=info.accept_arguments.identity,
|
|
accept_metadata=info.accept_arguments.metadata,
|
|
url=info.url,
|
|
token=info.token,
|
|
worker_id=info.worker_id,
|
|
mock_job=info.fake_job,
|
|
)
|
|
|
|
|
|
def running_job_from_proto(pb: agent_dev.RunningAgentJobInfo) -> RunningJobInfo:
|
|
return RunningJobInfo(
|
|
accept_arguments=JobAcceptArguments(
|
|
name=pb.accept_name,
|
|
identity=pb.accept_identity,
|
|
metadata=pb.accept_metadata,
|
|
),
|
|
job=agent.Job.FromString(pb.job),
|
|
url=pb.url,
|
|
token=pb.token,
|
|
worker_id=pb.worker_id,
|
|
fake_job=pb.mock_job,
|
|
)
|