Getting Started with Historian API: Python Client Tutorial

Introduction

In this guide, we will use the Factry Historian API for querying data using Python.

Prerequisites

Before you begin this guide you’ll need the following to be installed:

• python
• requests package
• json package

You can install the required packages using the following commands (Make sure you have Python and pip installed).

pip install requests
pip install json

An API token is required to be able to query data. The token can be created as can be seen here . This token will have the same privileges as the user.

Imports and Global Variables

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Import required packages.
import requests
import json

 headers = {
      "Content-Type": "application/json",
      "x-organization-uuid": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
      "accept": "application/json",
      "Authorization": "<Token Type TOKEN>"
  }

Query Time Series Data

Here, we query the time series data using following parameters:

Swagger: https://historian.mycompany.com:8000/api/swagger/#/timeseries/queryTimeseries

Note

Replace historian.mycompany.com:8000 with the url to your Factry Historian webpage.

The documentation for all possible parameters to perform the query can be found by clicking on Model.

Description

• Aggregation [optional]: Defines an aggregation function for time series data. An aggregate function performs a calculation on a set of values, and returns a single value.
• Desc [optional] (boolean): Lists the time series data according to their timestamps. Desc: false will result in a list that has ascending timestamps. Desc: true will result in a list of data points that is ordered according to descending timestamps.
• End [optional] (string): Defines the end time of the query (not inclusive).
• GroupBy [optional] (string): An array of tags to group by.
  • Example: status will group the datapoints according to their status.
• Join [optional] (boolean): Joins the results based on their timestamps and fills in null values to provide a data point for every timestamp.
• Limit [optional] (integer): Defines the maximum data points that will be returned per measurement.
  • Example: Limit: 10 will return 10 data points within the given time range for each measurement UUID. Limit: 0 defines no limit.
• MeasurementUUIDs [required]: The UUID of the measurement(s).
• Measurements [required]: Defines the measurement(s) by their database name and measurement name.
  • Database (string): Name of the database.
  • Measurement (string): Name of the measurement.

MeasurementUUID and Measurements can be used interchangeably. Defining one of these is enough for indicating the measurement(s).

• Start [required] (string): Defines the initial timestamp of time series data (inclusive).
• Tags [optional]: Defines tags for data points.
  • Example: Tags: {"status": "Good"} will return data points with “Good” status.

Example Script

Define a function which queries time series data and returns a JSON response. Either measurement name and database name or measurement UUID can be used for querying time series data.

 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

def query_ts_data(headers, url, start_time, stop_time, limit, measurementUUIDs, measurements):
    if len(measurementUUIDs) != 0:
        payload = {
            "End": stop_time,
            "Limit": limit,
            "MeasurementUUIDs": measurementUUIDs,
            "Offset": 0,
            "Start": start_time
        }

    if len(measurements) != 0:
        payload = {
            "End": stop_time,
            "Limit": limit,
            "Measurements": measurements,
            "Offset": 0,
            "Start": start_time
        }

    response = requests.request("POST", url, json=payload, headers=headers)
    return response.json()

# Call the function with measurement UUIDs.
ts_query_result_1 = query_ts_data(headers, "https://historian.mycompany.com:8000/api/timeseries/query", "2023-10-16T09:08:57.732Z", "2023-10-16T09:08:57.732Z", 10, ["xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"], [])

# Call the function with measurement names.
ts_query_result_2 = query_ts_data(headers, "https://historian.mycompany.com:8000/api/timeseries/query", "2023-10-09T09:08:57.732Z", "2023-10-14T09:08:57.732Z", 10, [], [{"Database":"MyDatabase", "Measurement":"MyMeasurement"}])

Get a List of Time Series Databases

Here, we get a list of time series databases for the organization. The organization UUID is required in the header field in order to perform this API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/timeseries-databases/getTimeseriesDatabases

Example Script

Define a function that returns a list of time series databases in JSON format given the organization UUID.

1
2
3
4
5
6

def get_ts_databases(headers, url):
  response = requests.request("GET", url, data="", headers=headers)
  return response.json()

