Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions PIconnect/PI.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""PI - Core containers for connections to PI databases."""

import warnings
from typing import Any, Dict, List, Optional, Union, cast
from typing import Any, cast

import PIconnect.PIPoint as PIPoint_
from PIconnect import AF, PIConsts
Expand All @@ -14,8 +14,8 @@
_DEFAULT_AUTH_MODE = PIConsts.AuthenticationMode.PI_USER_AUTHENTICATION


def _lookup_servers() -> Dict[str, AF.PI.PIServer]:
servers: Dict[str, AF.PI.PIServer] = {}
def _lookup_servers() -> dict[str, AF.PI.PIServer]:
servers: dict[str, AF.PI.PIServer] = {}

for server in AF.PI.PIServers():
try:
Expand All @@ -30,7 +30,7 @@ def _lookup_servers() -> Dict[str, AF.PI.PIServer]:
return servers


def _lookup_default_server() -> Optional[AF.PI.PIServer]:
def _lookup_default_server() -> AF.PI.PIServer | None:
default_server = None
try:
default_server = AF.PI.PIServers().DefaultPIServer
Expand Down Expand Up @@ -65,12 +65,12 @@ class PIServer(object): # pylint: disable=useless-object-inheritance

def __init__(
self,
server: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None,
domain: Optional[str] = None,
server: str | None = None,
username: str | None = None,
password: str | None = None,
domain: str | None = None,
authentication_mode: PIConsts.AuthenticationMode = _DEFAULT_AUTH_MODE,
timeout: Optional[int] = None,
timeout: int | None = None,
) -> None:
if server is None:
if self.default_server is None:
Expand Down Expand Up @@ -138,8 +138,8 @@ def server_name(self):
return self.connection.Name

def search(
self, query: Union[str, List[str]], source: Optional[str] = None
) -> List[PIPoint_.PIPoint]:
self, query: str | list[str], source: str | None = None
) -> list[PIPoint_.PIPoint]:
"""Search PIPoints on the PIServer.

Parameters
Expand Down
24 changes: 11 additions & 13 deletions PIconnect/PIAF.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import dataclasses
import warnings
from typing import Any, Optional, Union, cast
from typing import Any, cast

import pandas as pd

Expand All @@ -20,12 +20,12 @@ class PIAFServer:
server: AF.PISystem
databases: dict[str, AF.AFDatabase] = dataclasses.field(default_factory=dict)

def __getitem__(self, attr: str) -> Union[AF.PISystem, dict[str, AF.AFDatabase]]:
def __getitem__(self, attr: str) -> AF.PISystem | dict[str, AF.AFDatabase]:
"""Allow access to attributes as if they were dictionary items."""
return getattr(self, attr)


ServerSpec = dict[str, Union[AF.PISystem, dict[str, AF.AFDatabase]]]
ServerSpec = dict[str, AF.PISystem | dict[str, AF.AFDatabase]]


def _lookup_servers() -> dict[str, ServerSpec]:
Expand Down Expand Up @@ -59,7 +59,7 @@ def _lookup_servers() -> dict[str, ServerSpec]:
}


def _lookup_default_server() -> Optional[ServerSpec]:
def _lookup_default_server() -> ServerSpec | None:
servers = _lookup_servers()
if AF.PISystems().DefaultPISystem:
return servers[AF.PISystems().DefaultPISystem.Name]
Expand All @@ -75,14 +75,14 @@ class PIAFDatabase(object):
version = "0.3.0"

servers: dict[str, ServerSpec] = _lookup_servers()
default_server: Optional[ServerSpec] = _lookup_default_server()
default_server: ServerSpec | None = _lookup_default_server()

def __init__(self, server: Optional[str] = None, database: Optional[str] = None) -> None:
def __init__(self, server: str | None = None, database: str | None = None) -> None:
server_spec = self._initialise_server(server)
self.server: AF.PISystem = server_spec["server"] # type: ignore
self.database: AF.AFDatabase = self._initialise_database(server_spec, database)

def _initialise_server(self, server: Optional[str]) -> ServerSpec:
def _initialise_server(self, server: str | None) -> ServerSpec:
if server is None:
if self.default_server is None:
raise ValueError("No server specified and no default server found.")
Expand All @@ -99,9 +99,7 @@ def _initialise_server(self, server: Optional[str]) -> ServerSpec:

return self.servers[server]

def _initialise_database(
self, server: ServerSpec, database: Optional[str]
) -> AF.AFDatabase:
def _initialise_database(self, server: ServerSpec, database: str | None) -> AF.AFDatabase:
def default_db():
default = self.server.Databases.DefaultDatabase
if default is None:
Expand Down Expand Up @@ -161,7 +159,7 @@ def descendant(self, path: str) -> "PIAFElement":
"""Return a descendant of the database from an exact path."""
return PIAFElement(self.database.Elements.get_Item(path))

