API reference

datafarmclient.DatafarmClient

DatafarmClient(api_url)

Base DatafarmClient class, the main entrypoint to use the client.

ATTRIBUTE	DESCRIPTION
api_url	The instantiated api url.
timeseries	The timeseries subclass
variables	The variables subclass
entities	The entities subclass that you instantiate with an Entitytype

Examples:

>>> from datafarmclient import DatafarmClient
>>> client = DatafarmClient(api_url=api_url)
>>> client.login(api_key=api_key)
True

Base-class initializer.

PARAMETER	DESCRIPTION
api_url	The URL for which api server to use. TYPE: str

Source code in datafarmclient/base.py

def __init__(self, api_url: str):
    """Base-class initializer.

    Args:
        api_url: The URL for which api server to use.
    """
    self.api_url = self.validate_api_url(api_url)
    self.session = requests.Session()
    self.connected = False
    self.timeseries = datafarmclient.timeseries.TimeSeries(client=self)
    self.variables = datafarmclient.variables.Variables(client=self)
    self.entities = partial(datafarmclient.entities.Entities, client=self)

api_url instance-attribute

api_url = validate_api_url(api_url)

session instance-attribute

session = Session()

connected instance-attribute

connected = False

timeseries instance-attribute

timeseries = TimeSeries(client=self)

variables instance-attribute

variables = Variables(client=self)

entities instance-attribute

entities = partial(Entities, client=self)

time_series_source_descriptions cached property

time_series_source_descriptions

units cached property

units

time_series_types cached property

time_series_types

time_series_status cached property

time_series_status

qualities cached property

qualities

parameters cached property

parameters

medias cached property

medias

locations cached property

locations

quality_level_to_name cached property

quality_level_to_name

quality_name_to_level cached property

quality_name_to_level

validate_api_url staticmethod

validate_api_url(host_address)

Source code in datafarmclient/base.py

@staticmethod
def validate_api_url(host_address: str):
    response = requests.get(f"{host_address}/api/System/ServerTime")
    try:
        response.raise_for_status()
    except requests.exceptions.HTTPError:
        logging.error("No datafarm api found at that url")
    return f"{host_address}/api"

login

login(api_key)

Connect to the Datafarm API.

PARAMETER	DESCRIPTION
api_key	A valid API key for the class api url. TYPE: str

RETURNS	DESCRIPTION
bool	Whether or not you succesfully logged in.

Source code in datafarmclient/base.py

def login(self, api_key: str) -> bool:
    """Connect to the Datafarm API.

    Args:
        api_key: A valid API key for the class api url.

    Returns:
        Whether or not you succesfully logged in.
    """
    url = self.api_url + "/Login/Login"
    data = {"Token": api_key}
    try:
        response = self.session.post(url, json=data)
        response.raise_for_status()
    except requests.exceptions.HTTPError as err:
        reason = err.response.json()["error"]
        raise Exception(f"Connection attempt failed due to {reason}")
    else:
        try:
            access_token = response.headers["Access-Token"]
        except KeyError:
            logging.error("No Access-Token in response from server")
            return False
        else:
            self.session.headers.update({"Access-Token": access_token})
            self.connected = True
            logging.info("Succesfully logged in")
            return True

logoff

logoff()

Close the connection to the Datafarm API.

Source code in datafarmclient/base.py

def logoff(self) -> None:
    """Close the connection to the Datafarm API."""
    url = self.api_url + "/Login/Logoff"
    response = self.session.post(url)
    response.raise_for_status()
    self.session.headers.update({"Access-Token": ""})
    self.connected = False
    logging.info("Logged off")

list

list(class_id)

Source code in datafarmclient/base.py

@ensure_auth
def list(self, class_id: EntityType):
    return self.entities(class_id=class_id).list()

datafarmclient.Entities

Entities(class_id, client)

A Entitites subclass for the base class.

Entities need to be instantiated with an entitytype.

PARAMETER	DESCRIPTION
class_id	Entitytype specifying which entity to work with TYPE: EntityType
client	The base client allowing the subclass to use the shared session and client methods. TYPE: DatafarmClient

Source code in datafarmclient/entities/_entities.py

