SW-540: moved ssh_cmd_list method into seperate module ssh_utils.py. Added...

SW-540: moved ssh_cmd_list method into seperate module ssh_utils.py. Added module lcu_utils.py with wrappers around lcurun and the stations.txt file and fetching station calibration tables.

.gitattributes

@@ -1684,8 +1684,10 @@ LCS/PyCommon/datetimeutils.py -text
 LCS/PyCommon/defaultmailaddresses.py -text
 LCS/PyCommon/factory.py -text
 LCS/PyCommon/flask_utils.py -text
+LCS/PyCommon/lcu_utils.py -text
 LCS/PyCommon/math.py -text
 LCS/PyCommon/postgres.py -text
+LCS/PyCommon/ssh_utils.py -text
 LCS/PyCommon/subprocess_utils.py -text
 LCS/PyCommon/test/python-coverage.sh eol=lf
 LCS/PyCommon/test/t_cache.py -text

LCS/PyCommon/CMakeLists.txt

@@ -10,7 +10,9 @@ find_python_module(psycopg2)
 
 set(_py_files
   __init__.py
+  ssh_utils.py
   cep4_utils.py
+  lcu_utils.py
   cache.py
   dbcredentials.py
   defaultmailaddresses.py

LCS/PyCommon/cep4_utils.py

@@ -15,6 +15,7 @@
 # You should have received a copy of the GNU General Public License along
 # with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
 
+from .ssh_utils import ssh_cmd_list
 from subprocess import check_output, Popen, PIPE
 from random import randint
 import math
@@ -25,19 +26,6 @@ from datetime import datetime, timedelta
 import logging
 logger = logging.getLogger(__name__)
 
-
-def ssh_cmd_list(host):
-    '''
-    returns a subprocess compliant command list to do an ssh call to the given node
-    uses ssh option -tt to force remote pseudo terminal
-    uses ssh option -q for ssh quiet mode (no ssh warnings/errors)
-    uses ssh option -o StrictHostKeyChecking=no to prevent prompts about host keys
-    :param host: the node name or ip address
-    :return: a subprocess compliant command list
-    '''
-    return ['ssh', '-T', '-q', '-o StrictHostKeyChecking=no', host]
-
-
 def wrap_command_in_cep4_head_node_ssh_call(cmd):
     '''wrap the command in an ssh call to head.cep4
     :param list cmd: a subprocess cmd list

LCS/PyCommon/lcu_utils.py

+# Copyright (C) 2012-2015  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite 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.
+#
+# The LOFAR software suite 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 the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+
+from lofar.common.ssh_utils import ssh_cmd_list
+from subprocess import Popen, PIPE
+
+import os
+import time
+import uuid
+import struct
+
+import logging
+logger = logging.getLogger(__name__)
+
+class LCURuntimeError(RuntimeError):
+    pass
+
+def wrap_command_in_lcu_head_node_ssh_call(cmd):
+    '''wrap the command in an ssh call to head.cep4
+    :param list cmd: a subprocess cmd list
+    cpu node. Otherwise, the command is executed on the head node.
+    :return: the same subprocess cmd list, but then wrapped with cep4 ssh calls
+    '''
+    return ssh_cmd_list('lcuhead.control.lofar', 'lofarsys') + cmd
+
+def wrap_command_in_lcu_station_ssh_call(cmd, station, via_head=True):
+    '''wrap the command in an ssh call the given station lcu node (via lcuhead)
+    :param list cmd: a subprocess cmd list
+    :param string station: the station name or lcu hostname (a station name is automagically converted to the lcu hostname)
+    :param bool via_head: when True, route the cmd first via the lcuhead node
+    :return: the same subprocess cmd list, but then wrapped with lcu ssh calls
+    '''
+    remote_cmd = ssh_cmd_list(stationname2hostname(station), 'lofarsys') + cmd
+    if via_head:
+        return wrap_command_in_lcu_head_node_ssh_call(remote_cmd)
+
+    return remote_cmd
+
+def get_current_stations(station_group='today', as_host_names=True):
+    '''
+    Wrapper function around the amazing lcurun and stations.txt operators system.
+    Get a list of the currently used station names, either as hostname, or as parset-like station name (default)
+    :param station_group - string: one of the predefined operator station groups, like: 'today', 'today_nl', 'core', etc. Defaults to 'today' which means all active stations.
+    :param as_host_names - bool: return the station names as ssh-able hostnames if True (like cs001c, cs002c). return the station names as parset-like VirtualInstrument.stationList names if False (like CS001, CS002).
+    :return: the station names for the given station_group as ssh-able hostnames if as_host_names=True (like cs001c, cs002c) or as parset-like VirtualInstrument.stationList names if as_host_names=False (like CS001, CS002).
+    '''
+    cmd = ['cat', '/opt/operations/bin/stations.txt']
+    cmd = wrap_command_in_lcu_head_node_ssh_call(cmd)
+    logger.debug('executing cmd: %s', ' '.join(cmd))
+    proc = Popen(cmd, stdout=PIPE, stderr=PIPE)
+    out, err = proc.communicate()
+
+    if proc.returncode != 0:
+        raise LCURuntimeError("Could not fetch stations.txt file. sdterr=%s" % (err, ))
+
+    station_file_lines = out.splitlines(False)
+    station_group_filter = station_group.strip()+' '
+    station_group_line = next(l for l in station_file_lines if l.startswith(station_group_filter))
+    station_aliases = station_group_line.split(' ')[-1].split(',')
+    station_hostnames = []
+    for station_alias in station_aliases:
+        # the alias mapping is at the top of the file, so the first matching line holds the mapping
+        station_alias_line = next(l for l in station_file_lines if station_alias in l)
+        station_hostname = station_alias_line.split()[0].strip()
+        station_hostnames.append(station_hostname)
+
+    if as_host_names:
+        logger.info("stations in group '%s': %s", station_group, ' '.join(station_hostnames))
+        return station_hostnames
+
+    station_names = [hostname2stationname(x) for x in station_hostnames]
+    logger.info("stations in group '%s': %s", station_group, ' '.join(station_names))
+    return station_names
+
+def stationname2hostname(station_name):
+    '''Convert a parset-like station name to a lcu hostname, like CS001 to cs001c'''
+    # assume a hostname is encoded as stationname in lowercase with a c appended, like cs001c for CS001
+    if not station_name.islower() or not (station_name.endswith('c') or station_name.endswith('control.lofar')):
+        return station_name.lower().strip() + 'c'
+
+    #assume given station_name is already in the form of an lcu hostname
+    return station_name
+
+def hostname2stationname(station_hostname):
+    '''Convert a lcu hostname to a parset-like station name , like cs001c or cs001c.control.lofar to CS001'''
+    # assume a hostname is encoded as stationname in lowercase with a c appended, like cs001c for CS001
+    return station_name.split('.')[0].strip()[-1].upper()
+
+def get_stations_rcu_mode(stations=None):
+    '''
+    Get the current rcu mode of a station.
+    :param stations - string or list of strings: 1 or more station names, or lcu hostnames
+    :return: dict with station rcu mode integer pairs
+    '''
+
+    if stations == None:
+        stations = get_current_stations(as_host_names=True)
+    elif isinstance(stations, basestring):
+        stations = [stations]
+
+    procs = {}
+    for station in stations:
+        cmd = ["rspctl", "--rcu | grep ON | awk '{ print $5 }' | grep mode | cut -c 6-6 | sort -u | head -n 1"]
+        cmd = wrap_command_in_lcu_station_ssh_call(cmd, station, via_head=True)
+        logger.debug('executing cmd: %s', ' '.join(cmd))
+        proc = Popen(cmd, stdout=PIPE, stderr=PIPE)
+        procs[station] = proc
+
+    result = {}
+    for station, proc in procs.items():
+        out, err = proc.communicate()
+
+        if proc.returncode != 0:
+            logger.warning("Could not determine rcu mode for station %s. sdterr=%s" % (station, err))
+
+        try:
+            rcu_mode = int(out.strip())
+            logger.debug('station %s is in rcumode=%s', station, rcu_mode)
+            result[station] = rcu_mode
+        except ValueError:
+            logger.warning("Could not determine rcu mode for station %s. sdterr=%s" % (station, err))
+
+    return result
+
+def get_station_calibration_tables(stations=None, antenna_set_and_filter=None):
+    '''
+    Get a dict of stationname:caltable pairs for the given list of stations (or all current stations if None given)
+    :param stations - string or list of strings: station name(s) or lcu hostname(s)
+    :param antenna_set_and_filter - string: the antenna_set name and filter'name' from the parset like: LBA_INNER-10_90, or HBA-170_230 etc. If None, then the current rcu_mode is retreived and the caltables for the current rcu mode are returned
+    :return: dict like {<station_name>: tuple(calibration_header_dict, numpy complex array)}
+    '''
+    if stations == None:
+        stations = get_current_stations(as_host_names=True)
+    elif isinstance(stations, basestring):
+        stations = [stations]
+
+    caltable_files = {}
+    caltable_procs = {}
+    caltable_postfixes = {}
+
+    # caltable files have either a LBA_INNER-10_90, or HBA-170_230 etc postfix, or a mode<nr> postfix
+    # these are essentially the same, and on the lcu's the files are equal (in fact they are symlinked.)
+    # So, depending on the knowledge of the requester, one eiter asks for an 'antenna_set_and_filter' postfix, or an 'rcu_mode' postfix,
+    # but the result is the same.
+    if antenna_set_and_filter:
+        for station in stations:
+            caltable_postfixes[station] = '-%s' % (antenna_set_and_filter,)
+        logger.info('fetching calibration table(s) for %s for stations %s', antenna_set_and_filter, ' '.join(stations))
+    else:
+        rcu_modes = get_stations_rcu_mode(stations)
+        for station, rcu_mode in rcu_modes.items(): # only loop over stations which have valid rcu_mode
+            caltable_postfixes[station] = '_mode%s' % (rcu_mode,)
+
+        logger.info('fetching calibration table(s) for rcu mode(s) %s for stations %s', ' '.join(sorted(list(set(rcu_modes.values())))), ' '.join(sorted(rcu_modes.keys())))
+    try:
+        for station, postfix in caltable_postfixes.items():
+            # fetch the caltable without intermediate saves to disk using multiple ssh's and pipes.
+            # write the result in a local temp file for further processing.
+            # local temp files are removed at end.
+            tmpfilename = '/tmp/caltable_%s_%s_%s' % (postfix, station, uuid.uuid4())
+            caltable_files[station] = tmpfilename
+            logger.debug('writing caltable for station %s in file %s', station, tmpfilename)
+            cmd = ['cat', '''/opt/lofar/etc/CalTable%s.dat''' % (postfix, )]
+            cmd = wrap_command_in_lcu_station_ssh_call(cmd, station, via_head=True)
+            logger.debug('executing cmd: %s', ' '.join(cmd))
+            tmpfile = open(tmpfilename, 'wb')
+            proc = Popen(cmd, stdout=tmpfile, stderr=PIPE, close_fds=True)
+            caltable_procs[station] = proc
+
+        # wait for all fetching procs to finish...
+        #TODO: add timeout?
+        for station, proc in caltable_procs.items():
+            out, err = proc.communicate()
+            if proc.returncode != 0:
+                logger.warning("Could not fetch calibration table for station %s. stderr=%s", station, err)
+
+        # gather results...
+        caltables = {}
+        # for each station, parse temp file
+        for station, filename in caltable_files.items():
+            try:
+                # store header and complex table in result dict
+                caltables[station] = read_station_calibration_file(filename)
+            except Exception as e:
+                # log exception anbd just continue with next station caltable file
+                logger.error("error while parsing calibration file for station %s: %s", station, e)
+
+    finally:
+        # cleanup all temp files
+        for filename in caltable_files.values():
+            try:
+                logger.debug('deleting local intermediate caltable file %s', tmpfilename)
+                os.remove(filename)
+            except OSError:
+                pass
+
+    logger.info('fetched calibration table(s) for stations %s', ' '.join(sorted(caltables.keys())))
+
+    return caltables
+
+def read_station_calibration_file(filename):
+    '''
+    read and parse a station calibration file and return a tuple of the header and a numpy complex array of cal-values
+    '''
+    try:
+        # import numpy here and not at top of file, so anybody can use lcu_utils methods, but is not forced into having numpy.
+        import numpy as np
+    except ImportError:
+        raise LCURuntimeError("Cannot interpret station calibration tables without numpy. Please install numpy.")
+
+    with open(filename, 'rb') as tmpfile:
+        rawdata = tmpfile.read()
+
+        # read header and convert it to a key-value dict
+        HEADER_END = 'HeaderStop\n' #assume unix line endings
+        header_end_idx = rawdata.find(HEADER_END) + len(HEADER_END)
+        header_dict = {}
+        for line in rawdata[:header_end_idx].splitlines():
+            if '=' in line:
+                items = line.partition('=')
+                header_dict[items[0].strip()] = items[2].strip()
+
+        # magic numbers (stolen from MAC/APL/PAC/ITRFBeamServer/src/StatCal.cc )
+        NUMBER_OF_SUBBANDS = 512
+        COMPLEX_SIZE = 2 # a complex is two doubles
+
+        header_station = header_dict.get('CalTableHeader.Observation.Station')
+        if header_station.startswith('CS') or header_station.startswith('RS'):
+            NUMBER_OF_ANTENNA = 96
+        else:
+            NUMBER_OF_ANTENNA = 192 #international
+
+        # using the magic numbers and knowledge from MAC/APL/PAC/ITRFBeamServer/src/StatCal.cc
+        # interpret the byte array as list of doubles,
+        # convert to numpy array,
+        # convert to complex, and transpose for correct axes (antenna first, then frequency)
+        fmt = '%dd' % (NUMBER_OF_ANTENNA * NUMBER_OF_SUBBANDS * COMPLEX_SIZE)
+        data = np.array(struct.unpack(fmt, rawdata[header_end_idx:header_end_idx+struct.calcsize(fmt)]))
+        data.resize(NUMBER_OF_SUBBANDS, NUMBER_OF_ANTENNA, COMPLEX_SIZE)
+
+        complexdata = np.empty(shape=(NUMBER_OF_SUBBANDS, NUMBER_OF_ANTENNA), dtype=complex)
+        complexdata.real = data[:, :, 0]
+        complexdata.imag = data[:, :, 1]
+        complexdata = complexdata.transpose()
+
+        # return tuple of header and complex table in result dict
+        return (header_dict, complexdata)
+
+
+
+if __name__ == '__main__':
+    logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.DEBUG)
+    print get_station_calibration_tables(['CS001', 'RS407'], antenna_set_and_filter='LBA_INNER-10_90') #['CS001', 'DE601'])

LCS/PyCommon/ssh_utils.py

+# Copyright (C) 2012-2015  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite 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.
+#
+# The LOFAR software suite 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 the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+logger = logging.getLogger(__name__)
+
+def ssh_cmd_list(host, user='lofarsys'):
+    '''
+    returns a subprocess compliant command list to do an ssh call to the given node
+    uses ssh option -T to disable remote pseudo terminal
+    uses ssh option -q for ssh quiet mode (no ssh warnings/errors)
+    uses ssh option -o StrictHostKeyChecking=no to prevent prompts about host keys
+    :param host: the node name or ip address
+    :param user: optional username, defaults to 'lofarsys'
+    :return: a subprocess compliant command list
+    '''
+    return ['ssh', '-T', '-q', '-o StrictHostKeyChecking=no', user+'@'+host]
+
+