Spaces:
Sleeping
Sleeping
File size: 12,414 Bytes
deb7c43 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 | """
Based on https://kernelpanic.io/the-modern-way-to-call-apis-in-python
TODO : use async call to API
"""
# from httpx import AsyncClient
import dataclasses
import json
from datetime import timedelta, tzinfo
import arrow
import requests
from codecarbon.core.schemas import (
EmissionCreate,
ExperimentCreate,
OrganizationCreate,
ProjectCreate,
RunCreate,
)
from codecarbon.external.logger import logger
# from codecarbon.output import EmissionsData
def get_datetime_with_timezone():
timestamp = str(arrow.now().isoformat())
return timestamp
class ApiClient: # (AsyncClient)
"""
This class call the Code Carbon API
"""
run_id = None
def __init__(
self,
endpoint_url="https://api.codecarbon.io",
experiment_id=None,
api_key=None,
access_token=None,
conf=None,
create_run_automatically=True,
):
"""
:endpoint_url: URL of the API endpoint
:experiment_id: ID of the experiment
:api_key: Code Carbon API_KEY
:access_token: Code Carbon API access token
:conf: Metadata of the experiment
:create_run_automatically: If False, do not create a run. To use API in read only mode.
"""
# super().__init__(base_url=endpoint_url) # (AsyncClient)
self.url = endpoint_url
self.experiment_id = experiment_id
self.api_key = api_key
self.conf = conf
self.access_token = access_token
if self.experiment_id is not None and create_run_automatically:
self._create_run(self.experiment_id)
def _get_headers(self):
headers = {"Content-Type": "application/json"}
if self.api_key:
# set the x-api-token header
headers["x-api-token"] = self.api_key
elif self.access_token:
headers["Authorization"] = f"Bearer {self.access_token}"
return headers
def set_access_token(self, token: str):
"""This method sets the access token to be used for the API.
Args:
token (str): access token to be used for the API
"""
self.access_token = token
def check_auth(self):
"""
Check API access to user account
"""
url = self.url + "/auth/check"
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def get_list_organizations(self):
"""
List all organizations
"""
url = self.url + "/organizations"
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def check_organization_exists(self, organization_name: str):
"""
Check if an organization exists
"""
organizations = self.get_list_organizations()
if organizations is None:
return False
for organization in organizations:
if organization["name"] == organization_name:
return organization
return False
def create_organization(self, organization: OrganizationCreate):
"""
Create an organization
"""
payload = dataclasses.asdict(organization)
url = self.url + "/organizations"
if organization := self.check_organization_exists(organization.name):
logger.warning(
f"Organization {organization['name']} already exists. Skipping creation."
)
return organization
else:
headers = self._get_headers()
r = requests.post(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 201:
self._log_error(url, payload, r)
return None
return r.json()
def get_organization(self, organization_id):
"""
Get an organization
"""
headers = self._get_headers()
url = self.url + "/organizations/" + organization_id
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def update_organization(self, organization: OrganizationCreate):
"""
Update an organization
"""
payload = dataclasses.asdict(organization)
headers = self._get_headers()
url = self.url + "/organizations/" + organization.id
r = requests.patch(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, payload, r)
return None
return r.json()
def list_projects_from_organization(self, organization_id):
"""
List all projects
"""
url = self.url + "/organizations/" + organization_id + "/projects"
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def create_project(self, project: ProjectCreate):
"""
Create a project
"""
payload = dataclasses.asdict(project)
url = self.url + "/projects"
headers = self._get_headers()
r = requests.post(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 201:
self._log_error(url, payload, r)
return None
return r.json()
def get_project(self, project_id):
"""
Get a project
"""
url = self.url + "/projects/" + project_id
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def add_emission(self, carbon_emission: dict):
assert self.experiment_id is not None
if self.run_id is None:
logger.warning(
"ApiClient.add_emission() need a run_id : the initial call may "
+ "have failed. Retrying..."
)
self._create_run(self.experiment_id)
if self.run_id is None:
logger.error(
"ApiClient.add_emission still no run_id, aborting for this time !"
)
return False
if carbon_emission["duration"] < 1:
logger.warning(
"ApiClient : emissions not sent because of a duration smaller than 1."
)
return False
emission = EmissionCreate(
timestamp=get_datetime_with_timezone(),
run_id=self.run_id,
duration=int(carbon_emission["duration"]),
emissions_sum=carbon_emission["emissions"],
emissions_rate=carbon_emission["emissions_rate"],
cpu_power=carbon_emission["cpu_power"],
gpu_power=carbon_emission["gpu_power"],
ram_power=carbon_emission["ram_power"],
cpu_energy=carbon_emission["cpu_energy"],
gpu_energy=carbon_emission["gpu_energy"],
ram_energy=carbon_emission["ram_energy"],
energy_consumed=carbon_emission["energy_consumed"],
)
try:
payload = dataclasses.asdict(emission)
url = self.url + "/emissions"
headers = self._get_headers()
r = requests.post(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 201:
self._log_error(url, payload, r)
return False
logger.debug(f"ApiClient - Successful upload emission {payload} to {url}")
except Exception as e:
logger.error(e, exc_info=True)
return False
return True
def _create_run(self, experiment_id: str):
"""
Create the experiment for project_id
"""
if self.experiment_id is None:
# TODO : raise an Exception ?
logger.error(
"ApiClient FATAL The ApiClient._create_run() needs an experiment_id !"
)
return None
try:
run = RunCreate(
timestamp=get_datetime_with_timezone(),
experiment_id=experiment_id,
os=self.conf.get("os"),
python_version=self.conf.get("python_version"),
codecarbon_version=self.conf.get("codecarbon_version"),
cpu_count=self.conf.get("cpu_count"),
cpu_model=self.conf.get("cpu_model"),
gpu_count=self.conf.get("gpu_count"),
gpu_model=self.conf.get("gpu_model"),
# Reduce precision for Privacy
longitude=round(self.conf.get("longitude", 0), 1),
latitude=round(self.conf.get("latitude", 0), 1),
region=self.conf.get("region"),
provider=self.conf.get("provider"),
ram_total_size=self.conf.get("ram_total_size"),
tracking_mode=self.conf.get("tracking_mode"),
)
payload = dataclasses.asdict(run)
url = self.url + "/runs"
headers = self._get_headers()
r = requests.post(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 201:
self._log_error(url, payload, r)
return None
self.run_id = r.json()["id"]
logger.info(
"ApiClient Successfully registered your run on the API.\n\n"
+ f"Run ID: {self.run_id}\n"
+ f"Experiment ID: {self.experiment_id}\n"
)
return self.run_id
except requests.exceptions.ConnectionError as e:
logger.error(
f"Failed to connect to API, please check the configuration. {e}",
exc_info=False,
)
except Exception as e:
logger.error(e, exc_info=True)
def list_experiments_from_project(self, project_id: str):
"""
List all experiments for a project
"""
url = self.url + "/projects/" + project_id + "/experiments"
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return []
return r.json()
def set_experiment(self, experiment_id: str):
"""
Set the experiment id
"""
self.experiment_id = experiment_id
def add_experiment(self, experiment: ExperimentCreate):
"""
Create an experiment, used by the CLI, not the package.
::experiment:: The experiment to create.
"""
payload = dataclasses.asdict(experiment)
url = self.url + "/experiments"
headers = self._get_headers()
r = requests.post(url=url, json=payload, timeout=2, headers=headers)
if r.status_code != 201:
self._log_error(url, payload, r)
return None
return r.json()
def get_experiment(self, experiment_id):
"""
Get an experiment by id
"""
url = self.url + "/experiments/" + experiment_id
headers = self._get_headers()
r = requests.get(url=url, timeout=2, headers=headers)
if r.status_code != 200:
self._log_error(url, {}, r)
return None
return r.json()
def _log_error(self, url, payload, response):
if len(payload) > 0:
logger.error(
f"ApiClient Error when calling the API on {url} with : {json.dumps(payload)}"
)
else:
logger.error(f"ApiClient Error when calling the API on {url}")
logger.error(
f"ApiClient API return http code {response.status_code} and answer : {response.text}"
)
def close_experiment(self):
"""
Tell the API that the experiment has ended.
"""
class simple_utc(tzinfo):
def tzname(self, **kwargs):
return "UTC"
def utcoffset(self, dt):
return timedelta(0)
|