def __init__(self, class_id: EntityType, client: DatafarmClient):
    """Entities need to be instantiated with an entitytype.

    Args:
        class_id: Entitytype specifying which entity to work with
        client: The base client allowing the subclass to use
            the shared session and client methods.
    """
    self._client = client
    self.class_id = class_id

class_id instance-attribute

class_id = class_id

create

create(id_name, fields=None)

Create an entity.

PARAMETER	DESCRIPTION
id_name	The ID name of the new entity. TYPE: str
fields	The fields of the new entity. TYPE: Optional[Dict[str, str]] DEFAULT: None

RETURNS	DESCRIPTION
str	Id of the newly created entity

RAISES	DESCRIPTION
EntityExistsError	The entity already exists in the system
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def create(self, id_name: str, fields: Optional[Dict[str, str]] = None) -> str:
    """Create an entity.

    Args:
        id_name: The ID name of the new entity.
        fields: The fields of the new entity.

    Returns:
        Id of the newly created entity

    Raises:
        EntityExistsError: The entity already exists in the system
        HttpError: Requests threw an exception
    """
    field_names, field_values = self._get_field_names_and_values(fields)

    body = {
        "ClassID": self.class_id,
        "IDName": id_name,
        "FieldNames": field_names,
        "FieldValues": field_values,
    }
    url = self._client.api_url + "/Entities/Create"
    response = self._client.session.post(url, json=body)
    try:
        response.raise_for_status()
    except requests.HTTPError as err:
        if err.response.status_code == 409:
            raise EntityExistsError(
                f"Entity with ID {id_name!r} already exists in class {self.class_id!r}."
            )
        else:
            raise
    return response.json()["EntityID"]

exists

exists(id_name, fields=None)

Check if an entity exists or not

PARAMETER	DESCRIPTION
id_name	The ID name of the entity. TYPE: str
fields	The fields of the entity. TYPE: Optional[Dict[str, str]] DEFAULT: None

RETURNS	DESCRIPTION
bool	A boolean result based on whether or not an entity
bool	with that ID name already exists

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def exists(self, id_name: str, fields: Optional[Dict[str, str]] = None) -> bool:
    """Check if an entity exists or not

    Args:
        id_name: The ID name of the entity.
        fields: The fields of the entity.

    Returns:
        A boolean result based on whether or not an entity
        with that ID name already exists

    Raises:
        HttpError: Requests threw an exception
    """
    field_names, field_values = self._get_field_names_and_values(fields)
    body = {
        "ClassID": self.class_id,
        "IDName": id_name,
        "FieldNames": field_names,
        "FieldValues": field_values,
    }
    url = self._client.api_url + "/Entities/Exists"
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    if "EntityID" not in response.json():
        return False
    return True

get

get(id_name)

Get the ID of an entity.

PARAMETER	DESCRIPTION
id_name	The ID name of the entity. TYPE: str

RETURNS	DESCRIPTION
str	The id of the entity

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def get(
    self,
    id_name: str,
) -> str:
    """Get the ID of an entity.

    Args:
        id_name: The ID name of the entity.

    Returns:
        The id of the entity

    Raises:
        HttpError: Requests threw an exception
    """

    body = {
        "ClassID": self.class_id,
        "IDName": id_name,
    }
    url = self._client.api_url + "/Entities/Exists"
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    if "EntityID" not in response.json():
        raise EntityNotFoundError("Entity does not exist")
    return response.json()["EntityID"]

update

update(entity_id, fields=None)