# Call the function.
ts_db_list_result = get_ts_databases(headers, "https://historian.mycompany.com:8000/api/timeseries-databases")

Get All Collectors in an Organization

Here, we get a list of all the collectors in an organization. The organization UUID is required in the header field in order to perform this API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/collectors/getCollectors

Example Script

Define a function that returns a list of collectors in JSON format given the organization UUID.

1
2
3
4
5
6

def get_collectors(headers, url):
  response = requests.request("GET", url, data="", headers=headers)
  return response.json()

# Call the function.
collector_list_result = get_collectors(headers, "https://historian.mycompany.com:8000/api/collectors")

Get All Measurements With and Without a Filter

Measurements can be fetched with or without filters. Various filters can be used for filtering the measurements.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/measurements/getMeasurements

Available filter options can be found in Parameters.

Some filters are as follows:

Description

• Keyword (string): The pattern that is used to filter the measurements. This will match the measurements names that contain the given keyword.
  • Example: “open” will return measurements that contains “open” in their name.
• Statuses: Status of the measurement. Multiple statuses can be provided so that when matching one of the given statuses, this data is included.
  • Example: “Active” will return measurements with Status = Active.
• WithBadQualityOnly (boolean): Filtering of the measurements with bad quality. True will result with measurements that have bad quality. False will return all measurements regardless of their quality.
• ExcludeCalculations (boolean): Excludes calculations from returned result.

These filters are added on the URL of the API.

• Example: https://historian.mycompany.com:8000/api/measurements?Keyword=open&WithBadQualityOnly=true&Statuses[0]=Paused&Statuses[1]=Active

This URL will return all the measurements that has “open” in their name and has bad quality and have status “Paused” or status “Active”.

If no filters are defined, all measurements will be returned.

Example Script

Define a function that returns a list of measurements in JSON format given the organization UUID.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def get_measurements(headers, url, parameters=None):
  # If filters are provided, construct the URL with them
  if parameters:
    url += "?" + "&".join(f"{key}={value}" for key, value in parameters.items())

  response = requests.request("GET", url, data="", headers=headers)
  return response.json()

# Define the parameters as a dictionary
filter_parameters = {
    "Keyword": "open",
    "WithBadQualityOnly": "true",
    "Statuses[0]": "Paused",
    "Statuses[1]": "Active"
}

# Call the function.
measurement_list_result = get_measurements(headers, "https://historian.mycompany.com:8000/api/measurements", filter_parameters)

Creating a Simple CSV File Collector

In this section, the setup of a collector will be explained.

Create Collector

Create a collector given an organization UUID. The organization UUID is required in the header field in order to perform this API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/collectors/createCollector

A CSV collector is used as an example for this section. CSV collector is a collector that creates data points from parsing CSV files.

The CSV collector expects CSV files with 3 columns where columns contain:

• name of the measurement,
• a timestamp,
• a value

The collector then checks the measurement name from the first column and writes the point to that measurement. An example CSV data can be given as follows:

Description

• BuildVersion [optional] (string): Version number of the collector software. Build version will be set when registering the collector.

  • Example: “1.0.0”

  Note
  Required for registering a collector.

• CollectorType [required] (string): The type of the collector. Collector type will be set when registering the collector.

  • Example: “OPC-UA”

  Note
  Required for registering a collector.

• DefaultDatabaseUUID [optional] (string): UUID of the default database.

• Description [optional] (string): Brief explanation given to the collector.

  • Example: “This is my collector”.

• Health [required] (dict): Health of the collector.

  • Example:

    1 2 3 4 5

    "Health": { "CollectorUUID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", "Health": "Collecting", "Timestamp": "2023-12-28T14:13:46.657Z" }

• Name [required] (string): Name assigned to the collector.

  • Note: Name of the collector can be changed afterwards. However, this is not the case for the collector UUID.

• Status [required] (string): Current status of the collector. Status will be set to Initial when creating a collector.

  • Example: “Active”

Settings and Setting Schemas

JSON collector schema is used to validate the settings configured for the collector.

JSON Schema Documentation .

