Scenario: Oil Refinery
The refinery operations system is alive and well. Data is trended, displayed, stored, calculated, simulated, etc.… Then something happens. Murphy and his Law visit the plant. The process engineer in charge doesn't have a spreadsheet, display, log or report that provides the information he needs so he fires up his AI Agent to investigate.
The engineer needs to analyze the behavior of the misbehaving process to gain insight through advanced time-series analysis using descriptive (central tendency, dispersion, shape), inferential (correlation and regression) and/or multivariate statistics.
The AI Agent has a model of the plant so the engineer asks the agent to save all tags to a file that are related to the section of the plant that is of interest (or drag and drop tags from a graphic). The engineer writes and runs a prompt telling the agent what to do. The engineer waits for the response, changes the prompt, and runs the prompt again. Old School successive refinement in action. If the engineer desires, he can trigger the agent to run periodically.
Industrial AI Agent
An "AI Agent" is a mathematically-literate Large Language Model (LLM) advisor that is connected to the real world and can ingest and analyze data that is accessible through a permissioned application programming interface (API). An agent can sample and process data at whatever time interval makes sense and send the results to humans and/or other systems.
An "Industrial AI Agent" is an AI Agent that has access to industrial process data.
Let's modify the AI Power Agent to access and reason about plant/refinery/factory process data.
https://upperbay.com/v1/blogs/beachblogs/ai-power-agents
Integrating AI with PI (API^2?)
AVEVA (Oil Systems) PI is the industry standard data historian used in a wide range of industrial industries from oil and gas to the power grid. PI has a RESTful web interface that enables secure access to the PI real-time time-series database.
A Python module is needed that accesses the PI database and creates a CSV (comma separated value) file of the returned process data. This module can be easily stitched into the Omnibus runpump.py application. Let's offload the task of coding the baseline module to OpenAI GPT-4o.
GPT-4o Prompt to Create Python PI Data Access Module Script (Version 0.00.00.00.001)
"Write a python module called xpilib.py with a class called XPI. XPI has a method called get_pi_data. This method takes parameters pointnamesfile, outputcsvfile and starttime of the data and uses the PI Web API to get a number of data points from an OSI PI database and create a CSV file with point names as the column headers. The method is to include error recovery and exception handling. The point names are read from a pointnamesfile with each tag on a different line. The method should return OK if no error and ERROR if there was an error."
GPT-4o Outputs Generated Code (Version 0.00.00.00.001) (see below)
Not sure if this code will work but it looks like a good place to start.
Next Steps
1. Stitch the xpilib.py module into runpump.py
2. Find a test PI server
3. Test/Debug
Easing into AI makes the most sense. It will take time to experiment and build up trust in this powerful new technology. No time like the present to start working with AI!!!!
Having Fun with Agents!
Dave Hardin
============= #Module xpilib.py
import requests from requests.auth import HTTPBasicAuth import json from datetime import datetime, timedelta import pandas as pd import sys
class XPI: def init(self, base_url, username, password, pi_server): self.base_url = base_url self.username = username self.password = password self.pi_server = pi_server
def read_point_names(self, pointnamesfile):
try:
with open(pointnamesfile, 'r') as file:
point_names = [line.strip() for line in file if line.strip()]
return point_names
except Exception as e:
print(f"Error reading point names file: {e}")
return None
def get_point_webid(self, point_name):
try:
url = f"{self.base_url}/points?path=\\{self.pi_server}\\{point_name}"
response = requests.get(url, auth=HTTPBasicAuth(self.username, self.password), verify=False)
response.raise_for_status()
point = response.json()
return point['WebId']
except requests.exceptions.RequestException as e:
print(f"Error retrieving PI point {point_name}: {e}")
return None
def get_recorded_values(self, webid, start_time):
try:
end_time = datetime.now()
url = f"{self.base_url}/streams/{webid}/recorded"
params = {
"startTime": start_time.isoformat(),
"endTime": end_time.isoformat(),
"maxCount": 1000
}
response = requests.get(url, params=params, auth=HTTPBasicAuth(self.username, self.password), verify=False)
response.raise_for_status()
recorded_values = response.json()
return recorded_values['Items']
except requests.exceptions.RequestException as e:
print(f"Error retrieving recorded values for WebId {webid}: {e}")
return []
def get_pi_data(self, pointnamesfile, outputcsvfile, starttime):
point_names = self.read_point_names(pointnamesfile)
if point_names is None:
return "ERROR"
try:
start_time = datetime.fromisoformat(starttime)
except ValueError as e:
print(f"Invalid start time format: {e}")
return "ERROR"
data = {}
timestamps = set()
for point_name in point_names:
webid = self.get_point_webid(point_name)
if webid:
recorded_values = self.get_recorded_values(webid, start_time)
if recorded_values:
data[point_name] = {value['Timestamp']: value['Value'] for value in recorded_values}
timestamps.update(data[point_name].keys())
if data:
try:
timestamps = sorted(timestamps)
df = pd.DataFrame(index=timestamps)
for point_name in point_names:
if point_name in data:
df[point_name] = df.index.map(data[point_name].get)
df.to_csv(outputcsvfile, index_label="Timestamp")
print(f"Data has been written to {outputcsvfile}")
return "OK"
except Exception as e:
print(f"Error writing to CSV file: {e}")
return "ERROR"
else:
print("No data to write to CSV file.")
return "ERROR"
#Example usage if name == "main": base_url = "https://<your-pi-web-api-server>/piwebapi" username = "<username>" password = "<password>" pi_server = "<your-pi-server>"
xpi = XPI(base_url, username, password, pi_server)
result = xpi.get_pi_data("pointnames.txt", "pi_data.csv", "2024-05-28T00:00:00")
print(result)