Update an entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity. TYPE: str
fields	The fields to update of the entity. TYPE: Optional[Dict[str, str]] DEFAULT: None

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def update(
    self,
    entity_id: str,
    fields: Optional[Dict[str, str]] = None,
) -> requests.Response:
    """Update an entity.

    Args:
        entity_id: The ID of the entity.
        fields: The fields to update of the entity.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    field_names, field_values = self._get_field_names_and_values(fields)

    body = {
        "ClassID": self.class_id,
        "EntityID": entity_id,
        "FieldNames": field_names,
        "FieldValues": field_values,
    }
    url = self._client.api_url + "/Entities/Update"
    response = self._client.session.post(url, json=body)
    try:
        response.raise_for_status()
    except requests.HTTPError as err:
        if err.response.status_code == 404:
            raise EntityNotFoundError(
                f"Entity with ID {entity_id!r} not found in class {self.class_id!r}."
            )
        else:
            raise
    return response.json()

delete

delete(entity_id)

Delete an entity.

PARAMETER	DESCRIPTION
entity_id	The ID of the entity to delete. TYPE: str

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def delete(self, entity_id: str) -> requests.Response:
    """Delete an entity.

    Args:
        entity_id: The ID of the entity to delete.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """

    body = {"ClassID": self.class_id, "EntityID": entity_id}
    url = self._client.api_url + "/Entities/Delete"
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    return response

metadata

metadata()

Get the metadata of an EntityType.

RETURNS	DESCRIPTION
DataFrame	Pandas DataFrame containing the results

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def metadata(self) -> pd.DataFrame:
    """Get the metadata of an EntityType.

    Returns:
        Pandas DataFrame containing the results

    Raises:
        HttpError: Requests threw an exception
    """

    body = {"ClassID": self.class_id}
    url = f"{self._client.api_url}/Entities/MetaData"
    response = self._client.session.post(url, json=body)
    response.raise_for_status()

    return json_to_pandas(json.dumps(response.json()))

list

list(fields=None)

List all entities of the chosen type.

RETURNS	DESCRIPTION
DataFrame	Pandas DataFrame containing the results

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/entities/_entities.py

@ensure_auth
def list(self, fields: Optional[List[str]] = None) -> pd.DataFrame:
    """List all entities of the chosen type.

    Returns:
        Pandas DataFrame containing the results

    Raises:
        HttpError: Requests threw an exception
    """
    url = self._client.api_url + "/Entities/List"
    body = {
        "ClassID": self.class_id,
        "Fields": [] if not fields else fields,
    }
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    data = response.json()

    return json_to_pandas(json.dumps(data))

datafarmclient.TimeSeries

TimeSeries(client)

A TimeSeries subclass for the base class.

PARAMETER	DESCRIPTION
client	The base client allowing the subclass to use the shared session and client methods. TYPE: DatafarmClient

Source code in datafarmclient/timeseries/_timeseries.py

def __init__(self, client: DatafarmClient):
    """
    Args:
        client: The base client allowing the subclass to use
            the shared session and client methods.
    """
    self._client = client

get_data

get_data(
    time_series_id,
    start=datetime(1900, 1, 1, 0, 0, 0, 0),
    end=datetime(2100, 1, 1, 0, 0, 0, 0),
    fields=None,
    qualities=None,
    aggregation=None,
    limit=0,
    ascending=True,
)

Get data from Datafarm.

PARAMETER	DESCRIPTION
time_series_id	The time series to get data from. TYPE: Union[str, List[str]]
start	The start of the range to get data from. TYPE: DateTime DEFAULT: datetime(1900, 1, 1, 0, 0, 0, 0)
end	The end of the range to get data from. TYPE: DateTime DEFAULT: datetime(2100, 1, 1, 0, 0, 0, 0)
fields	fields/columns to return TYPE: Optional[List[Fields]] DEFAULT: None
qualities	Filter the data by qualities. TYPE: Optional[List[str]] DEFAULT: None
aggregation	Complex data structure that gets constructed with create_aggregation() or create_resample(). TYPE: Optional[TAggregation] DEFAULT: None
limit	The maximum number of rows to return. TYPE: int DEFAULT: 0
ascending	Whether to sort the data in ascending order. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
List[DataFrame]	List of dataframes.

RAISES	DESCRIPTION
EntityNotFoundError	An error occurs if no timeseries are found.

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def get_data(
    self,
    time_series_id: Union[str, List[str]],
    start: DateTime = datetime.datetime(1900, 1, 1, 0, 0, 0, 0),
    end: DateTime = datetime.datetime(2100, 1, 1, 0, 0, 0, 0),
    fields: Optional[List[Fields]] = None,
    qualities: Optional[List[str]] = None,
    aggregation: Optional[TAggregation] = None,
    limit: int = 0,
    ascending: bool = True,
) -> List[pd.DataFrame]:
    """Get data from Datafarm.

    Args:
        time_series_id: The time series to get data from.
        start: The start of the range to get data from.
        end: The end of the range to get data from.
        fields: fields/columns to return
        qualities: Filter the data by qualities.
        aggregation: Complex data structure that gets constructed with
            create_aggregation() or create_resample().
        limit: The maximum number of rows to return.
        ascending: Whether to sort the data in ascending order.

    Returns:
        List of dataframes.

    Raises:
        EntityNotFoundError: An error occurs if no timeseries are found.
    """
    start = format_datetime(start)
    end = format_datetime(end)
    qualities = qualities or []
    fields = fields or []
    aggregation = aggregation.model_dump(mode="json") if aggregation else {}
    sort_order = "soAscending" if ascending else "soDescending"
    if isinstance(time_series_id, str):
        time_series_id = [time_series_id]

    url = self._client.api_url + "/TimeSeries/ExtractData"
    body = {
        "TimeSeries": time_series_id,
        "ISO8601_TimeStamp": False,
        "LimitRowCount": limit,
        "Qualities": qualities,
        "RangeEnd": end,
        "RangeStart": start,
        "SortOrder": sort_order,
        "Fields": fields,
        "Aggregation": aggregation,
    }
    response = self._client.session.post(url, json=body)
    response.raise_for_status()

    if "error" in response.json():
        raise SystemError(response.json()["error"])    
    else:
        datasets = response.json()["datasets"]
        if datasets:
            resp = []
            for dataset in datasets:
                resp.append(json_to_pandas(json.dumps(dataset)))
            return resp
        else:
            raise EntityNotFoundError("No timeseries found for that request")

delete_data

delete_data(
    time_series_id, timestamps=None, start=None, end=None
)

Delete data from a time series. Either timestamps or start and end must be provided.

PARAMETER	DESCRIPTION
time_series_id	The time series to delete data from. TYPE: str
timestamps	The timestamps to delete. TYPE: Optional[List[DateTime]] DEFAULT: None
start	The start of the range to delete data from. TYPE: Optional[DateTime] DEFAULT: None
end	The end of the range to delete data from. Note that this is NOT inclusive. TYPE: Optional[DateTime] DEFAULT: None

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def delete_data(
    self,
    time_series_id: str,
    timestamps: Optional[List[DateTime]] = None,
    start: Optional[DateTime] = None,
    end: Optional[DateTime] = None,
) -> requests.Response:
    """
    Delete data from a time series.
    Either timestamps or start and end must be provided.

    Args:
        time_series_id: The time series to delete data from.
        timestamps: The timestamps to delete.
        start: The start of the range to delete data from.
        end: The end of the range to delete data from.
            Note that this is NOT inclusive.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    if timestamps is None and (start is None or end is None):
        raise ValueError("Either timestamps or start and end must be provided.")
    if timestamps is not None and (start is not None or end is not None):
        raise ValueError("Either timestamps or start and end must be provided.")
    if timestamps is not None:
        return self._delete_data_timestamps(time_series_id, timestamps)

    return self._delete_data_range(time_series_id, start, end)