• Settings [optional] (dict): Collector settings.

  • Example:

    1 2 3

    "Settings": { "IncomingDirectoryPath": "/incoming/directory/path" }

• SettingsSchema [optional] (dict): JSON Schema to define the structure of the collector settings.

  • Example:

    1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

    "SettingsSchema": { "$schema": "http://json-schema.org/draft-06/schema#", "definitions": {}, "id": "http://factry.io/collectorsettings.json", "properties": { "IncomingDirectoryPath": { "title": "IncomingDirectoryPath", "description": "The directory to monitor for files. Paths are absolute.", "id": "/properties/IncomingDirectoryPath", "type": "string", "order": 1 } }, "required": [ "IncomingDirectoryPath" ], "type": "object" }

  Note
  Required for registering a collector.

• MeasurementSettingsSchema [optional] (dict): Schema or structure for measurement settings.

  • Example:

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

  "MeasurementSettingsSchema": { "$schema": "http://json-schema.org/draft-06/schema#", "definitions": {}, "id": "http://factry.io/metricsettings.json", "properties": { "NameInCSV": { "title": "NameInCSV", "description": "The name of the measurement coming from the CSV file that should be mapped to the actual measurement.", "id": "/properties/NameInCSV", "type": "string", "order": 1 } }, "required": [ "NameInCSV" ], "type": "object" }

Note
Required for registering a collector.

High Availability Settings

High availability settings only need to be used in case you want to setup a failover mechanism between a main and a failover collector. The High Availability settings are set on the failover collector.

Note

Failover settings can not be set when creating a collector.

• HA (High Availability) [optional] (dict):
  • HealthPort (integer): Port the main collector will listen on.
  • Host (string): Host address of the main collector.
  • MainCollectorUUID (string): UUID of the main collector in the high availability setup.
  • StopDelay (integer): Delay (in seconds) the failover collector keeps collecting after the main collector returns.
  • Timeout (integer): Timeout (in seconds) after which the failover collector takes over.
  • Example:

    1 2 3 4 5 6 7 8 9 10

    "HA": { "Attributes": { "additionalProp1": "string" }, "HealthPort": 2020, "Host": "string", "MainCollectorUUID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", "StopDelay": 0, "Timeout": 0 }

• HAUUID (string): UUID of the high availability setup.

Example Script

Define a function that creates a collector for an organization.

 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

def create_collector(headers, url, payload):
   response = requests.request("POST", url, json=payload, headers=headers)
   return response.json()

def update_collector(headers, url, payload, collectorUUID):
    url += "/" + collectorUUID
    response = requests.request("PUT", url, json=payload, headers=headers)
    return response.json()

def set_collector_status(headers, url, payload, collectorUUID):
    url += "/" + collectorUUID + "/status"
    response = requests.request("PATCH", url, json=payload, headers=headers)
    return response.json()


# Define settings.
create_settings = {
    "Description": "This is my collector",
    "Name": "My Collector",
    "DefaultDatabaseUUID": "3fa85f64-5717-4562-b3fc-2c963f66afa6"
}

setup_settings = {
    "BuildVersion": "1.0.0",
    "CollectorType": "CSV",
    "MeasurementSettingsSchema": {
        "$schema": "http://json-schema.org/draft-06/schema#",
        "definitions": {},
        "id": "http://factry.io/metricsettings.json",
        "properties": {
        "NameInCSV": {
            "title": "NameInCSV",
            "description": "The name of the measurement coming from the CSV file that should be mapped to the actual measurement.",
            "id": "/properties/NameInCSV",
            "type": "string",
            "order": 1
            }
        },
        "required": [
        "NameInCSV"
        ],
        "type": "object"
    },
    "SettingsSchema": {
        "$schema": "http://json-schema.org/draft-06/schema#",
        "definitions": {},
        "id": "http://factry.io/collectorsettings.json",
        "properties": {
        "IncomingDirectoryPath": {
            "title": "IncomingDirectoryPath",
            "description": "The directory to monitor for files. Paths are absolute.",
            "id": "/properties/IncomingDirectoryPath",
            "type": "string",
            "order": 1
        }
        },
        "required": [
        "IncomingDirectoryPath"
        ],
        "type": "object"
    }
}

