refactor(system): add historian.types submodule

requirements.txt

@@ -9,9 +9,9 @@ enum34==1.1.10 \
     --hash=sha256:c3858660960c984d6ab0ebad691265180da2b43f07e061c0f8dca9ef3cffd328 \
     --hash=sha256:cce6a7477ed816bd2542d03d53db9f0db935dd013b70f336a95c73979289f248
     # via -r requirements.in
-java-api==17.26.0 \
-    --hash=sha256:a797a9695e712c9499a900d0376cd56dd49658b5f5cea58dbc49b01c4112249e \
-    --hash=sha256:c1ed6087588b3487983d77f20040378a4cab442780a7be3efd094e26d052aabf
+java-api==17.26.5 \
+    --hash=sha256:9bb74897b0838d6e87a8dc118a56c284fdf660f9db2f2dfbe5b95b672af9cf1e \
+    --hash=sha256:bf68ebcd91e9c8ea222fed6e13e989745eecc2ae8679984b9a9a3b03c1a64edb
     # via -r requirements.in
 typing==3.10.0.0 \
     --hash=sha256:12fbdfbe7d6cca1a42e485229afcb0b0c8259258cfb919b8a5e2a5c953742f89 \

src/com/inductiveautomation/historian/common/model/data/__init__.py

@@ -0,0 +1,274 @@
+from __future__ import print_function
+
+__all__ = [
+    "Annotation",
+    "AnnotationPoint",
+    "AtomicPoint",
+    "DataPoint",
+    "MetadataPoint",
+    "SnapshotCapable",
+    "StandardComplexPointType",
+    "TemporalPoint",
+]
+
+from typing import Any, List, Optional, Union
+
+from java.lang import Class, Comparable, Enum, Object, Record
+from java.time import Instant
+from java.util import UUID
+
+from com.inductiveautomation.ignition.common import QualifiedPath
+from com.inductiveautomation.ignition.common.config import PropertySet, PropertyValue
+from com.inductiveautomation.ignition.common.model.values import QualityCode
+
+
+class SnapshotCapable(object):
+    def isSnapshot(self):
+        # type: () -> bool
+        pass
+
+    def snapshotTime(self):
+        # type: () -> Optional[Instant]
+        pass
+
+    def withSnapshot(self, timestamp=None):
+        # type: (Optional[Instant]) -> SnapshotCapable
+        pass
+
+
+class TemporalPoint(Comparable):
+    def compareTo(self, other):
+        # type: (TemporalPoint) -> int
+        pass
+
+    def source(self):
+        # type: () -> QualifiedPath
+        pass
+
+    def timestamp(self):
+        # type: () -> Instant
+        pass
+
+    def type(self):
+        # type: () -> Any
+        pass
+
+    def value(self):
+        # type: () -> Any
+        pass
+
+
+class DataPoint(SnapshotCapable, TemporalPoint):
+    def quality(self):
+        # type: () -> QualityCode
+        pass
+
+    def valueClass(self):
+        # type: () -> Any
+        pass
+
+
+class AtomicPoint(DataPoint):
+    pass
+
+
+class Annotation(Record):
+    def __init__(
+        self,
+        notes,  # type: Union[str, unicode]
+        type_,  # type: Union[str, unicode]
+        author,  # type: Union[str, unicode]
+    ):
+        # type: (...) -> None
+        super(Annotation, self).__init__()
+
+    def author(self):
+        # type: () -> Union[str, unicode]
+        pass
+
+    def notes(self):
+        # type: () -> Union[str, unicode]
+        pass
+
+    def type(self):
+        # type: () -> Union[str, unicode]
+        pass
+
+
+class AnnotationPoint(Record):
+    class Builder(Object):
+        def __init__(self):
+            # type: () -> None
+            super(AnnotationPoint.Builder, self).__init__()
+
+        def annotationType(self, annotationType):
+            # type: (Union[str, unicode]) -> AnnotationPoint.Builder
+            pass
+
+        def author(self, author):
+            # type: (Union[str, unicode]) -> AnnotationPoint.Builder
+            pass
+
+        def build(self):
+            # type: () -> AnnotationPoint
+            pass
+
+        def endTime(self, endTime):
+            # type: (Instant) -> AnnotationPoint.Builder
+            pass
+
+        def identifier(
+            self,
+            identifier,  # type: Union[UUID, str, unicode]
+        ):
+            # type: (...) -> AnnotationPoint.Builder
+            pass
+
+        def lastUpdated(self, lastUpdated):
+            # type: (Instant) -> AnnotationPoint.Builder
+            pass
+
+        def notes(self, notes):
+            # type: (Union[str, unicode]) -> AnnotationPoint.Builder
+            pass
+
+        def source(self, source):
+            # type: (QualifiedPath) -> AnnotationPoint.Builder
+            pass
+
+        def startTime(self, startTime):
+            # type: (Instant) -> AnnotationPoint.Builder
+            pass
+
+    def __init__(
+        self,
+        identifier,  # type: UUID
+        source,  # type: QualifiedPath
+        value,  # type: Annotation
+        startTime,  # type: Instant
+        endTime=None,  # type: Optional[Instant]
+        lastUpdated=None,  # type: Optional[Instant]
+    ):
+        # type: (...) -> None
+        super(AnnotationPoint, self).__init__()
+        print(identifier, source, value, startTime, endTime, lastUpdated)
+
+    @staticmethod
+    def builder():
+        # type: () -> AnnotationPoint.Builder
+        pass
+
+    def endTime(self):
+        # type: () -> Optional[Instant]
+        pass
+
+    def identifier(self):
+        # type: () -> UUID
+        pass
+
+    def lastUpdated(self):
+        # type: () -> Optional[Instant]
+        pass
+
+    def source(self):
+        # type: () -> QualifiedPath
+        pass
+
+    def startTime(self):
+        # type: () -> Instant
+        pass
+
+    def timestamp(self):
+        # type: () -> Instant
+        pass
+
+    def type(self):
+        # type: () -> StandardComplexPointType
+        pass
+
+    def value(self):
+        # type: () -> Annotation
+        pass
+
+    def withSource(self, source):
+        # type: (QualifiedPath) -> AnnotationPoint
+        pass
+
+
+class MetadataPoint(Record, TemporalPoint):
+    class Builder(Object):
+        def __init__(self):
+            # type: () -> None
+            super(MetadataPoint.Builder, self).__init__()
+
+        def addValue(self, value):
+            # type: (PropertyValue) -> MetadataPoint.Builder
+            pass
+
+        def build(self):
+            # type: () -> MetadataPoint
+            pass
+
+        def quality(self, quality):
+            # type: (QualityCode) -> MetadataPoint.Builder
+            pass
+
+        def source(self, source):
+            # type: (QualifiedPath) -> MetadataPoint.Builder
+            pass
+
+        def timestamp(self, timestamp):
+            # type: (Instant) -> MetadataPoint.Builder
+            pass
+
+        def values(self, values):
+            # type: (PropertySet) -> MetadataPoint.Builder
+            pass
+
+    def __init__(
+        self,
+        value,  # type: PropertySet
+        quality,  # type: QualityCode
+        timestamp,  # type: Instant
+        source,  # type: QualifiedPath
+    ):
+        # type: (...) -> None
+        super(MetadataPoint, self).__init__()
+        print(value, quality, timestamp, source)
+
+    @staticmethod
+    def builder(metadataPoint=None):
+        # type: (Optional[MetadataPoint]) -> MetadataPoint.Builder
+        pass
+
+    @staticmethod
+    def empty(source):
+        # type: (QualifiedPath) -> MetadataPoint
+        pass
+
+    def quality(self):
+        # type: () -> QualityCode
+        pass
+
+    def withSource(self, source):
+        # type: (QualifiedPath) -> MetadataPoint
+        pass
+
+
+class StandardComplexPointType(Enum):
+    ANNOTATION = None  # type: StandardComplexPointType
+    GENERIC = None  # type: StandardComplexPointType
+    METADATA = None  # type: StandardComplexPointType
+
+    def getPointClass(self):
+        # type: () -> Class
+        pass
+
+    def getQueryOptionsClass(self):
+        # type: () -> Class
+        pass
+
+    @staticmethod
+    def values():
+        # type: () -> List[StandardComplexPointType]
+        pass