update_data_quality

update_data_quality(time_series_id, qualities)

Update the quality of data in a time series.

PARAMETER	DESCRIPTION
time_series_id	The time series to update. TYPE: str
qualities	The qualities to set. TYPE: Series

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def update_data_quality(
    self,
    time_series_id: str,
    qualities: pd.Series,
) -> requests.Response:
    """Update the quality of data in a time series.

    Args:
        time_series_id: The time series to update.
        qualities: The qualities to set.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    try:
        new_qualities = [
            int(self._client.quality_name_to_level[q])
            if isinstance(q, str)
            else int(q)
            for q in qualities
        ]
    except KeyError:
        raise QualityError(
            "Quality must be one of: {}".format(
                ", ".join(self._client.quality_name_to_level.keys())
            )
        )

    endpoint = "/TimeSeries/UpdateDataQuality"
    url = self._client.api_url + endpoint
    body = {
        "TimeSeriesName": time_series_id,
        "TimeStamp": [format_datetime(ts) for ts in list(qualities.index)],
        "QualityLevel": new_qualities,
    }
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    return response

update_data_quality_range

update_data_quality_range(
    time_series_id, quality, start, end
)

Update the quality of data in a range in a time series.

PARAMETER	DESCRIPTION
time_series_id	The time series to update. TYPE: str
start	The start of the range to update data quality from. TYPE: DateTime
end	The end of the range to update data quality from. Note that this is NOT inclusive. TYPE: DateTime
quality	The quality to set. TYPE: Union[int, str]

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def update_data_quality_range(
    self,
    time_series_id: str,
    quality: Union[int, str],
    start: DateTime,
    end: DateTime,
) -> requests.Response:
    """Update the quality of data in a range in a time series.

    Args:
        time_series_id: The time series to update.
        start: The start of the range to update data quality from.
        end: The end of the range to update data quality from.
            Note that this is NOT inclusive.
        quality: The quality to set.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    if isinstance(quality, str):
        try:
            new_quality = int(self._client.quality_name_to_level[quality])
        except KeyError:
            raise QualityError(
                "Quality must be one of: {}".format(
                    ", ".join(self._client.quality_name_to_level.keys())
                )
            )
    else:
        new_quality = quality

    endpoint = "/TimeSeries/UpdateDataQualityRange"
    url = self._client.api_url + endpoint
    body = {
        "TimeSeriesName": time_series_id,
        "RangeStart": format_datetime(start),
        "RangeFinish": format_datetime(end),
        "QualityLevel": new_quality,
    }
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    return response