# Call the function.
new_collector = create_collector(headers, "https://historian.mycompany.com:8000/api/collectors", create_settings)
# Get the UUID of the new collector.
collectorUUID = new_collector['UUID']

# Generate a token for the collector.
# Perform the installation of the collector on the Historian server.
# The collector will be registered, and this is where information about the collector, such as collector type, build version, settings, etc., is added.

# Set the collector active.
status_collector = set_collector_status(headers, "https://historian.mycompany.com:8000/api/collectors", "Active", collectorUUID)

Manage Measurements

In this section, some of the API functionalities concerning measurements will be described.

Creating a Measurement

Create a measurement given organization UUID. The organization UUID is required in the header field in order to perform this API request.

This call can also be used for adding mutiple measurements at once.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/measurements/bulkUpdateMeasurements

Description

• Attributes [optional] (dict): Custom attributes for engineering specs and timeseries tags.
  • Example:

  1 2 3 4 5 6 7 8 9 10 11 12

  "Attributes": { "Config": { "UoM": "", "LimitHi": null, "LimitLo": null, "ValueMax": null, "ValueMin": null }, "Tags": { "key": "value" } }

• CollectorUUID [required] (string): The collector UUID to add the measurement to.
• DatabaseUUID [optional] (string): The UUID representing the timeseries database UUID where the measurement data will be stored into.
• Datatype [required] (string): Datatype of the measurement.
  • Example: “number”, “boolean”, “string”
• Description [optional] (string): Description of the measurement.
  • Example: “This is the temperature of tank1”
• Labels [optional] (dict):
  • Color: Color associated with the label.
  • Name: Name of the label.
• Name [required] (string): The name of the measurement.
• Quality [optional] (dict):
  • MeasurementUUID: UUID of the measurement.
  • Quality: Quality status of the measurement.
    • Example: “Good”
  • Timestamp: Timestamp of the quality status update.
    • Example: “2023-12-28T08:09:02.001Z”
• Settings [optional] (dict): An optional JSON configuration.
  • Example:

  1 2 3 4 5 6 7 8 9

  "Settings": { "NameInCSV": { "title": "NameInCSV", "description": "The name of the measurement coming from the CSV file that should be mapped to the actual measurement.", "id": "/properties/NameInCSV", "type": "string", "order": 1 } }

• Status [required] (string): Status of the measurement.
  • Example: “Active”, “Paused”, “Deleted”

Example Script

Define a function that creates a measurement/measurements given the organization UUID.

 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

def create_measurements(headers, url, payload):
    response = requests.request("POST", url, json=payload, headers=headers)

    return response.json()

settings= [
            {
                "Attributes": {
                    "Config": {
                        "UoM": "",
                        "LimitHi": null,
                        "LimitLo": null,
                        "ValueMax": null,
                        "ValueMin": null
                        },
                    "Tags": {
                            "key": "value"
                        }
                    },
                "CollectorUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                "DatabaseUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                "Datatype": "number",
                "Description": "This is the temperature of tank1",
                "Labels": [
                    {
                        "Color": "green",
                        "Name": "example label"
                    }
                ],
                "Name": "tank1.temperature",
                "Quality": {
                    "MeasurementUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                    "Quality": "Good",
                    "Timestamp": "2024-01-02T09:39:06.424Z"
                },
                "Settings": {
                    "IncomingDirectoryPath": "/incoming/directory/path"
                },
                "Status": "Active"
            }
        ]

# Call the function.
result = create_measurements(headers, "https://historian.mycompany.com:8000/api/measurements", settings)

Deleting a Measurement

Delete a measurement given measurement UUID. The organization UUID is required in the header field in order to perform this API request and the measurement UUID is required in the path of the API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/measurements/deleteMeasurement

Example Script

Define a function that deletes a measurement given the measurement UUID.

1
2
3
4
5
6
7

def delete_measurement(headers, url, measurementUUID):
    url += "/" + measurementUUID
    response = requests.request("DELETE", url, headers=headers)
    return response.json()

