API Reference#

PIconnect - Connector to the OSISoft PI and PI-AF databases.

class PIconnect.AFDatabase(server: str | None = None, database: str | None = None)#

Bases: object

Context manager for connections to the PI Asset Framework database.

property children: dict[str, AFElement]#

Return a dictionary of the direct child elements of the database.

property database_name: str#

Return the name of the connected PI AF database.

classmethod default_server() → PISystem | None#

Return the default server.

descendant(path: str) → AFElement#

Return a descendant of the database from an exact path.

event_frames(start_time: str | datetime = '', start_index: int = 0, max_count: int = 1000, search_mode: EventFrameSearchMode = EventFrameSearchMode.FORWARD_FROM_START_TIME, search_full_hierarchy: bool = False) → dict[str, AFEventFrame]#

Search for event frames in the database.

property server_name: str#

Return the name of the connected PI AF server.

classmethod servers() → dict[str, PISystem]#

Return a dictionary of the known servers.

property tables: dict[str, AFTable]#

Return a dictionary of the tables in the database.

version = '0.3.0'#
class PIconnect.PIAFDatabase(server: str | None = None, database: str | None = None)#

Bases: AFDatabase

Context manager for connections to the PI Asset Framework database.

property children: dict[str, AFElement]#

Return a dictionary of the direct child elements of the database.

property database_name: str#

Return the name of the connected PI AF database.

classmethod default_server() → PISystem | None#

Return the default server.

descendant(path: str) → AFElement#

Return a descendant of the database from an exact path.

event_frames(start_time: str | datetime = '', start_index: int = 0, max_count: int = 1000, search_mode: EventFrameSearchMode = EventFrameSearchMode.FORWARD_FROM_START_TIME, search_full_hierarchy: bool = False) → dict[str, AFEventFrame]#

Search for event frames in the database.

property server_name: str#

Return the name of the connected PI AF server.

classmethod servers() → dict[str, PISystem]#

Return a dictionary of the known servers.

property tables: dict[str, AFTable]#

Return a dictionary of the tables in the database.

version = '0.3.0'#
class PIconnect.PIServer(server: str | None = None, username: str | None = None, password: str | None = None, domain: str | None = None, authentication_mode: AuthenticationMode = AuthenticationMode.PI_USER_AUTHENTICATION, timeout: int | None = None)#

Bases: object

PIServer is a connection to an OSIsoft PI Server.

Parameters:

  • (str (password)

  • optional) (the maximum seconds an operation can take)

  • (str

  • optional)

  • (str

  • optional)

  • todo (domain, auth)

  • (int (timeout)

  • optional)

Note

If the specified server is unknown a warning is thrown and the connection is redirected to the default server, as if no server was passed. The list of known servers is available in the PIServer.servers dictionary.

classmethod default_server() → PIServer | None#

Return the default server.

search(query: str | list[str], source: str | None = None) → list[PIPoint]#

Search PIPoints on the PIServer.

Parameters:

  • [str]) (query (str or)

  • (str (source)

  • optional) (Defaults to None. Point source to limit the results)

Returns:

list

Return type:

A list of PIPoint objects as a result of the query

Todo

Reject searches while not connected

property server_name#

Name of the connected server.

classmethod servers() → dict[str, PIServer]#

Return a dictionary of the known servers.

version = '0.2.2'#
PIconnect.load_SDK(assembly_path: str | Path | None = None) → None#

Load the AF SDK from the specified path.

Parameters:

  • Path (assembly_path (str |) – installation path will be used.

  • optional) (Path to the AF SDK assembly. If None, the default) – installation path will be used.

Raises:

ImportError – If the AF SDK cannot be found or loaded.:

Generic utility modules#