insert_data

insert_data(
    time_series_id,
    data,
    bulk_insert=False,
    values_col="Data",
    qualities_col="QualityTxt",
    confidence_col="Confidence",
    duration_col="Duration",
    filepath_col="FilePath",
)

Insert data into a time series.

It is assumed that the dataframe has a DateTimeIndex and not a TimeStamp column

PARAMETER	DESCRIPTION
time_series_id	The time series to update. TYPE: str
data	The dataframe containing the data to insert. TYPE: DataFrame
bulk_insert	bool = False, TYPE: bool DEFAULT: False
values_col	str = "Data", TYPE: str DEFAULT: 'Data'
qualities_col	str = "QualityTxt", TYPE: str DEFAULT: 'QualityTxt'
confidence_col	str = "Confidence", TYPE: str DEFAULT: 'Confidence'
duration_col	str = "Duration", TYPE: str DEFAULT: 'Duration'
filepath_col	str = "FilePath", TYPE: str DEFAULT: 'FilePath'

RETURNS	DESCRIPTION
Response	Response from the API

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def insert_data(
    self,
    time_series_id: str,
    data: pd.DataFrame,
    bulk_insert: bool = False,
    values_col: str = "Data",
    qualities_col: str = "QualityTxt",
    confidence_col: str = "Confidence",
    duration_col: str = "Duration",
    filepath_col: str = "FilePath",
) -> requests.Response:
    """Insert data into a time series.

    It is assumed that the dataframe has a DateTimeIndex
    and not a TimeStamp column

    Args:
        time_series_id: The time series to update.
        data: The dataframe containing the data to insert.
        bulk_insert: bool = False,
        values_col: str = "Data",
        qualities_col: str = "QualityTxt",
        confidence_col: str = "Confidence",
        duration_col: str = "Duration",
        filepath_col: str = "FilePath",

    Returns:
        Response from the API

    Raises:
        HttpError: Requests threw an exception
    """
    if not isinstance(data.index, pd.DatetimeIndex):
        raise Exception("Dataframe missing datetimeindex")
    data = self._format_dataframe(
        data,
        values_col=values_col,
        qualities_col=qualities_col,
        confidence_col=confidence_col,
        duration_col=duration_col,
        filepath_col=filepath_col,
    )
    body = self._get_insert_data_body(
        time_series_id,
        data,
        bulk_insert,
    )
    endpoint = "/TimeSeries/InsertData"
    url = self._client.api_url + endpoint
    try:
        response = self._client.session.post(url, json=body)
        response.raise_for_status()
    except requests.HTTPError as e:
        raise Exception(e.response.text)
    return response

statistics

statistics(time_series_id)

Get statistics for a time series or a list of time series.

PARAMETER	DESCRIPTION
time_series_id	The time series to get statistics for. TYPE: Union[str, List[str]]

