idcp-drive-info displays information about undulator drives.
Warning, cannot access the index:
_darcs/index: opening of '_darcs/index' failed: permission denied (Permission denied)
diff -rN -u old-bii_scripts/Makefile new-bii_scripts/Makefile
--- old-bii_scripts/Makefile 2022-12-09 20:53:15.979793611 +0100
+++ new-bii_scripts/Makefile 2022-12-09 20:53:15.979793611 +0100
@@ -289,6 +289,7 @@
git-meld \
hg-compare-repos \
idcp-dist-info \
+ idcp-drive-info \
idcp-restore \
psh \
substdiff\
diff -rN -u old-bii_scripts/bin/idcp-drive-info new-bii_scripts/bin/idcp-drive-info
--- old-bii_scripts/bin/idcp-drive-info 1970-01-01 01:00:00.000000000 +0100
+++ new-bii_scripts/bin/idcp-drive-info 2022-12-09 20:53:15.979793611 +0100
@@ -0,0 +1,512 @@
+#! /usr/bin/env python3
+# -*- coding: UTF-8 -*-
+
+# Copyright 2021 Helmholtz-Zentrum Berlin für Materialien und Energie GmbH
+# <https://www.helmholtz-berlin.de>
+#
+# Author: Goetz Pfeiffer <Goetz.Pfeiffer@helmholtz-berlin.de>
+#
+# This program is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free Software
+# Foundation, either version 3 of the License, or (at your option) any later
+# version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
+# details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""A script to display idcp drive information.
+"""
+
+# pylint: disable= invalid-name
+
+import argparse
+import os
+import os.path
+import sys
+import subprocess
+
+# pylint: disable=invalid-name
+
+VERSION= "1.0"
+
+SUMMARY="Displays idcp drive information"
+
+USAGE= "%(prog)s [OPTIONS] COMMAND"
+
+DESC= '''
+This program displays idcp drive information.
+
+known commands:
+ list : List all known IDs, simply calls "iddb"
+ drivetypes [ID...] :
+ List drivetypes. If ID is not given, list drivetypes
+ for all IDs, otherwise only for the given ones.
+ axles [ID...] :
+ List axle numbers. If ID is not given, list axle numbers
+ for all IDs, otherwise only for the given ones.
+ versions [ID...] :
+ Print *Unidrive* program versions. If ID is not given,
+ list versions for all IDs with unidrive devices,
+ otherwise only for the given ones.
+'''
+
+DRIVENAMES=("gap","shift","chicane","pseudo")
+DRIVETYPES= { "Un": "unidrive", "Mo": "MOCON", "_": ""}
+
+# -----------------------------------------------
+# basic system utilities
+# -----------------------------------------------
+
+# standard set of environment variables here:
+_new_env = dict(os.environ)
+
+# Only on Unix-Like systems:
+# Ensure that language settings for called commands are english, keep current
+# character encoding:
+if os.name=="posix" and "LANG" in _new_env:
+ _l= _new_env["LANG"].split(".")
+ if len(_l)==2:
+ _l[0]= "en_US"
+ _new_env["LANG"]= ".".join(_l)
+
+def copy_env():
+ """create a new environment that the user may change."""
+ return dict(_new_env)
+
+def system_rc(cmd, catch_stdout, catch_stderr, env, verbose, dry_run):
+ """execute a command.
+
+ execute a command and return the programs output
+ may raise:
+ IOError(errcode,stderr)
+ OSError(errno,strerr)
+ ValueError
+ """
+ # pylint: disable=too-many-arguments
+ def to_str(data):
+ """decode byte stream to unicode string."""
+ if data is None:
+ return None
+ return data.decode()
+ if dry_run or verbose:
+ print(">", cmd)
+ if dry_run:
+ return (None, None, 0)
+ if catch_stdout:
+ stdout_par=subprocess.PIPE
+ else:
+ stdout_par=None
+
+ if catch_stderr:
+ stderr_par=subprocess.PIPE
+ else:
+ stderr_par=None
+ if env is None:
+ env= _new_env
+
+ p= subprocess.Popen(cmd, shell=True,
+ stdout=stdout_par, stderr=stderr_par,
+ close_fds=True,
+ env= env
+ )
+ (child_stdout, child_stderr) = p.communicate()
+ # pylint: disable=E1101
+ # "Instance 'Popen'has no 'returncode' member
+ return (to_str(child_stdout), to_str(child_stderr), p.returncode)
+
+def system(cmd, catch_stdout, catch_stderr, env, verbose, dry_run):
+ """execute a command with returncode.
+
+ execute a command and return the programs output
+ may raise:
+ IOError(errcode,stderr)
+ OSError(errno,strerr)
+ ValueError
+ """
+ # pylint: disable=too-many-arguments
+ (child_stdout, child_stderr, rc)= system_rc(cmd,
+ catch_stdout, catch_stderr,
+ env,
+ verbose, dry_run)
+ if rc!=0:
+ # pylint: disable=no-else-raise
+ if catch_stderr:
+ raise IOError(rc,
+ "cmd \"%s\", errmsg \"%s\"" % (cmd,child_stderr))
+ else:
+ raise IOError(rc,
+ "cmd \"%s\", rc %d" % (cmd, rc))
+ return (child_stdout, child_stderr)
+
+def syscall(command, verbose, dry_run, catch_stderr= False):
+ """shortcut for system command."""
+ (output, _)= system(command,
+ catch_stdout= True,
+ catch_stderr= catch_stderr,
+ env= None,
+ verbose= verbose,
+ dry_run= dry_run)
+ return output.splitlines()
+
+def syscall_line1(command, verbose, dry_run):
+ """system command, return 1st line."""
+ return syscall(command, verbose, dry_run)[0]
+
+def cmd_ssh(ssh_arg, cmd):
+ """prepend ssh call to command."""
+ if not ssh_arg:
+ return cmd
+ cmd= cmd.replace('"', r'\"')
+ return 'ssh %s "%s"' % (ssh_arg, cmd)
+
+# -----------------------------------------------
+# StructuredData handling
+# -----------------------------------------------
+
+def sd_split_find(l):
+ """split a line of a find command."""
+ (k,v)= l.strip().split(":")
+ path= k.strip()
+ val = v.strip()
+ return (path, val)
+
+def sd_split_path(path):
+ """split structureddata path."""
+ return path.split(".")
+
+# -----------------------------------------------
+# id name utilities
+# -----------------------------------------------
+
+id_data= {}
+aliases= {}
+
+def init_id_data(ssh_arg, verbose, dry_run):
+ """get some standard information for all ids.
+ """
+ if id_data:
+ return
+ lines= syscall(cmd_ssh(ssh_arg, "iddb -N list -a idcp"), verbose, dry_run)
+ for l in lines:
+ elms= l.strip().split()
+ # name devicename key prefix application group status
+ id_data[elms[0]]= { "devicename" : elms[1],
+ "key" : elms[2],
+ "prefix" : elms[3],
+ "application" : elms[4],
+ "group" : elms[5],
+ "status" : elms[6]
+ }
+
+def init_aliases(ssh_arg, verbose, dry_run):
+ """init aliases."""
+ if aliases:
+ return
+ init_id_data(ssh_arg, verbose, dry_run)
+ for k, data in id_data.items():
+ aliases[k]= k
+ aliases[k.upper()]= k
+ aliases[k.lower()]= k
+ nk= k.replace("/", "-")
+ aliases[nk.upper()]= k
+ aliases[nk.lower()]= k
+ for dkey in ("devicename", "key", "prefix"):
+ dval= data[dkey]
+ aliases[dval.upper()]= k
+ aliases[dval.lower()]= k
+
+def id_names(ids, ssh_arg, verbose, dry_run):
+ """return the standard ID names for a list of ids.
+ """
+ init_aliases(ssh_arg, verbose, dry_run)
+ if not ids:
+ return ids
+ return [aliases[i] for i in ids]
+
+def devicename(id_, ssh_arg, verbose, dry_run):
+ """return devicename for an ID."""
+ init_id_data(ssh_arg, verbose, dry_run)
+ return id_data[id_]["devicename"]
+
+
+# -----------------------------------------------
+# lower level commands
+# -----------------------------------------------
+
+_drivetypes_dict= None
+
+def drivetypes_dict(ssh_arg, verbose, dry_run):
+ """return drivetypes for given ids.
+
+ ids must be a list of id names returned by id_names.
+ """
+ # pylint: disable= global-statement, too-many-locals
+ global _drivetypes_dict
+ if _drivetypes_dict:
+ return _drivetypes_dict
+ result= {}
+ lines= syscall(cmd_ssh(ssh_arg,
+ "iddb find 'id-data.*.physical.drivetypes'"),
+ verbose, dry_run)
+ for l in lines:
+ # example of line :
+ # "id-data.UE112.physical.drivetypes : Un,Un,_,_"
+ (path, val)= sd_split_find(l)
+ id_= sd_split_path(path)[1]
+ drives= val.split(",")
+ drive_dict= {}
+ for idx, d in enumerate(drives):
+ drivetype= DRIVETYPES[d]
+ drivename= DRIVENAMES[idx]
+ drive_dict[drivename]= drivetype
+ result[id_]= drive_dict
+ _drivetypes_dict= result
+ return result
+
+_axles_dict= None
+
+def axles_dict(ssh_arg, verbose, dry_run):
+ """return dict with all axes for all ids.
+ """
+ # pylint: disable= global-statement
+ global _axles_dict
+ if _axles_dict:
+ return _axles_dict
+
+ result= {}
+ lines= syscall(cmd_ssh(ssh_arg,
+ "iddb find 'id-data.*.physical.all_axles'"),
+ verbose, dry_run)
+ for l in lines:
+ # example of l :
+ # "id-data.UE48.physical.all_axles : 4,4,0,0"
+ (path, val)= sd_split_find(l)
+ id_= sd_split_path(path)[1]
+ id_axles= [int(i) for i in val.split(",")]
+ result[id_]= id_axles
+ _axles_dict= result
+ return result
+
+# -----------------------------------------------
+# middle level commands
+# -----------------------------------------------
+
+def drivetypes(ids, table, ssh_arg, verbose, dry_run):
+ """return drivetypes for a single or all IDs.
+
+ ids must be a list of id names returned by id_names.
+ """
+ drivetype_d= drivetypes_dict(ssh_arg, verbose, dry_run)
+ if not ids:
+ ids= sorted(drivetype_d.keys())
+ for id_ in ids:
+ data= drivetype_d[id_]
+ if not table:
+ print("%s:" % id_)
+ for drivename in DRIVENAMES:
+ drivetype= data[drivename]
+ if not drivetype:
+ continue # if empty
+ print(" %-8s: %s" % (drivename, drivetype))
+ else:
+ l= []
+ for drivename in DRIVENAMES:
+ l.append(data[drivename])
+ while l:
+ if not l[-1]:
+ l.pop()
+ else:
+ break
+ print("%-10s %s" % (id_, " ".join(["%-10s" % dt for dt in l])))
+
+def axles(ids, table, ssh_arg, verbose, dry_run):
+ """return axles for a single or all IDs.
+
+ ids must be a list of id names returned by id_names.
+ """
+ axles_d= axles_dict(ssh_arg, verbose, dry_run)
+ if not ids:
+ ids= sorted(axles_d.keys())
+ for id_ in ids:
+ data= axles_d[id_]
+ if not table:
+ print("%s:" % id_)
+ for idx, drive_axno in enumerate(data):
+ if drive_axno==0:
+ continue
+ print(" %-8s: %2d" % (DRIVENAMES[idx], drive_axno))
+ else:
+ id_axles= data[:]
+ while id_axles:
+ if id_axles[-1]==0:
+ id_axles.pop()
+ else:
+ break
+ print("%-10s %s" % (id_, " ".join(["%2d" % int(n) \
+ for n in id_axles])))
+
+def versions(ids, ssh_arg, verbose, dry_run):
+ """return unidrive versions for a single or all IDs.
+
+ ids must be a list of id names returned by id_names.
+ """
+ # pylint: disable= too-many-locals, too-many-nested-blocks
+ drivetype_d= drivetypes_dict(ssh_arg, verbose, dry_run)
+ axles_d= axles_dict(ssh_arg, verbose, dry_run)
+ if not ids:
+ ids= sorted(axles_d.keys())
+ for id_ in ids:
+ id_name_printed= False
+ dev= devicename(id_, ssh_arg, verbose, dry_run)
+ axle_list= axles_d[id_]
+ id_drivetypes= drivetype_d[id_]
+ idx=1
+ try:
+ for (drive_idx, drive_axles) in enumerate(axle_list):
+ if id_drivetypes[DRIVENAMES[drive_idx]]=="unidrive":
+ pvnames= []
+ for i in range(drive_axles):
+ pvnames.append("%s:AdiUn%dGblVer" % (dev, idx+i))
+ lines= syscall(cmd_ssh(ssh_arg,
+ "caget %s" % (" ".join(pvnames))),
+ verbose, dry_run,
+ catch_stderr= True)
+ # caution, ^^^ this means that no error messages
+ # from the 'caget' command are printed. However, an
+ # error always throws an IOError exception.
+ for i, l in enumerate(lines):
+ val= l.strip().split()[1]
+ if not id_name_printed:
+ print("%s:" % id_)
+ id_name_printed= True
+ print(" %2d: %s" % (idx+i, val))
+ idx+= axle_list[drive_idx]
+ except IOError:
+ if not id_name_printed:
+ print("%s:" % id_)
+ print(" cannot read pv")
+
+# -----------------------------------------------
+# main commands
+# -----------------------------------------------
+
+def cmd_list(args):
+ """list all IDs."""
+ system(cmd_ssh(args.ssh, "iddb"),
+ catch_stdout= False,
+ catch_stderr= False,
+ env= None,
+ verbose= args.verbose,
+ dry_run= args.dry_run)
+
+def cmd_drivetypes(args, ids):
+ """list drivetypes for given ids."""
+ drivetypes(id_names(ids, args.ssh, args.verbose, args.dry_run),
+ args.table,
+ args.ssh,
+ args.verbose, args.dry_run)
+
+def cmd_axles(args, ids):
+ """list axles for given ids."""
+ axles(id_names(ids, args.ssh, args.verbose, args.dry_run),
+ args.table,
+ args.ssh,
+ args.verbose, args.dry_run)
+
+def cmd_versions(args, ids):
+ """list unidrive versions for given ids."""
+ versions(id_names(ids, args.ssh, args.verbose, args.dry_run),
+ args.ssh,
+ args.verbose, args.dry_run)
+
+def process(args, rest):
+ """do all the work.
+ """
+ #print("args:",args)
+ #print("rest:",rest)
+ if args.summary:
+ print_summary()
+ sys.exit(0)
+ if not rest:
+ sys.exit("command missing")
+ if rest[0]=="list":
+ cmd_list(args)
+ return
+ if rest[0]=="drivetypes":
+ cmd_drivetypes(args, rest[1:])
+ return
+ if rest[0]=="axles":
+ cmd_axles(args, rest[1:])
+ return
+ if rest[0]=="versions":
+ cmd_versions(args, rest[1:])
+ return
+ sys.exit("unknown command: %s" % repr(rest[0]))
+
+
+def script_shortname():
+ """return the name of this script without a path component."""
+ return os.path.basename(sys.argv[0])
+
+def print_summary():
+ """print a short summary of the scripts function."""
+ print("%-20s: %s\n" % (script_shortname(), SUMMARY))
+
+
+def main():
+ """The main function.
+
+ parse the command-line options and perform the command
+ """
+ parser = argparse.ArgumentParser(\
+ usage= USAGE,
+ description= DESC,
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ )
+ parser.add_argument('--version', action='version', version='%%(prog)s %s' % VERSION)
+
+ parser.add_argument("--summary",
+ action="store_true",
+ help="print a summary of the function of the program",
+ )
+ #parser.add_argument("-f", "--file",
+ # help="specify the FILE",
+ # metavar="FILE"
+ # )
+ parser.add_argument("--ssh",
+ help="specify ssh host for caget and iddb command. "
+ "This may be 'hostname' or 'user@hostname'."
+ )
+ parser.add_argument("-t", "--table",
+ action="store_true",
+ help="table output, one line per item",
+ )
+ parser.add_argument("-v", "--verbose",
+ action="store_true",
+ help="print to the screen what the program does",
+ )
+ parser.add_argument("--dry-run",
+ action="store_true",
+ help="do not apply any changes",
+ )
+
+ (args, rest) = parser.parse_known_args()
+ if rest:
+ for r in rest:
+ if r.startswith("-"):
+ sys.exit("unknown option: %s" % repr(r))
+
+ if args.summary:
+ print_summary()
+ sys.exit(0)
+
+ process(args, rest)
+ sys.exit(0)
+
+if __name__ == "__main__":
+ main()
diff -rN -u old-bii_scripts/doc/txt/CONTENTS.rst new-bii_scripts/doc/txt/CONTENTS.rst
--- old-bii_scripts/doc/txt/CONTENTS.rst 2022-12-09 20:53:15.979793611 +0100
+++ new-bii_scripts/doc/txt/CONTENTS.rst 2022-12-09 20:53:15.979793611 +0100
@@ -364,6 +364,12 @@
control) IOCs
| `documentation <scripts/idcp-dist-info>`__
+**idcp-drive-info**
+ | Displays information about drives of IDCP controled undulators.
+ | This script can also show the software version of the uniserv program for
+ each undulator.
+ | `documentation <scripts/idcp-drive-info>`__
+
**idcp-get-source**
| Get and and optionally build the source of installed versions of IDCP.
| You can provide the undulator name or the version string. You can also
patch a8a5db164597679b806d6fb9a3b7abe19e55a504
Author: Goetz.Pfeiffer@helmholtz-berlin.de
Date: Fri Apr 30 15:33:29 CEST 2021
* idcp-drive-info displays information about undulator drives