New methods for Handler and SharedPV

example/persist.py

@@ -0,0 +1,205 @@
+"""
+Use a handler to automatically persist values to an SQLite3 file database.
+Any values persisted this way will be automatically restored when the
+program is rerun. The details of users (account name and IP address) are
+recorded for puts.
+
+Try monitoring the PV `demo:pv:optime` then quit, wait, and restart the
+program while continuing to monitor the PV. Compare with the value of
+`demo:pv:uptime` which resets on each program start. Try setting the value of
+demo:pv:optime while continuing to monitor it. It is recommended to
+inspect the persisted file, e.g. `sqlite3 persist_pvs.db "select * from pvs"`.
+
+There is an important caveat for this simple demo:
+The `PersistHandler` will not work as expected if anything other than the
+value of a field is changed, e.g. if a Control field was added to an NTScalar
+if would not be persisted correctly. This could be resolved by correctly
+merging the pv.current().raw and value.raw appropriately in the post().
+"""
+
+import json
+import sqlite3
+import time
+
+from p4p import Value
+from p4p.nt.scalar import NTScalar
+from p4p.server import Server, ServerOperation
+from p4p.server.raw import Handler
+from p4p.server.thread import SharedPV
+
+
+class PersistHandler(Handler):
+    """
+    A handler that will allow simple persistence of values and timestamps using
+    SQLite3 across retarts.
+
+    It requires the open handler in order to restore values from a previous run.
+    It requires a post handler in order to persist values set within the program.
+    """
+
+    def __init__(self, pv_name: str, conn: sqlite3.Connection, open_restore=True):
+        self._conn = conn
+        self._pv_name = pv_name
+        self._open_restore = open_restore
+
+    def open(self, value: Value, **_kws):
+        """
+        Restore values from the previous run, if possible.
+        If there isn't a previous value then we persist this initial value.
+        """
+
+        # If there is a value already in the database we always use that
+        # instead of the supplied initial value, unless the
+        # _open_restore flag indicates otherwise.
+        if not self._open_restore:
+            return
+
+        # We could, in theory, re-apply authentication here if we queried for
+        # that information and then did something with it!
+        res = self._conn.execute("SELECT data FROM pvs WHERE id=?", [self._pv_name])
+        query_val = res.fetchone()
+
+        if query_val is not None:
+            json_val = json.loads(query_val[0])
+            print(f"Will restore to {self._pv_name} value: {json_val['value']}")
+
+            # Override initial value
+            value["value"] = json_val["value"]
+
+            # Override current timestamp (if any) with time the new value was set
+            value["timeStamp.secondsPastEpoch"] = json_val["timeStamp"][
+                "secondsPastEpoch"
+            ]
+            value["timeStamp.nanoseconds"] = json_val["timeStamp"]["nanoseconds"]
+        else:
+            # We are using an initial value so persist it
+            self._upsert(value)
+
+    def post(self, pv: SharedPV, value: Value, **_kws):
+        """Update timestamp of the PV and store its new value."""
+        self._update_timestamp(value)
+
+        self._upsert(value)
+
+    def put(self, pv: SharedPV, op: ServerOperation):
+        """
+        Update timestamp of the PV and store its new value. Unlike the post()
+        we include the account and peer update in the data stored.
+        """
+
+        self._update_timestamp(op.value())
+
+        self._upsert(op.value(), op.account(), op.peer())
+
+        op.done()
+
+    def _update_timestamp(self, value) -> None:
+        """Update the timestamp of the PV to the current time."""
+
+        if not value.changed("timeStamp") or (
+            value["timeStamp.nanoseconds"] == value["timeStamp.nanoseconds"] == 0
+        ):
+            now = time.time()
+            value["timeStamp.secondsPastEpoch"] = now // 1
+            value["timeStamp.nanoseconds"] = int((now % 1) * 1e9)
+
+    def _upsert(self, value, account=None, peer=None) -> None:
+        # Persist the data; turn into JSON and write it to the DB
+        val_json = json.dumps(value.todict())
+
+        # Use UPSERT: https://sqlite.org/lang_upsert.html
+        self._conn.execute(
+            """
+            INSERT INTO pvs (id, data, account, peer) 
+                        VALUES (:name, :json_data, :account, :peer)
+            ON CONFLICT(id)
+            DO UPDATE SET data = :json_data, account = :account, peer = :peer;
+            """,
+            {
+                "name": self._pv_name,
+                "json_data": val_json,
+                "account": account,
+                "peer": peer,
+            },
+        )
+        self._conn.commit()
+
+
+def main() -> None:
+    """
+    Create SQLite3 database and a set of PVs to demonstrate persistance
+    using a p4p Handler.
+    """
+
+    # Create an SQLite dayabase to function as our persistence store
+    conn = sqlite3.connect("persist_pvs.db", check_same_thread=False)
+    # conn.execute("DROP TABLE IF EXISTS pvs")
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS pvs (id VARCHAR(255), data JSON, account VARCHAR(30), peer VARCHAR(55), PRIMARY KEY (id));"
+    )  # IPv6 addresses can be long and will contain port number as well!
+
+    # Create the example PVs.
+    # Note that we switch off restores for `demo:pv:uptime`. If we simply
+    # didn't use the PersistHandler the timestamp of the PV would not be
+    # updated.
+    duplicate_pv = SharedPV(
+        nt=NTScalar("i"), handler=PersistHandler("demo:pv:int", conn), initial=12
+    )
+    pvs = {
+        "demo:pv:optime": SharedPV(
+            nt=NTScalar("i"),
+            handler=PersistHandler("demo:pv:optime", conn),
+            initial=0,
+        ),  # Operational time; total time running
+        "demo:pv:uptime": SharedPV(
+            nt=NTScalar("i"),
+            handler=PersistHandler("demo:pv:uptime", conn, open_restore=False),
+            timestamp=time.time(),
+            initial=0,
+        ),  # Uptime since most recent (re)start
+        "demo:pv:int": duplicate_pv,
+        "demo:pv:float": SharedPV(
+            nt=NTScalar("d"),
+            handler=PersistHandler("demo:pv:float", conn),
+            initial=9.99,
+        ),
+        "demo:pv:string": SharedPV(
+            nt=NTScalar("s"),
+            handler=PersistHandler("demo:pv:string", conn),
+            initial="Hello!",
+        ),
+        "demo:pv:alias_int": duplicate_pv,  # It works except for reporting its restore
+    }
+
+    # Make the uptime PV readonly; maybe we want to be able to update optime
+    # after major system upgrades?
+    uptime_pv = pvs["demo:pv:uptime"]
+
+    @uptime_pv.put
+    def read_only(_pv: SharedPV, op: ServerOperation):
+        """Don't allow put operations on this PV."""
+
+        op.done(error="Read-only")
+
+    print(f"Starting server with the following PVs: {pvs}")
+
+    server = None
+    try:
+        server = Server(providers=[pvs])
+        while True:
+            # Every second increment the values of uptime and optime
+            time.sleep(1)
+            increment_value = pvs["demo:pv:uptime"].current().raw["value"] + 1
+            pvs["demo:pv:uptime"].post(increment_value)
+            increment_value = pvs["demo:pv:optime"].current().raw["value"] + 1
+            pvs["demo:pv:optime"].post(increment_value)
+    except KeyboardInterrupt:
+        pass
+    finally:
+        if server:
+            server.stop()
+        conn.close()
+
+
+if __name__ == "__main__":
+    main()