RETURNS	DESCRIPTION
DataFrame	Pandas dataframe with statistics about the timeseries dataset.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def statistics(self, time_series_id: Union[str, List[str]]) -> pd.DataFrame:
    """Get statistics for a time series or a list of time series.

    Args:
        time_series_id: The time series to get statistics for.

    Returns:
        Pandas dataframe with statistics about the timeseries dataset.

    Raises:
        HttpError: Requests threw an exception
    """
    url = self._client.api_url + "/TimeSeries/Statistics"
    if isinstance(time_series_id, str):
        time_series_id = [time_series_id]
    body = {"TimeSeries": list(time_series_id), "ISO8601_Timestamp": True}
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    data = response.json()

    return json_to_pandas(json.dumps(data))

list

list()

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def list(self) -> pd.DataFrame:
    return self._client.entities("enTimeSeries").list()

latest_values

latest_values(time_series_id, before=None)

Get data from Datafarm.

PARAMETER	DESCRIPTION
time_series_id	The time series to get data from. TYPE: Union[str, List[str]]
before	Get values from before this timestamp. TYPE: Optional[DateTime] DEFAULT: None

Returns List of dataframes

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/timeseries/_timeseries.py

@ensure_auth
def latest_values(
    self,
    time_series_id: Union[str, List[str]],
    before: Optional[DateTime] = None,
) -> List[pd.DataFrame]:
    """Get data from Datafarm.

    Args:
        time_series_id: The time series to get data from.
        before: Get values from before this timestamp.

    Returns
        List of dataframes

    Raises:
        HttpError: Requests threw an exception
    """
    if not before:
        before = datetime.datetime.now()
    before = format_datetime(before)
    if isinstance(time_series_id, str):
        time_series_id = [time_series_id]

    url = self._client.api_url + "/TimeSeries/LatestValues"
    body = {
        "TimeSeries": time_series_id,
        "ISO8601_TimeStamp": False,
        "Before": before,
    }
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    data = response.json()

    return json_to_pandas(json.dumps(data))

create_resample staticmethod

create_resample(
    *,
    method,
    repetition_type,
    period="apHour",
    period_offset=0,
    user_def_period=0,
    user_def_time_ref=now(),
)

Helper function for creating a timeseries resampling object.

PARAMETER	DESCRIPTION
method	Choose from 2 different resampling methods. TYPE: Literal['interpolation', 'repetition']
repetition_type	Stairs or inverted stairs. TYPE: Literal['stairs', 'inverted_stairs']
period	Aggregation period to use. TYPE: TAggregationPeriod DEFAULT: 'apHour'
period_offset	An optional offset to use for the period. This is ignored for period=apUserDefinedSeconds TYPE: float DEFAULT: 0
user_def_period	If period=apUserDefinedSeconds then this parameter is used TYPE: int DEFAULT: 0
user_def_time_ref	If period=apUserDefinedSeconds then this parameter is used TYPE: DateTime DEFAULT: now()

Returns TAggregation

Source code in datafarmclient/timeseries/_timeseries.py

@staticmethod
def create_resample(
    *,
    method: Literal["interpolation", "repetition"],
    repetition_type: Literal["stairs", "inverted_stairs"],
    period: TAggregationPeriod = "apHour",
    period_offset: float = 0,
    user_def_period: int = 0,
    user_def_time_ref: DateTime = datetime.datetime.now(),
) -> TAggregation:
    """Helper function for creating a timeseries resampling object.

    Args:
        method: Choose from 2 different resampling methods.
        repetition_type: Stairs or inverted stairs.
        period: Aggregation period to use.
        period_offset: An optional offset to use for the period.
            This is ignored for period=apUserDefinedSeconds
        user_def_period: If period=apUserDefinedSeconds then this parameter is used
        user_def_time_ref: If period=apUserDefinedSeconds then this parameter is used

    Returns
        TAggregation
    """
    method = "fmInterpolation" if method == "interpolation" else "fmRepetition"
    rep_type = "rtStairs" if repetition_type == "stairs" else "rtInvertedStairs"
    resampling = TGapFill(Enabled=True, Method=method, RepetitionType=rep_type)
    agg = TAggregation(
        Method="amResample",
        MethodConstant=0,
        Period=period,
        PeriodOffset=period_offset,
        Resampling=resampling,
        UserDefPeriod=user_def_period,
        UserDefTimeRef=user_def_time_ref,
    )
    return agg

create_aggregation staticmethod

