Skip to content
Snippets Groups Projects

Draft: Change input to file and iterate over multiple IOCs

Open Henrique Silva requested to merge multiple_iocs_check_pvs into master
1 file
+ 17
14
Compare changes
  • Side-by-side
  • Inline
+ 17
14
@@ -5,7 +5,6 @@ import p4p
import helpers
from p4p.client.thread import Context
class EpicsException(BaseException):
pass
@@ -14,14 +13,14 @@ class ApplicationError(BaseException):
pass
def get_list_pv(ctx, pv_name):
content = ctx.get(pv_name, timeout=5)
def get_list_pv(ctx, pv_name, timeout=10):
content = ctx.get(pv_name, timeout=timeout)
pv_list = content.items()[0][1]
return pv_list
def get_pv_value(ctx, pv):
pv_read = ctx.get(pv, timeout=5)
def get_pv_value(ctx, pv, timeout=10):
pv_read = ctx.get(pv, timeout=timeout)
pv_type = type(pv_read).__name__
raw_value = pv_read.raw.items()[0][1]
@@ -39,19 +38,19 @@ def get_pv_value(ctx, pv):
return {'pv': pv, 'type': pv_type, 'value': value, 'str_value': str_value}
def check_pvs(prefix, output_file, save=False, print_json=False):
def check_pvs(prefix, output_file, save=False, print_json=False, timeout=10):
save_res_pv_name = helpers.get_save_res_pv(prefix)
print("Save res pv name: ", save_res_pv_name)
print("Output file: ", output_file)
ctx = Context('pva')
save_res_pv_list = get_list_pv(ctx, save_res_pv_name)
ctx = Context('pva', useenv=False, conf={"EPICS_PVA_ADDR_LIST":"idmz-ro-epics-gw-tn.esss.lu.se"})
save_res_pv_list = get_list_pv(ctx, save_res_pv_name, timeout=timeout)
pv_data = {save_res_pv_name: []}
save_res_pv_data = []
for pv in save_res_pv_list:
save_res_pv_data.append(get_pv_value(ctx, pv))
save_res_pv_data.append(get_pv_value(ctx, pv, timeout=timeout))
pv_data = {save_res_pv_name: save_res_pv_data}
@@ -68,19 +67,23 @@ def check_pvs(prefix, output_file, save=False, print_json=False):
def main():
parser = argparse.ArgumentParser(description="Script to read all PV-s marked as save and restore")
parser.add_argument("prefix", type=str, help="System Prefix")
parser.add_argument('--output_file', type=str, help='Output file name', default="")
parser.add_argument("ioc_list", type=argparse.FileType("r"), help="IOC List")
parser.add_argument('--output_folder', type=str, help='Output folder name', default=".")
parser.add_argument('--save', help="Save to file", default=False, action='store_true')
parser.add_argument('--print', help="Print output json", default=False, action='store_true')
parser.add_argument('--when', type=str, help='Indicate when pv check is done', default="before", choices=["before", "after"])
parser.add_argument('--timeout', type=int, help='Timeout for PVget operations', default=10)
parser.set_defaults(save=True)
args = parser.parse_args()
prefix_fixed = args.prefix.replace(':', '_').strip()
output_file = args.output_file if args.output_file else f"{prefix_fixed}_{args.when}.json"
check_pvs(args.prefix, output_file, args.save, args.print)
iocs = args.ioc_list.read().splitlines()
iocs = [ ioc_name for ioc_name in iocs if "#" not in ioc_name and ioc_name ]
for ioc in iocs:
prefix_fixed = ioc.replace(':', '_').strip()
output_file = f"{args.output_folder}/{prefix_fixed}_{args.when}.json"
check_pvs(ioc, output_file, args.save, args.print, timeout=args.timeout)
if __name__ == "__main__":
main()
Loading