Source code for nrgpy.api.catalog
try:
from nrgpy import logger
except ImportError:
pass
from datetime import datetime
import json
from .auth import nrg_api, data_catalog_url
import requests
[docs]class nrg_api_catalog(nrg_api):
"""Uses NRG hosted web-based API to catalog of available data in text format
To sign up for the service, go to https://services.nrgsystems.com/
Parameters
----------
serial_number : str or int
serial number of data logger (like, 820612345)
site_number : str or int
up to 6-digit site number
start_date : str
"YYYY-MM-DD HH:MM:SS" format, if just date it will return the whole day
times are in logger local time
end_date : str
"YYYY-MM-DD HH:MM:SS" format, if just date it will return the whole day
times are in logger local time
client_id : str
provided by NRG Systems
client_secret : str
provided by NRG Systems
Returns
-------
object
export object
Examples
--------
Check for available data files for site number 6
>>> import nrgpy
>>> client_id = "contact support@nrgsystems.com for access"
>>> client_secret = "contact support@nrgsystems.com for access"
>>> catalog = nrgpy.nrg_api_catalog(
client_id=client_id,
client_secret=client_secret,
site_number=6,
serial_number=820600019,
start_date="2020-05-01",
end_date="2020-05-03",
save_file=False
)
"""
def __init__(
self,
serial_number="",
site_number="",
start_date="2014-01-01",
end_date="2023-12-31",
client_id="",
client_secret="",
**kwargs
):
super().__init__(client_id, client_secret)
self.site_number = str(site_number).zfill(6)
self.serial_number = str(serial_number)[-5:]
self.site_number = str(site_number)
self.start_date = start_date
self.end_date = end_date
self.data_catalog()
[docs] def data_catalog(self):
import pandas as pd
self.headers = {"Authorization": "Bearer " + self.session_token}
self.data = {
"loggerserialnumber": self.serial_number,
"sitenumber": self.site_number,
"startdate": self.start_date,
"enddate": self.end_date,
}
self.request_time = datetime.now()
self.resp = requests.post(
data=self.data, url=data_catalog_url, headers=self.headers
)
self.request_duration = datetime.now() - self.request_time
if self.resp.status_code == 200:
self.f = json.loads(self.resp.content)
self.json = json.dumps(self.f, indent=2, sort_keys=True)
self.df = pd.read_json(self.json)
else:
print(self.resp.status_code)
print(self.resp.reason)
return False