Manage Calculations
Create a single calculation
This code creates a calculation (my Swagger). This is the same endpoint used to create a measurement. However, for calculations there are some more settings to consider:
- DatabaseUUID: retrieve by listing the time series databases.
- CollectorUUID: retrieve by listing the collectors and extracting the UUID of the Calculation Collector.
- Datatype: e.g. number, boolean, string ...
- Name: e.g. MyCalculation
- Status: e.g. Active, Paused ...
- Settings:
- Type: e.g sampled, raw or simulated
- Inputs: List of input measurements or calculations to retrieve from Manage Measurements (only when calculation type is raw or sampled):
- Type: Measurement or Calculation
- UUID: The UUID of the measurement or calculation
- Aggregation: The aggregation to use per calculation inteval e.g. last, first, mean, ... (only when calculation type is sampled)
- Script: The calculation script in which newline characters should be replaced with \n and tabs with \t.
- Interval: The calculation interval at which the calculation script is evaluated e.g. 5s (only when calculation type is sampled or simulated)
- IntervalOffset: The offset in seconds on the calculation interval e.g. 0s (only when calculation type is sampled or simulated)
- EvaluateInsufficientData: To evaluate the calculation script if one of the inputs has no data (insufficient data) e.g. true or false (only when calculation type is sampled)
- Attributes:
- Config:
- UoM: The unit of measure e.g. m
- LimitHi: The upper limit of the calculation
- LimitLo: The lower limit of the calculation
- ValueMax: The maximum value to be shown on the y-axis of a graph in Grafana using the Factry Historian datasource
- ValueMin: The minimum value to be shown on the y-axis of a graph in Grafana using the Factry Historian datasource
def create_calculation(
database_uuid: Optional[str],
collector_uuid: str,
datatype: str,
name: str,
status: str,
settings: Dict[str, Any],
description: Optional[str],
attributes: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
payload = {
"DatabaseUUID": database_uuid,
"CollectorUUID": collector_uuid,
"Datatype": datatype,
"Name": name,
"Status": status,
"Settings": settings,
"Description": description
}
if attributes is not None:
payload["Attributes"] = attributes
resp = session.post(f"{BASE_URL}/measurements", json=payload, timeout=60)
resp.raise_for_status()
return resp.json()
attributes = {
"Config": {
"UoM": "m",
"LimitHi": 3,
"LimitLo": 7,
"ValueMax": 0,
"ValueMin": 10
}
}
settings = {
"Type": "sampled",
"Inputs": {"_A": {
"Type": "Measurement",
"UUID": "my-input-measurement-uuid",
"Aggregation": "last"
}},
"Script": "set_tag(\"a\", \"a\")\ntest := rand.float() * 10.0\nreturn test * _A.Value",
"Interval": "5s",
"IntervalOffset": "0s",
"EvaluateInsufficientData": False
}
new_calculation = create_calculation("my-database-uuid", "my-calculation-collector-uuid", "number", "MyCalculation", "Active", settings, "", attributes)
print(json.dumps(new_calculation, indent=2))
Bulk create calculations
This code creates multiple calculations in one request (my Swagger). Note that the payload is an array of JSON objects, where the structure of each JSON object is identical to creating a single calculation.
def create_calculations(payload):
resp = session.put(f"{BASE_URL}/measurements", json=payload, timeout=60)
resp.raise_for_status()
return resp.json()
payload = [
{
"Attributes": {
"Config": {
"UoM": "a",
"LimitHi": 1,
"LimitLo": 1,
"ValueMax": 1,
"ValueMin": 1
}
},
"OrganizationUUID": "my-organization-uuid",
"CollectorUUID": "my-calculation-collector-uuid",
"DatabaseUUID": "my-database-uuid",
"Datatype": "number",
"Description": "This is a description",
"Name": "MyCalculation1",
"Settings": {
"Type": "sampled",
"Inputs": {"_A": {
"Type": "Measurement",
"UUID": "xxxxxxx-xxxx-xxxxxxxxxxxxxx",
"Aggregation": "last"
}},
"Script": "test := 6\nreturn rand.float() * 10.0 * _A.Value",
"Interval": "5s",
"IntervalOffset": "0s",
"EvaluateInsufficientData": False
},
"Status": "Active"
},
{
"Attributes": {
"Config": {
"UoM": "a",
"LimitHi": 1,
"LimitLo": 1,
"ValueMax": 1,
"ValueMin": 1
}
},
"OrganizationUUID": "my-organization-uuid",
"CollectorUUID": "my-calculation-collector-uuid",
"DatabaseUUID": "my-database-uuid",
"Datatype": "number",
"Description": "This is a description",
"Name": "MyCalculation2",
"Settings": {
"Type": "sampled",
"Inputs": {"_A": {
"Type": "Measurement",
"UUID": "my-input-measurement-uuid",
"Aggregation": "last"
}},
"Script": "test := 6\nreturn rand.float() * 10.0 * _A.Value",
"Interval": "5s",
"IntervalOffset": "0s",
"EvaluateInsufficientData": False
},
"Status": "Active"
}
]
create_calculations_result = create_calculations(payload)
print(json.dumps(create_calculations_result, indent=2))
For Historian versions below v8.0.0, the PUT https://historian.mycompany.com:18000/api/measurements endpoint for bulk updates is PATCH https://historian.mycompany.com:18000/api/measurements instead. In v8.0.0 and higher, PATCH https://historian.mycompany.com:18000/api/measurements still exists, but expects a different data structure. This is a breaking change as listed in the changelogchangelog.