create_aggregation(
    *,
    method,
    period,
    method_constant=0,
    direction="pdStraight",
    period_offset=0,
    user_def_period=0,
    user_def_time_ref=now(),
)

Helper function for creating a timeseries aggregation object.

PARAMETER	DESCRIPTION
method	Choose from 10 different aggregation methods. TYPE: TAggregationMethod
period	Aggregation period to use. TYPE: TAggregationPeriod
period_offset	An optional offset to use for the period. This is ignored for period=apUserDefinedSeconds TYPE: float DEFAULT: 0
direction	How to do the aggregation. TYPE: TPeriodDirection DEFAULT: 'pdStraight'
user_def_period	If period=apUserDefinedSeconds then this parameter is used. TYPE: int DEFAULT: 0
user_def_time_ref	If period=apUserDefinedSeconds then this parameter is used. TYPE: DateTime DEFAULT: now()

Returns TAggregation

Source code in datafarmclient/timeseries/_timeseries.py

@staticmethod
def create_aggregation(
    *,
    method: TAggregationMethod,
    period: TAggregationPeriod,
    method_constant: float = 0,
    direction: TPeriodDirection = "pdStraight",
    period_offset: float = 0,
    user_def_period: int = 0,
    user_def_time_ref: DateTime = datetime.datetime.now(),
) -> TAggregation:
    """Helper function for creating a timeseries aggregation object.

    Args:
        method: Choose from 10 different aggregation methods.
        period: Aggregation period to use.
        period_offset: An optional offset to use for the period.
            This is ignored for period=apUserDefinedSeconds
        direction: How to do the aggregation.
        user_def_period: If period=apUserDefinedSeconds then this parameter is used.
        user_def_time_ref: If period=apUserDefinedSeconds then this parameter is used.

     Returns
         TAggregation
    """
    resampling = TGapFill(
        Enabled=False, Method="fmInterpolation", RepetitionType="rtStairs"
    )
    agg = TAggregation(
        Direction=direction,
        Method=method,
        MethodConstant=method_constant,
        Period=period,
        PeriodOffset=period_offset,
        Resampling=resampling,
        UserDefPeriod=user_def_period,
        UserDefTimeRef=user_def_time_ref,
    )
    return agg

datafarmclient.Variables

Variables(client)

A Variables subclass for the base class.

PARAMETER	DESCRIPTION
client	The base client allowing the subclass to use the shared session and client methods. TYPE: DatafarmClient

Source code in datafarmclient/variables/_variables.py

def __init__(self, client: DatafarmClient):
    """
    Args:
        client: The base client allowing the subclass to use
            the shared session and client methods.
    """
    self._client = client

get_object_variable

get_object_variable(
    name, output_dir=None, output_path=None
)

Get an object variable.

PARAMETER	DESCRIPTION
name	The name of the variable. TYPE: str
output_dir	The directory to save the object to. The file name will be read from the system. Defaults to None. TYPE: Optional[str] DEFAULT: None
output_path	The path to save the object to. Defaults to None. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Union[str, bytes]	If neither output_dir nor output_path are specified,
Union[str, bytes]	the object is returned as bytes.
Union[str, bytes]	Otherwise, the object is saved to the specified path.

RAISES	DESCRIPTION
HttpError	Requests threw an exception.
VariableNotFoundError	If specificed variable doesn't exists.
ValueError	If variable is not an object.

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def get_object_variable(
    self,
    name: str,
    output_dir: Optional[str] = None,
    output_path: Optional[str] = None,
) -> Union[str, bytes]:
    """Get an object variable.

    Args:
        name: The name of the variable.
        output_dir: The directory to save the object to.
            The file name will be read from the system.
            Defaults to None.
        output_path: The path to save the object to.
            Defaults to None.

    Returns:
        If neither `output_dir` nor `output_path` are specified,
        the object is returned as bytes.
        Otherwise, the object is saved to the specified path.

    Raises:
        HttpError: Requests threw an exception.
        VariableNotFoundError: If specificed variable doesn't exists.
        ValueError: If variable is not an object.
    """
    body = {"Name": name}
    endpoint = "/Variables/GetValue"
    url = self._client.api_url + endpoint
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    response_body = response.json()
    if "VariableType" not in response_body:
        raise VariableNotFoundError("Variable does not exist")
    variable_type = response_body["VariableType"]
    if variable_type != "vtObject":
        raise ValueError("Variable is not an object")
    base64_data = response_body["ObjectBase64"]

    if output_dir is None and output_path is None:
        return base64.b64decode(base64_data)
    else:
        output_path = output_path or os.path.join(
            output_dir, response_body["ObjectFileName"]
        )
        with open(output_path, "wb") as f:
            f.write(base64.b64decode(base64_data))
        return output_path