def search(self, query: Union[str, list[str]]) -> list[PIAFAttribute.PIAFAttribute]:
def search(self, query: str | list[str]) -> list[PIAFAttribute.PIAFAttribute]:
"""Search PIAFAttributes by element|attribute path strings.

Return a list of PIAFAttributes directly from a list of element|attribute path strings
Expand Down Expand Up @@ -220,7 +218,7 @@ class PIAFElement(PIAFBase.PIAFBaseElement[AF.Asset.AFElement]):
version = "0.1.0"

@property
def parent(self) -> Optional["PIAFElement"]:
def parent(self) -> "PIAFElement | None":
"""Return the parent element of the current element, or None if it has none."""
if not self.element.Parent:
return None
Expand All @@ -247,7 +245,7 @@ def event_frame(self) -> AF.EventFrame.AFEventFrame:
return self.element

@property
def parent(self) -> Optional["PIAFEventFrame"]:
def parent(self) -> "PIAFEventFrame | None":
"""Return the parent element of the current event frame, or None if it has none."""
if not self.element.Parent:
return None
Expand Down
8 changes: 4 additions & 4 deletions PIconnect/PIAFAttribute.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import dataclasses
import datetime
from typing import Any, Dict, Optional
from typing import Any

from PIconnect import AF, PIData, PIPoint, _time

Expand All @@ -21,7 +21,7 @@ def name(self) -> str:
return self.data_reference.Name

@property
def pi_point(self) -> Optional[PIPoint.PIPoint]:
def pi_point(self) -> PIPoint.PIPoint | None:
if self.data_reference.PIPoint is not None:
return PIPoint.PIPoint(self.data_reference.PIPoint)

Expand Down Expand Up @@ -56,14 +56,14 @@ def name(self) -> str:
return self.attribute.Name

@property
def parent(self) -> Optional["PIAFAttribute"]:
def parent(self) -> "PIAFAttribute | None":
"""Return the parent attribute of the current attribute, or None if it has none."""
if not self.attribute.Parent:
return None
return self.__class__(self.element, self.attribute.Parent)

@property
def children(self) -> Dict[str, "PIAFAttribute"]:
def children(self) -> dict[str, "PIAFAttribute"]:
"""Return a dictionary of the direct child attributes of the current attribute."""
return {a.Name: self.__class__(self.element, a) for a in self.attribute.Attributes}

Expand Down
4 changes: 2 additions & 2 deletions PIconnect/PIAFBase.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Base element class for PI AF elements."""

from typing import Dict, Generic, TypeVar
from typing import Generic, TypeVar

import PIconnect.PIAFAttribute as PIattr
from PIconnect import AF
Expand All @@ -26,7 +26,7 @@ def name(self) -> str:
return self.element.Name

@property
def attributes(self) -> Dict[str, PIattr.PIAFAttribute]:
def attributes(self) -> dict[str, PIattr.PIAFAttribute]:
"""Return a dictionary of the attributes of the current element."""
return {a.Name: PIattr.PIAFAttribute(self.element, a) for a in self.element.Attributes}

Expand Down
26 changes: 13 additions & 13 deletions PIconnect/PIData.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import abc
import datetime
from typing import Any, List, Optional
from typing import Any

import pandas as pd

Expand All @@ -24,9 +24,9 @@ class PISeries(pd.Series): # type: ignore
Parameters
----------
tag (str): Name of the new series
timestamp (List[datetime]): List of datetime objects to
timestamp (list[datetime]): List of datetime objects to
create the new index
value (List): List of values for the timeseries, should be equally long
value (list): List of values for the timeseries, should be equally long
as the `timestamp` argument
uom (str, optional): Defaults to None. Unit of measurement for the
series
Expand All @@ -42,9 +42,9 @@ class PISeries(pd.Series): # type: ignore
def __init__(
self,
tag: str,
timestamp: List[datetime.datetime],
value: List[Any],
uom: Optional[str] = None,
timestamp: list[datetime.datetime],
value: list[Any],
uom: str | None = None,
*args: Any,
**kwargs: Any,
) -> None:
Expand Down Expand Up @@ -87,7 +87,7 @@ def filtered_summaries(
summary_types: PIConsts.SummaryType,
calculation_basis: PIConsts.CalculationBasis = _DEFAULT_CALCULATION_BASIS,
filter_evaluation: PIConsts.ExpressionSampleType = _DEFAULT_FILTER_EVALUATION,
filter_interval: Optional[str] = None,
filter_interval: str | None = None,
time_type: PIConsts.TimestampCalculation = PIConsts.TimestampCalculation.AUTO,
) -> pd.DataFrame:
"""Return one or more summary values for each interval within a time range.
Expand Down Expand Up @@ -259,8 +259,8 @@ def interpolated_values(
_filter_expression = self._normalize_filter_expression(filter_expression)
pivalues = self._interpolated_values(time_range, _interval, _filter_expression)

timestamps: List[datetime.datetime] = []
values: List[Any] = []
timestamps: list[datetime.datetime] = []
values: list[Any] = []
for value in pivalues:
timestamps.append(_time.timestamp_to_index(value.Timestamp.UtcTime))
values.append(value.Value)
Expand Down Expand Up @@ -393,8 +393,8 @@ def recorded_values(

pivalues = self._recorded_values(time_range, _boundary_type, _filter_expression)

timestamps: List[datetime.datetime] = []
values: List[Any] = []
timestamps: list[datetime.datetime] = []
values: list[Any] = []
for value in pivalues:
timestamps.append(_time.timestamp_to_index(value.Timestamp.UtcTime))
values.append(value.Value)
Expand Down Expand Up @@ -558,14 +558,14 @@ def _summaries(

@property
@abc.abstractmethod
def units_of_measurement(self) -> Optional[str]:
def units_of_measurement(self) -> str | None:
"""Return the units of measurment of the values in the current object."""
pass

def update_value(
self,
value: Any,
time: Optional[_time.TimeLike] = None,
time: _time.TimeLike | None = None,
update_mode: PIConsts.UpdateMode = PIConsts.UpdateMode.NO_REPLACE,
buffer_mode: PIConsts.BufferMode = PIConsts.BufferMode.BUFFER_IF_POSSIBLE,
) -> None:
Expand Down
6 changes: 3 additions & 3 deletions PIconnect/PIPoint.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""PIPoint."""