src/system/historian/types.py

@@ -0,0 +1,104 @@
+"""Historian types."""
+
+__all__ = ["annotationPoint", "dataPoint", "metadataPoint"]
+
+from typing import Any, Dict, Optional, Union
+
+from java.util import Date
+
+from com.inductiveautomation.historian.common.model.data import (
+    AnnotationPoint,
+    AtomicPoint,
+    MetadataPoint,
+)
+
+
+def annotationPoint(
+    source,  # type: Union[str, unicode]
+    startTime,  # type: Date
+    endTime=None,  # type: Optional[Date]
+    annotationType=None,  # type: Optional[Union[str, unicode]]
+    data=None,  # type: Optional[Union[str, unicode]]
+    identifier=None,  # type: Optional[Union[str, unicode]]
+):
+    # type: (...) -> AnnotationPoint
+    """Creates an annotation point that can be stored to a historian.
+
+    Annotation points represent time-based context, such as operator
+    notes, events, or system-generated markers associated with a
+    historical path.
+
+    Args:
+        source: The historical path where the annotation point will be
+            stored.
+        startTime: The start time associated with the annotation.
+        endTime: The end time associated with the annotation. If
+            omitted, the annotation has no end time. Optional.
+        annotationType: A string used to categorize the annotation
+            (for example, "marker", "note", or "event"). If omitted,
+            "marker" is used. Optional.
+        data: A string payload associated with the annotation. This can
+            contain plain text or structured data, such as JSON. If
+            omitted, an empty string is used. Optional.
+        identifier: An identifier used to indicate that an existing
+            annotation should be updated. Optional.
+
+    Returns:
+        An annotation point object that can be passed to
+        system.historian.storeAnnotations.
+    """
+    builder = AnnotationPoint.builder()
+    return builder.build()
+
+
+def dataPoint(
+    source,  # type: Union[str, unicode]
+    value,  # type: object
+    timestamp=None,  # type: Optional[Date]
+    quality=None,  # type: Optional[int]
+):
+    # type: (...) -> AtomicPoint
+    """Creates a data point that can be stored to a historian.
+
+    Data points represent individual values associated with a historical
+    path, timestamp, and quality.
+
+    Args:
+        source: The historical path where the data point will be stored.
+        value: The value to be stored in the historian.
+        timestamp: The timestamp when the data point was recorded. If
+            omitted, the current time is used. Optional.
+        quality: The quality code of the data point. If omitted, a
+            "good" quality is used. Optional.
+
+    Returns:
+        A data point object that can be passed to
+        system.historian.storeDataPoints.
+    """
+    return AtomicPoint()
+
+
+def metadataPoint(
+    source,  # type: Union[str, unicode]
+    properties,  # type: Dict[Union[str, unicode], Any]
+    timestamp,  # type: Date
+):
+    # type: (...) -> MetadataPoint
+    """Creates a metadata point that can be stored to a historian.
+
+    Metadata points allow you to store additional properties associated
+    with a historical path at a specific point in time.
+
+    Args:
+        source: The historical path where the metadata point will be
+            stored.
+        properties: A dictionary of properties to be stored as
+            historical metadata.
+        timestamp: The timestamp when the metadata point was recorded.
+
+    Returns:
+        A metadata point object that can be passed to
+        system.historian.storeMetadata.
+    """
+    builder = MetadataPoint.builder()
+    return builder.build()