set_object_variable

set_object_variable(name, file_path)

Set an object in the session.

PARAMETER	DESCRIPTION
name	The name of the object. TYPE: str
file_path	The file path of the object. TYPE: str

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception.

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def set_object_variable(self, name: str, file_path: str) -> requests.Response:
    """Set an object in the session.

    Args:
        name: The name of the object.
        file_path: The file path of the object.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception.
    """
    object_file_name = os.path.basename(file_path)
    object_base64 = base64.b64encode(open(file_path, "rb").read()).decode("utf-8")
    body = {
        "Name": name,
        "ObjectBase64": object_base64,
        "ObjectFileName": object_file_name,
        "VariableType": "vtObject",
    }
    endpoint = "/Variables/SetValue"
    url = self._client.api_url + endpoint
    response = self._client.session.post(url, json=body)
    response.raise_for_status()
    return response

create_category

create_category(name)

Create a new variable category.

PARAMETER	DESCRIPTION
name	The name of the category. TYPE: str

RETURNS	DESCRIPTION
str	Id of the newly created variable category.

RAISES	DESCRIPTION
EntityExistsError	The entity already exists in the system.
HttpError	Requests threw an exception.

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def create_category(self, name: str) -> str:
    """Create a new variable category.

    Args:
        name: The name of the category.

    Returns:
        Id of the newly created variable category.

    Raises:
        EntityExistsError: The entity already exists in the system.
        HttpError: Requests threw an exception.
    """
    return self._client.entities("enGlobalVariableCategory").create(id_name=name)

create_type

create_type(name)

Create a new variable type.

PARAMETER	DESCRIPTION
name	The name of the type. TYPE: str

RETURNS	DESCRIPTION
str	Id of the newly created variable type.

RAISES	DESCRIPTION
EntityExistsError	The entity already exists in the system.
HttpError	Requests threw an exception.

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def create_type(self, name: str) -> str:
    """Create a new variable type.

    Args:
        name: The name of the type.

    Returns:
        Id of the newly created variable type.

    Raises:
        EntityExistsError: The entity already exists in the system.
        HttpError: Requests threw an exception.
    """
    return self._client.entities("enGlobalVariableType").create(id_name=name)

delete_category

delete_category(name)

Delete a variable category

PARAMETER	DESCRIPTION
name	The name of the category. TYPE: str

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def delete_category(self, name: str) -> requests.Response:
    """Delete a variable category

    Args:
        name: The name of the category.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    entity_id = self._client.entities("enGlobalVariableCategory").get(id_name=name)
    return self._client.entities("enGlobalVariableCategory").delete(
        entity_id=entity_id
    )

delete_type

delete_type(name)

Delete a variable type

PARAMETER	DESCRIPTION
name	The name of the type. TYPE: str

RETURNS	DESCRIPTION
Response	Response from the API.

RAISES	DESCRIPTION
HttpError	Requests threw an exception

Source code in datafarmclient/variables/_variables.py

@ensure_auth
def delete_type(self, name: str) -> requests.Response:
    """Delete a variable type

    Args:
        name: The name of the type.

    Returns:
        Response from the API.

    Raises:
        HttpError: Requests threw an exception
    """
    entity_id = self._client.entities("enGlobalVariableType").get(id_name=name)
    return self._client.entities("enGlobalVariableType").delete(entity_id=entity_id)

Schemas

EntityType

• enTimeSeries
• enTimeSeriesQuality
• enTimeSeriesDatasourceDescription
• enTimeSeriesType
• enTimeSeriesStatus
• enTimeSeriesUnit
• enTimeSeriesMedia
• enTimeSeriesParameter
• enLocation
• enGlobalVariable
• enGlobalVariableCategory
• enGlobalVariableType