matpva/pva_callback.py at master · slaclab/matpva · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# -----------------------------------------------------------------------------
# Title      : pva callback
# -----------------------------------------------------------------------------
# File       : pva_callback.py
# Author     : Kuktae Kim, ktkim@slac.stanford.edu
# Created    : 2023-11-02
# Last update: 2023-11-02
# -----------------------------------------------------------------------------
# Description:
# Python script for mpvaMonitor function.
# -----------------------------------------------------------------------------
# This file is part of matpva. It is subject to the license terms in the
# LICENSE.txt file found in the top-level directory of this distribution
# and at: https://confluence.slac.stanford.edu/display/ppareg/LICENSE.html.
# No part of matpva, including this file, may be copied, modified,
# propagated, or distributed except according to the terms contained in
# the LICENSE.txt file.
# -----------------------------------------------------------------------------
from p4p.client.thread import Context
import time
import sys
from tabulate import tabulate

# This python script is developed to monitor PVs using the p4p module.
def my_value_processor(pv):
	"""Print updated PV.

	:param value: the PV name
	"""
	#print(f'NEW: {sys.argv[1]}      {value}')
	if "NTTable" in pv.getID():
		print(f'NEW: {sys.argv[1]}   {time.asctime(time.localtime(pv.timeStamp.secondsPastEpoch))}')
		print(tabulate(pv.value.todict(), headers="keys"))
	else:
		print(f'NEW: {sys.argv[1]}   {time.asctime(time.localtime(pv.timeStamp.secondsPastEpoch))}   {pv.value}')


ctxt = Context('pva', nt=False)
subscription = ctxt.monitor(sys.argv[1], my_value_processor)

while True:
	time.sleep(0.1)