from typing import Any, Dict, Optional
from typing import Any

import PIconnect._typing.AF as _AFtyping
from PIconnect import AF, PIData, _time
Expand Down Expand Up @@ -56,13 +56,13 @@ def name(self) -> str:
return self.tag

@property
def raw_attributes(self) -> Dict[str, Any]:
def raw_attributes(self) -> dict[str, Any]:
"""Return a dictionary of the raw attributes of the PI Point."""
self.__load_attributes()
return self.__raw_attributes

@property
def units_of_measurement(self) -> Optional[str]:
def units_of_measurement(self) -> str | None:
"""Return the units of measument in which values for this PI Point are reported."""
return self.raw_attributes["engunits"]

Expand Down
3 changes: 1 addition & 2 deletions PIconnect/_time.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,11 @@
# pyright: strict
import datetime
import zoneinfo
from typing import Union

from PIconnect import AF, PIConfig
from PIconnect.AFSDK import System

TimeLike = Union[str, datetime.datetime]
TimeLike = str | datetime.datetime


def to_af_time_range(start_time: TimeLike, end_time: TimeLike) -> AF.Time.AFTimeRange:
Expand Down
10 changes: 6 additions & 4 deletions PIconnect/_typing/AF.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Mock classes for the AF namespace of the OSIsoft PI-AF SDK."""

from typing import Iterator, List
from collections.abc import Iterator

from . import PI, Asset, Data, EventFrame, Time, UnitsOfMeasure

Expand All @@ -22,8 +22,8 @@ class AFCategory:
"""Mock class of the AF.AFCategory class."""


class AFCategories(List[AFCategory]):
def __init__(self, elements: List[AFCategory]) -> None:
class AFCategories(list[AFCategory]):
def __init__(self, elements: list[AFCategory]) -> None:
self.Count: int
self._values = elements

Expand All @@ -33,7 +33,9 @@ class AFDatabase:

def __init__(self, name: str) -> None:
self.Name = name
self.Elements = Asset.AFElements([Asset.AFElement("TestElement")])
self.Elements = Asset.AFElements(
[Asset.AFElement("TestElement"), Asset.AFElement("BaseElement")]
)
self.Tables = Asset.AFTables([Asset.AFTable("TestTable")])


Expand Down
10 changes: 5 additions & 5 deletions PIconnect/_typing/Asset.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Mock classes for the AF module."""

from collections.abc import Iterator
from typing import Optional, Union, cast
from typing import cast

from . import AF, Data, Generic
from . import UnitsOfMeasure as UOM
Expand All @@ -23,7 +23,7 @@


class AFAttribute:
def __init__(self, name: str, parent: Optional["AFAttribute"] = None) -> None:
def __init__(self, name: str, parent: "AFAttribute | None" = None) -> None:
self.Attributes: AFAttributes
if parent is None:
self.Attributes = AFAttributes(
Expand Down Expand Up @@ -55,7 +55,7 @@ def __iter__(self) -> Iterator[AFAttribute]:


class AFBaseElement:
def __init__(self, name: str, parent: Optional["AFElement"] = None) -> None:
def __init__(self, name: str, parent: "AFElement | None" = None) -> None:
self.Attributes = AFAttributes(
[
AFAttribute("Attribute1"),
Expand Down Expand Up @@ -86,7 +86,7 @@ def __init__(self, elements: list[AFElement]) -> None:
self.Count: int
self._values = elements

def get_Item(self, name: Union[str, int]) -> AFElement:
def get_Item(self, name: str | int) -> AFElement:
"""Stub for the indexer."""
if isinstance(name, int):
return self._values[name]
Expand All @@ -104,7 +104,7 @@ class AFDataReference:
from . import PI

def __init__(
self, name: str, attribute: AFAttribute, pi_point: Optional[PI.PIPoint] = None
self, name: str, attribute: AFAttribute, pi_point: PI.PIPoint | None = None
) -> None:
self.Attribute = attribute
self.Name = name
Expand Down
Loading
Loading