src/p4p/server/raw.py

@@ -39,6 +39,13 @@ class Handler(object):
 
     Use of this as a base class is optional.
     """
+    def open(self, value, **kws):
+        """
+        Called each time an Open operation is performed on this Channel
+
+        :param value:  A Value, or appropriate object (see nt= and wrap= of the constructor).
+        """
+        pass
 
     def put(self, pv, op):
         """
@@ -50,6 +57,17 @@ def put(self, pv, op):
         """
         op.done(error='Not supported')
 
+    def post(self, pv, value, **kws):
+        """
+        Called each time a client issues a post
+        operation on this Channel.
+
+        :param SharedPV pv: The :py:class:`SharedPV` which this Handler is associated with.
+        :param value:  A Value, or appropriate object (see nt= and wrap= of the constructor).        
+        :param dict options: A dictionary of configuration options.
+        """
+        pass
+
     def rpc(self, pv, op):
         """
         Called each time a client issues a Remote Procedure Call
@@ -77,6 +95,14 @@ def onLastDisconnect(self, pv):
         pass
 
 
+    def close(self, pv):
+        """
+        Called when the Channel is closed.
+
+        :param SharedPV pv: The :py:class:`SharedPV` which this Handler is associated with.
+        """
+        pass
+
 class SharedPV(_SharedPV):
 
     """Shared state Process Variable.  Callback based implementation.
@@ -158,6 +184,14 @@ def open(self, value, nt=None, wrap=None, unwrap=None, **kws):
             V = self._wrap(value, **kws)
         except: # py3 will chain automatically, py2 won't
             raise ValueError("Unable to wrap %r with %r and %r"%(value, self._wrap, kws))
+
+        # Guard goes here because we can have handlers that don't inherit from 
+        # the Handler base class
+        try:
+            self._handler.open(V, **kws)
+        except AttributeError as err:
+            pass
+
         _SharedPV.open(self, V)
 
     def post(self, value, **kws):
@@ -174,8 +208,35 @@ def post(self, value, **kws):
             V = self._wrap(value, **kws)
         except: # py3 will chain automatically, py2 won't
             raise ValueError("Unable to wrap %r with %r and %r"%(value, self._wrap, kws))
+
+        # Guard goes here because we can have handlers that don't inherit from 
+        # the Handler base class
+        try:
+            self._handler.post(self, V, **kws)
+        except AttributeError:
+            pass
+
         _SharedPV.post(self, V)
 
+    def close(self, destroy=False, sync=False, timeout=None):
+        """Close PV, disconnecting any clients.
+
+        :param bool destroy: Indicate "permanent" closure.  Current clients will not see subsequent open().
+        :param bool sync: When block until any pending onLastDisconnect() is delivered (timeout applies).
+        :param float timeout: Applies only when sync=True.  None for no timeout, otherwise a non-negative floating point value.
+
+        close() with destory=True or sync=True will not prevent clients from re-connecting.
+        New clients may prevent sync=True from succeeding.
+        Prevent reconnection by __first__ stopping the Server, removing with :py:meth:`StaticProvider.remove()`,
+        or preventing a :py:class:`DynamicProvider` from making new channels to this SharedPV.
+        """
+        try:  
+            self._handler.close(self)
+        except AttributeError:
+            pass
+
+        _SharedPV.close(self)
+
     def current(self):
         V = _SharedPV.current(self)
         try:
@@ -208,6 +269,13 @@ def __init__(self, pv, real):
             self._pv = pv  # this creates a reference cycle, which should be collectable since SharedPV supports GC
             self._real = real
 
+        def open(self, value, **kws):
+            _log.debug('OPEN %s %s', self._pv, value)
+            try:
+                self._pv._exec(None, self._real.open, value, **kws)
+            except AttributeError:
+                pass
+
         def onFirstConnect(self):
             self._pv._exec(None, self._pv._onFirstConnect, None)
             try:  # user handler may omit onFirstConnect()
@@ -239,6 +307,20 @@ def rpc(self, op):
             except AttributeError:
                 op.done(error="RPC not supported")
 
+        def post(self, value, **kws):
+            _log.debug('POST %s %s', self._pv, value)
+            try:
+                self._pv._exec(None, self._real.post, self._pv, value, **kws)
+            except AttributeError:
+                pass
+
+        def close(self, pv):
+            _log.debug('CLOSE %s', self._pv)
+            try:
+                self._pv._exec(None, self._real.close, self._pv)
+            except AttributeError:
+                pass
+
     @property
     def onFirstConnect(self):
         def decorate(fn):
@@ -253,6 +335,20 @@ def decorate(fn):
             return fn
         return decorate
 
+    @property
+    def on_open(self):
+        def decorate(fn):
+            self._handler.open = fn
+            return fn
+        return decorate     
+
+    @property
+    def on_post(self):
+        def decorate(fn):
+            self._handler.post = fn
+            return fn
+        return decorate
+
     @property
     def put(self):
         def decorate(fn):
@@ -267,6 +363,13 @@ def decorate(fn):
             return fn
         return decorate
 
+    @property
+    def on_close(self):
+        def decorate(fn):
+            self._handler.close = fn
+            return fn
+        return decorate  
+
     def __repr__(self):
         if self.isOpen():
             return '%s(value=%s)' % (self.__class__.__name__, repr(self.current()))