Merge pull request #5 from shilorigins/devagr/add-skeletons

Add skeleton dataclasses, client, and backend

conda-recipe/meta.yaml

@@ -21,6 +21,7 @@ requirements:
     - setuptools_scm
   run:
     - python >=3.9
+    - apischema
     - pcdsutils
     - pyqt
     - qtpy

docs/source/upcoming_release_notes/5-add_dataclass_skeletons.rst

@@ -0,0 +1,22 @@
+5 add dataclass skeletons
+#################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- Add dataclasses and simple serialization tests
+
+Bugfixes
+--------
+- N/A
+
+Maintenance
+-----------
+- N/A
+
+Contributors
+------------
+- shilorigins

requirements.txt

@@ -1,4 +1,5 @@
 # List requirements here.
+apischema
 pcdsutils
 PyQt5
 qtpy

superscore/model.py

@@ -0,0 +1,125 @@
+"""Classes for representing data"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Flag, IntEnum, auto
+from typing import ClassVar, List, Optional, Set, Union
+from uuid import UUID, uuid4
+
+from superscore.type_hints import AnyEpicsType
+from superscore.utils import utcnow
+
+logger = logging.getLogger(__name__)
+
+
+class Severity(IntEnum):
+    NO_ALARM = auto()
+    MINOR = auto()
+    MAJOR = auto()
+    INVALID = auto()
+
+
+class Status(IntEnum):
+    NO_ALARM = auto()
+    READ = auto()
+    WRITE = auto()
+    HIHI = auto()
+    HIGH = auto()
+    LOLO = auto()
+    LOW = auto()
+    STATE = auto()
+    COS = auto()
+    COMM = auto()
+    TIMEOUT = auto()
+    HWLIMIT = auto()
+    CALC = auto()
+    SCAN = auto()
+    LINK = auto()
+    SOFT = auto()
+    BAD_SUB = auto()
+    UDF = auto()
+    DISABLE = auto()
+    SIMM = auto()
+    READ_ACCESS = auto()
+    WRITE_ACCESS = auto()
+
+
+class Tag(Flag):
+    pass
+
+
+@dataclass
+class Entry:
+    """
+    Base class for items in the data model
+    """
+    uuid: UUID = field(default_factory=uuid4)
+    description: str = ''
+    creation_time: datetime = field(default_factory=utcnow)
+
+
+@dataclass
+class Parameter(Entry):
+    """An Entry that stores a PV name"""
+    pv_name: str = ''
+    abs_tolerance: Optional[float] = None
+    rel_tolerance: Optional[float] = None
+    readback: Optional[Parameter] = None
+    read_only: bool = False
+
+
+@dataclass
+class Value(Entry):
+    """An Entry that stores a PV name and data pair"""
+    pv_name: str = ''
+    data: Optional[AnyEpicsType] = None
+    status: Status = Status.UDF
+    severity: Severity = Severity.INVALID
+
+
+@dataclass
+class Setpoint(Value):
+    """A Value that can be written to the EPICS environment"""
+    readback: Optional[Readback] = None
+
+
+@dataclass
+class Readback(Value):
+    """
+    A read-only Value representing machine state that cannot be written to. A
+    restore is considered complete when all Setpoint values are within
+    tolerance of their Readback values.
+
+    abs_tolerance - tolerance given in units matching the Setpoint
+    rel_tolerance - tolerance given as a percentage
+    timeout - time (seconds) after which a Setpoint restore is considered to
+              have failed
+    """
+    abs_tolerance: Optional[float] = None
+    rel_tolerance: Optional[float] = None
+    timeout: Optional[float] = None
+
+
+@dataclass
+class Collection(Entry):
+    """Nestable group of Parameters and Collections"""
+    meta_pvs: ClassVar[List[Parameter]] = []
+    all_tags: ClassVar[Set[Tag]] = set()
+
+    title: str = ""
+    children: List[Union[Parameter, Collection]] = field(default_factory=list)
+    tags: Set[Tag] = field(default_factory=set)
+
+
+@dataclass
+class Snapshot(Entry):
+    """
+    Nestable group of Values and Snapshots.  Effectively a data-filled Collection
+    """
+    title: str = ""
+    origin_collection: Optional[UUID] = None
+    children: List[Union[Value, Snapshot]] = field(default_factory=list)
+    tags: Set[Tag] = field(default_factory=set)
+    meta_pvs: List[Value] = field(default_factory=list)

superscore/tests/test_model.py

@@ -0,0 +1,38 @@
+import apischema
+
+from superscore.model import (Collection, Parameter, Severity, Snapshot,
+                              Status, Value)
+
+
+def test_serialize_collection_roundtrip():
+    p1 = Parameter(pv_name="TEST:PV1", description="First test Parameter")
+    p2 = Parameter(pv_name="TEST:PV2", description="Second test Parameter")
+    p3 = Parameter(pv_name="TEST:PV3", description="Third test Parameter")
+    p4 = Parameter(pv_name="TEST:PV4", description="Fourth test Parameter")
+    c1 = Collection(title="Collection 1", description="Inner Collection", children=[p1, p2])
+    c2 = Collection(title="Collection 2", description="Outer Collection", children=[p3, c1, p4])
+    serial = apischema.serialize(c2)
+    deserialized = apischema.deserialize(Collection, serial)
+    assert deserialized == c2
+    assert deserialized.children[0] == p3
+    assert deserialized.children[1] == c1
+    assert deserialized.children[2] == p4
+    assert deserialized.children[1].children[0] == p1
+    assert deserialized.children[1].children[1] == p2
+
+
+def test_serialize_snapshot_roundtrip():
+    v1 = Value(pv_name="TEST:PV1", description="First test Value", data=4, status=Status.NO_ALARM, severity=Severity.NO_ALARM)
+    v2 = Value(pv_name="TEST:PV2", description="Second test Value", data=1.8, status=Status.UDF, severity=Severity.INVALID)
+    v3 = Value(pv_name="TEST:PV3", description="Third test Value", data="TRIM", status=Status.DISABLE, severity=Severity.NO_ALARM)
+    v4 = Value(pv_name="TEST:PV4", description="Fourth test Value", data=False, status=Status.HIGH, severity=Severity.MAJOR)
+    s1 = Snapshot(title="Snapshot 1", description="Snapshot of Inner Collection", children=[v1, v2])
+    s2 = Snapshot(title="Snapshot 2", description="Snapshot of Outer Collection", children=[v3, s1, v4])
+    serial = apischema.serialize(s2)
+    deserialized = apischema.deserialize(Snapshot, serial)
+    assert deserialized == s2
+    assert deserialized.children[0] == v3
+    assert deserialized.children[1] == s1
+    assert deserialized.children[2] == v4
+    assert deserialized.children[1].children[0] == v1
+    assert deserialized.children[1].children[1] == v2

superscore/type_hints.py

@@ -0,0 +1,3 @@
+from typing import Union
+
+AnyEpicsType = Union[int, str, float, bool]

superscore/utils.py

@@ -1,3 +1,8 @@
+from datetime import datetime, timezone
 from pathlib import Path
 
 SUPERSCORE_SOURCE_PATH = Path(__file__).parent
+
+
+def utcnow():
+    return datetime.now(timezone.utc)