# Call the function.
result = delete_measurement(headers, "https://historian.mycompany.com:8000/api/measurements", "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")

Updating an Existing Measurement

Measurements can be updated using Historian API. The organization UUID is required in the header field in order to perform this API request and the measurement UUID is required in the path of the API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/measurements/updateMeasurement

Available parameter options can be found in Parameters under Body Object.

Some parameters are as follows:

Description

• Attributes [optional] (dict): Custom attributes for engineering specs and timeseries tags.
  • Example:

  1 2 3 4 5 6 7 8 9 10 11 12

  "Attributes": { "Config": { "UoM": "", "LimitHi": null, "LimitLo": null, "ValueMax": null, "ValueMin": null }, "Tags": { "key": "value" } }

• CollectorUUID [required] (string): The collector UUID to add the measurement to.
• DatabaseUUID [required] (string): The UUID representing the timeseries database UUID where the measurement data will be stored into.
• Datatype [required] (string): Datatype of the measurement.
  • Example: “number”, “boolean”, “string”
• Description [optional] (string): Description of the measurement.
  • Example: “This is the temperature of tank1”
• Labels [optional] (dict):
  • Color: Color associated with the label.
  • Name: Name of the label.
• Name [required] (string): Name of the measurement.
• Quality [optional] (dict):
  • MeasurementUUID: UUID of the measurement.
  • Quality: Quality status of the measurement.
    • Example: “Good”
  • Timestamp: Timestamp of the quality status update.
    • Example: “2023-12-28T08:09:02.001Z”
• Settings [optional] (dict): An optional JSON configuration.
  • Example:

  1 2 3

  "Settings": { "IncomingDirectoryPath": "/incoming/directory/path" }

• Status [required] (string): Status of the measurement.
  • Example: “Active”, “Paused”, “Deleted”

Example Script

Define a function that updates given measurement and returns updated measurement in JSON format.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def update_measurements(headers, url, measurementUUID, payload):
    url += "/" + measurementUUID
    response = requests.request("PUT", url, json=payload, headers=headers)
    return response.json()

update_parameters = {
    "CollectorUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    "Datatype": "string",
    "Name": "tank1.temperature",
    "Status": "Active",
    "DatabaseUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    "Description": "This is the temperature of tank1"
}

# Call the function.
measurement = update_measurements(headers, "https://historian.mycompany.com:8000/api/measurements", "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", update_parameters)

Writing Points to a Measurement

Writing data points to the measurements can be done using the Historian API. The organization UUID is required in the header field in order to perform this API request and the collector UUID is required in the path of the API request.

Swagger: https://historian.mycompany.com:8000/api/swagger/#/collectors/createPoints

Description

• ErrorStatus [optional] (string): Status of the error. If the ErrorStatus is empty, the status will be set to Good by default and filled into the tags with the key status.
• MeasurementUUID [required] (string): UUID of the measurement.
• Tags [optional] (dict): Tags of the data point.
  • Example:

  1 2 3 4

  "Tags": { "status": "Good", "sensor": "A1020" }

• Timestamp [optional] (string): Timestamp of the data point.
  • Example: “2023-12-28T14:51:33.854Z”
• Value [required] (dict): Value of the data point.

Example Script

Define a function that writes points to measurements and returns information about creation of the points in JSON format.

 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

def create_points(headers, url, collectorUUID, payload):
    url += "/" + collectorUUID + "/points"  # Collector UUID is added in the URL

    response = requests.request("POST", url, json=payload, headers=headers)
    return response.json()

# Define the parameters that will be used for creating a point.
points = [{
                "MeasurementUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                "Tags": {
                    "status": "Good"
                },
                "Timestamp": "2023-12-28T14:51:33.854Z",
                "Value": "string"
            },
            {
                "ErrorStatus": "This is a bad status!",
                "MeasurementUUID": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
                "Tags": {
                },
                "Timestamp": "2023-12-28T20:30:30.854Z",
                "Value": "string"
            }]

# Call the function.
point_info = create_points(headers, "http://historian.mycompany.com:8000/api/collector/collectors", "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", points)