diff --git a/PIconnect/AFSDK.py b/PIconnect/AFSDK.py index 33e560b0..b9f3444f 100644 --- a/PIconnect/AFSDK.py +++ b/PIconnect/AFSDK.py @@ -1,14 +1,74 @@ """AFSDK - Loads the .NET libraries from the OSIsoft AF SDK.""" +import dataclasses import logging import os +import pathlib import sys -import typing +from types import ModuleType +from typing import TYPE_CHECKING, cast __all__ = ["AF", "System", "AF_SDK_VERSION"] logger = logging.getLogger(__name__) +if TYPE_CHECKING: + from ._typing import AFType, SystemType +else: + AFType = ModuleType + SystemType = ModuleType + + +@dataclasses.dataclass(kw_only=True) +class PIConnector: + assembly_path: pathlib.Path + AF: AFType + System: SystemType + + # def PIAFSystems(self) -> dict[str, "PIAFSystem"]: + # return {srv.Name: PIAFSystem(srv) for srv in self.AF.PISystems} + + # def PIServers(self) -> dict[str, "PIServer"]: + # return {srv.Name: PIServer(srv) for srv in self.AF.PI.PIServers} + + # @property + # def version(self) -> str: + # return self.AF.PISystems().Version + + # def __str__(self) -> str: + # return f"PIConnector({self.assembly_path}, AF SDK version: {self.version})" + + +StrPath = str | pathlib.Path + + +def get_PI_connector(assembly_path: StrPath | None = None) -> PIConnector: + """Return a new instance of the PI connector.""" + full_path = _get_SDK_path(assembly_path) + if full_path is None: + if assembly_path: + raise ImportError(f"PIAF SDK not found at '{assembly_path}'") + raise ImportError( + "PIAF SDK not found, check installation " + "or pass valid path to directory containing SDK assembly." + ) + dotnetSDK = _get_dotnet_SDK(full_path) + return PIConnector(assembly_path=full_path, **dotnetSDK) + + +def _get_dotnet_SDK(full_path: pathlib.Path) -> dict[str, ModuleType]: + import clr # type: ignore + + sys.path.append(str(full_path)) + clr.AddReference("OSIsoft.AFSDK") # type: ignore ; pylint: disable=no-member + import System # type: ignore + from OSIsoft import AF # type: ignore + + _AF = cast(ModuleType, AF) + _System = cast(ModuleType, System) + return {"AF": _AF, "System": _System} + + # pragma pylint: disable=import-outside-toplevel @@ -28,6 +88,26 @@ def __fallback(): return _af, _System, _AF_SDK_version +def _get_SDK_path(full_path: StrPath | None = None) -> pathlib.Path | None: + if full_path: + assembly_directories = [pathlib.Path(full_path)] + else: + installation_directories = { + os.getenv("PIHOME"), + "C:\\Program Files\\PIPC", + "C:\\Program Files (x86)\\PIPC", + } + assembly_directories = ( + pathlib.Path(path) / "AF\\PublicAssemblies\\4.0\\" + for path in installation_directories + if path is not None + ) + for AF_dir in assembly_directories: + logging.debug("Full path to potential SDK location: '%s'", AF_dir) + if AF_dir.is_dir(): + return AF_dir + + if ( os.getenv("GITHUB_ACTIONS", "false").lower() == "true" or os.getenv("TF_BUILD", "false").lower() == "true" @@ -39,35 +119,22 @@ def __fallback(): # Get the installation directory from the environment variable or fall back # to the Windows default installation path - installation_directories = [ - os.getenv("PIHOME"), - "C:\\Program Files\\PIPC", - "C:\\Program Files (x86)\\PIPC", - ] - for directory in installation_directories: - logging.debug("Trying installation directory '%s'", directory) - if not directory: - continue - AF_dir = os.path.join(directory, "AF\\PublicAssemblies\\4.0\\") - logging.debug("Full path to potential SDK location: '%s'", AF_dir) - if os.path.isdir(AF_dir): - PIAF_SDK = AF_dir - break - else: + PIAF_SDK = _get_SDK_path() + if PIAF_SDK is None: raise ImportError("PIAF SDK not found, check installation") - sys.path.append(PIAF_SDK) + sys.path.append(str(PIAF_SDK)) clr.AddReference("OSIsoft.AFSDK") # type: ignore ; pylint: disable=no-member import System as _System # type: ignore from OSIsoft import AF as _af # type: ignore - _AF_SDK_version = typing.cast(str, _af.PISystems().Version) # type: ignore ; pylint: disable=no-member + _AF_SDK_version = cast(str, _af.PISystems().Version) # type: ignore ; pylint: disable=no-member print("OSIsoft(r) AF SDK Version: {}".format(_AF_SDK_version)) -if typing.TYPE_CHECKING: +if TYPE_CHECKING: # This branch is separate from previous one as otherwise no typechecking takes place # on the main logic. _af, _System, _AF_SDK_version = __fallback() diff --git a/PIconnect/__init__.py b/PIconnect/__init__.py index 8a8c650e..995af5af 100644 --- a/PIconnect/__init__.py +++ b/PIconnect/__init__.py @@ -1,6 +1,6 @@ """PIconnect - Connector to the OSISoft PI and PI-AF databases.""" -from PIconnect.AFSDK import AF, AF_SDK_VERSION +from PIconnect.AFSDK import AF, AF_SDK_VERSION, PIConnector, get_PI_connector from PIconnect.config import PIConfig from PIconnect.PI import PIServer from PIconnect.PIAF import PIAFDatabase @@ -15,6 +15,8 @@ "AF_SDK_VERSION", "PIAFDatabase", "PIConfig", + "PIConnector", "PIServer", + "get_PI_connector", "__sdk_version", ] diff --git a/PIconnect/_typing/__init__.py b/PIconnect/_typing/__init__.py index 67372785..cce28d2f 100644 --- a/PIconnect/_typing/__init__.py +++ b/PIconnect/_typing/__init__.py @@ -1,8 +1,86 @@ """Type stubs for the AF SDK and dotnet libraries.""" -from . import dotnet as System # noqa: I001 +from typing import Protocol + from . import AF +from . import dotnet as System + + +class AFType(Protocol): + # Modules + # Analysis = AF.Analysis + Asset = AF.Asset + # Collective = AF.Collective + Data = AF.Data + # Diagnostics = AF.Diagnostics + EventFrame = AF.EventFrame + # Modeling = AF.Modeling + # Notification = AF.Notification + PI = AF.PI + # Search = AF.Search + # Support = AF.Support + Time = AF.Time + # UI = AF.UI + UnitsOfMeasure = AF.UnitsOfMeasure + + # Classes + # AFActiveDirectoryProperties = AF.AFActiveDirectoryProperties + AFCategory = AF.AFCategory + AFCategories = AF.AFCategories + # AFChangedEventArgs = AF.AFChangedEventArgs + # AFCheckoutInfo = AF.AFCheckoutInfo + # AFClientRegistration = AF.AFClientRegistration + # AFCollection = AF.AFCollection + # AFCollectionList = AF.AFCollectionList + # AFConnectionInfo = AF.AFConnectionInfo + # AFContact = AF.AFContact + # AFCsvColumn = AF.AFCsvColumn + # AFCsvColumns = AF.AFCsvColumns + AFDatabase = AF.AFDatabase + # AFDatabases = AF.AFDatabases + # AFErrors = AF.AFErrors + # AFEventArgs = AF.AFEventArgs + # AFGlobalRestorer = AF.AFGlobalRestorer + # AFGlobalSettings = AF.AFGlobalSettings + # AFKeyedResults = AF.AFKeyedResults + # AFLibraries = AF.AFLibraries + # AFLibrary = AF.AFLibrary + # AFListResults = AF.AFListResults + # AFNamedCollection = AF.AFNamedCollection + # AFNamedCollectionList = AF.AFNamedCollectionList + # AFNameSubstitution = AF.AFNameSubstitution + # AFObject = AF.AFObject + # AFOidcIdentity = AF.AFOidcIdentity + # AFPlugin = AF.AFPlugin + # AFPlugins = AF.AFPlugins + # AFProgressEventArgs = AF.AFProgressEventArgs + # AFProvider = AF.AFProvider + # AFRole = AF.AFRole + # AFSDKExtension = AF.AFSDKExtension + # AFSecurity = AF.AFSecurity + # AFSecurityIdentities = AF.AFSecurityIdentities + # AFSecurityIdentity = AF.AFSecurityIdentity + # AFSecurityMapping = AF.AFSecurityMapping + # AFSecurityMappings = AF.AFSecurityMappings + # AFSecurityRightsExtension = AF.AFSecurityRightsExtension + # NumericStringComparer = AF.NumericStringComparer + PISystem = AF.PISystem + PISystems = AF.PISystems + # UniversalComparer = AF.UniversalComparer + + +class SystemType(Protocol): + # Modules + Data = System.Data + Net = System.Net + Security = System.Security + + # Classes + DateTime = System.DateTime + Exception = System.Exception + TimeSpan = System.TimeSpan + AF_SDK_VERSION = "2.7_compatible" -__all__ = ["AF", "AF_SDK_VERSION", "System"] +__all__ = ["AF", "AF_SDK_VERSION", "AFType", "System"] diff --git a/docs/conf.py b/docs/conf.py index bdbb1a77..b44562da 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -85,11 +85,11 @@ def __getattr__(cls, name) -> MagicMock: # type: ignore # built documents. # # The short X.Y version. -version = PIconnect.__version__ +version = '.'.join(PIconnect.__version__.split('.')[:2]) # The full version, including alpha/beta/rc tags. release = PIconnect.__version__ -extlinks = {"afsdk": ("https://docs.osisoft.com/bundle/af-sdk/page/html/%s", "")} +extlinks = {"afsdk": ("https://docs.aveva.com/bundle/af-sdk/page/html/%s", "")} intersphinx_mapping = { "python": ("https://docs.python.org/3.10", None), diff --git a/tests/common.py b/tests/common.py new file mode 100644 index 00000000..2fa314ae --- /dev/null +++ b/tests/common.py @@ -0,0 +1,17 @@ +"""Common fixtures for testing PIconnect.""" + +import os + +import pytest + + +def on_CI() -> bool: + """Return True if the tests are running on a CI environment.""" + return ( + os.getenv("GITHUB_ACTIONS", "false").lower() == "true" + or os.getenv("TF_BUILD", "false").lower() == "true" + or os.getenv("READTHEDOCS", "false").lower() == "true" + ) + + +skip_if_on_CI = pytest.mark.skipif(on_CI(), reason="Real SDK not available on CI") diff --git a/tests/test_load_SDK.py b/tests/test_load_SDK.py new file mode 100644 index 00000000..9cf3888a --- /dev/null +++ b/tests/test_load_SDK.py @@ -0,0 +1,52 @@ +"""Test the loading of the SDK connector.""" + +import pathlib + +import pytest + +import PIconnect as PI + +from .common import skip_if_on_CI + +# Skip this test module on CI as it requires the real SDK to be installed +pytestmark = skip_if_on_CI + + +def test_load_SDK_without_arguments_raises_no_exception() -> None: + """Test that loading the SDK object without arguments raises no exception.""" + try: + PI.get_PI_connector() + except Exception as e: + pytest.fail(f"Exception raised: {e}") + + +def test_load_SDK_returns_PIconnect_object() -> None: + """Test that loading the SDK object returns a PIConnector.""" + assert isinstance(PI.get_PI_connector(), PI.PIConnector) + + +def test_load_SDK_with_a_valid_path_returns_SDK_object() -> None: + """Test that loading the SDK object with a path returns a PIConnector.""" + assembly_path = "c:\\Program Files (x86)\\PIPC\\AF\\PublicAssemblies\\4.0\\" + assert isinstance(PI.get_PI_connector(assembly_path), PI.PIConnector) + + +def test_load_SDK_with_a_valid_path_stores_path_in_connector() -> None: + """Test that loading the SDK object with a path stores the path in the connector.""" + assembly_path = "c:\\Program Files (x86)\\PIPC\\AF\\PublicAssemblies\\4.0\\" + connector = PI.get_PI_connector(assembly_path) + assert connector.assembly_path == pathlib.Path(assembly_path) + + +def test_load_SDK_with_an_invalid_path_raises_import_error() -> None: + """Test that loading the SDK object with an invalid path raises an ImportError.""" + assembly_path = "c:\\invalid\\path\\" + with pytest.raises(ImportError, match="PIAF SDK not found at .*"): + PI.get_PI_connector(assembly_path) + + +def test_load_SDK_with_valid_path_has_SDK_reference() -> None: + """Test that loading the SDK object with a valid path has a reference to the SDK.""" + assembly_path = "c:\\Program Files (x86)\\PIPC\\AF\\PublicAssemblies\\4.0\\" + connector = PI.get_PI_connector(assembly_path) + assert connector.AF is not None