LCATtlmDb.py

Go to the documentation of this file.

#!/usr/local/bin/python
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "Wrapper class for LCAT based telemetry"
__author__   = "Selim Tuvi <stuvi@slac.stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2005/09/13 00:08:27"
__updated__  = "$Date: 2006/03/09 15:28:29 $"
__version__  = "$Revision: 1.18 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC

import logging as log
import threading

# import generated Python classes from the LCAT package
import LCAT_tlm_db
import LICOS.lib.cmdTlmDb.VSC_tlm_db as VSC_tlm_db    # Generated and copied

from   LCAT_decoder import ccsdsTlmHdr
from   LCAT_msg_db import LCAT_Msg_Db

from LICOS.lib.LATconstants import FSW_UNIT_NAMES


from cmdTlmDb import TlmDb

class LCATtlmDb(TlmDb):
  """!\brief Database wrapper class for LCAT based telemetry.

  The purpose of this class is to provide helper methods for
  translating telemetry mnemonic to apid and vice versa.
  """

  def __init__(self):
    """!\brief LCATtlmDb constructor.

    """
    TlmDb.__init__(self)
    self.__dbModule = LCAT_tlm_db
    self.__dbModule2 = VSC_tlm_db
    self.__tlmNameToApid = {}
    self.__apidToTlmName = {}
    self.__populateDicts()
    self.MSG = LCAT_Msg_Db()

  def __populateDicts(self):
    """!\brief Fill in the dictionaries used by
              getApidFromName and getNameFromApid.
    """
    tlmApidDict = self.__dbModule.TlmApidDict()
    for (apid, mnem) in tlmApidDict.apids.items():
      self.__tlmNameToApid[mnem] = apid
      self.__apidToTlmName[apid] = mnem
    td2 = self.__dbModule2.TlmApidDict()
    for (apid, mnem) in td2.apids.items():
      self.__tlmNameToApid[mnem] = apid
      self.__apidToTlmName[apid] = mnem



  def getNames(self):
    """!\brief Retrieve a list of all telemetry mnemonic names.

    \return A list containing all mnemonic names
    """
    return self.__tlmNameToApid.keys()

  def getApids(self):
    """!\brief Retrieve a list of all telemetry apids.

    \return A list containing all mnemonic apids
    """
    return self.__apidToTlmName.keys()

  def getApidFromName(self, name):
    """!\brief Retrieve the apid from the telemetry mnemonic name.

    \param name Telemetry mnemonic name

    \return Telemetry apid
    """
    fn = "getApidFromName"
    if name in self.__tlmNameToApid:
      return self.__tlmNameToApid[name]
    else:
      log.warn("%s: Telemetry mnemonic %s not found in the database" % (fn, name))

  def getApidsFromNames(self, nameList):
    """!\brief Retrieve the apid list corresponding to the
              given telemetry mnemonic name list.

    \param nameList Telemetry mnemonic name list

    \return Telemetry apid list
    """
    fn = "getApidsFromNames"
    apidList = []
    for name in nameList:
      if name in self.__tlmNameToApid:
        apidList.append(self.__tlmNameToApid[name])
      else:
        log.warn("%s: Telemetry mnemonic %s not found in the database" % (fn, name))
    return apidList

  def getNameFromApid(self, apid):
    """!\brief Retrieve the name from the telemetry apid.

    \param apid Telemetry apid

    \return Telemetry mnemonic name.
    """
    if apid in self.__apidToTlmName:
      return self.__apidToTlmName[apid]

  def getDbModule(self):
    """!\brief Return the generated database module.

    \return Database module
    """
    return self.__dbModule

  def msgByName(self, name):
    """!\brief Return the message by name.

    \param name Message name

    \return Message text
    """
    return self.MSG.msg_by_name(name)

  def msgByValue(self, value):
    """!\brief Return the message by value.

    \param value Message value

    \return Message tuple containing message code and description
    """
    msgTuple = self.MSG.msg_by_value(~(1 << 31) & value)
    if msgTuple is None:
      msgTuple = ('UNDEFINED', 'Undefined message 0x%08x' % (~(1 << 31) & value))
    return msgTuple

  def msgIsSuccess(self, value):
    """!\brief Check if the message status is success.

    \param value Message value

    \return True if the value is success, False otherwise
    """
    return (value & 1) == 0

  def decodeTelemetry(self, telemetry, dump=False):
    """!\brief Decode a telemetry packet and optionally dump the payloads.

    \param telemetry Raw CCSDS packet
    \param dump      Boolean indicating whether to dump the
                     telemetry payloads

    \return Decoded packet
    """
    pkt = LICOS_VscTlmPacketFactory(telemetry)
    pkt.decode_payload()
    if dump:
      for (name, value) in self.getPayloadData(pkt):
        log.debug("  %20s: 0x%x" % (name, value))
    return pkt

  def getPayloadData(self, pkt):
    """!\brief Retrieve payload name and values.

    \param pkt Decoded packet

    \return A list containing payload (name, value) tuples
    """
    return [(m[0], pkt.get_payload(m[0])) for m in pkt.mnemList]

  def msgOutToText(self, pkt):
    """!\Form a text string out of the message tuple

    \param pkt Decoded packet

    \return String containing a text version of the message tuple
    """
    msgTuple = self.interpretMsgOut(pkt)
    return "%s: " % FSW_UNIT_NAMES[msgTuple[0]] + "%s %s %s %s %s" % msgTuple[1:]

  def interpretMsgOut(self, pkt):
    """!\brief Interprets the LCMMSGOUTC telemetry packets
              and converts them to a tuple.

    \param pkt Decoded packet

    \return Tuple containing (source, task, func, facil, name, text)
    """
    source = pkt.get_payload("LCMMSGNODE")
    task = ""
    for i in range(8):
      task+=chr(pkt.get_payload("LCMMSGTASK%d" % i))
    func = ""
    for i in range(32):
      func+=chr(pkt.get_payload("LCMMSGFUNC%d" % i))
    facil = ""
    for i in range(4):
      facil+=chr(pkt.get_payload("LCMMSGFACIL%d" % i))
    name = ""
    for i in range(8):
      name+=chr(pkt.get_payload("LCMMSGNAME%d" % i))
    text = ""
    for i in range(128):
      text+=chr(pkt.get_payload("LCMMSGTEXT%d" % i))
    return  (source, task.rstrip(), func.rstrip(), facil.rstrip(),
             name.rstrip(), text.rstrip())

def LICOS_TlmPacketFactory(packet):
  """!\brief Take a binary string and create a packet from it

  \param packet   A binary representation of a CCSDS string
  \return A decodable packet object
  """
  import struct
  theShort = struct.unpack('!H', packet[0:2])
  apid = theShort[0] & 0x7ff
  try:
    pkt = ApidTlmClassMap[apid](packet)
    return pkt
  except:
    return None

def LICOS_VscTlmPacketFactory(packet):
  """!\brief Take a VSC telemetry object and create a packet from it

  \param packet   A VSC telemetry object
  \return A decodable packet object
  """
  pkt = None
  apid = packet.apid()
  try:
    pkt = ApidTlmClassMap[apid](packet)
  except:
    return None
  try:
    pkt.decode_vsc(packet)
  except Exception,e:
    print "LICOS tlm pkt apid", hex(apid), "decode error", e
    return None
  return pkt

# Generate a map of apid to telemetry classes to be used with the
# factory method above.
ApidTlmClassMap = {}
for i in LCAT_tlm_db.TlmApidDict().apids:
  ApidTlmClassMap[i] = getattr(LCAT_tlm_db, "LCAT_TlmPacket%03x" % i)

# JHP By hand, add the VSC telemetry info to the proper objects:
# 1: dispatch map
for i in VSC_tlm_db.TlmApidDict().apids:
  ApidTlmClassMap[i] = getattr(VSC_tlm_db, "LCAT_TlmPacket%03x" % i)

# 2: Mnemonic dicts
# print dir(LCAT_tlm_db.TlmMnemonicDict)
# print dir(VSC_tlm_db.TlmMnemonicDict)
# LCAT_tlm_db.TlmMnemonicDict.mnemonics["VSC"] = VSC_tlm_db.TlmMnemonicDict.mnemonics["VSC"]
# LCAT_tlm_db.TlmMnemonicDict.desc += VSC_tlm_db.TlmMnemonicDict.desc

# 3: bitfield definitions
#for i in VSC_tlm_db.LCAT_tlmPacket.bitDict:
#  LCAT_tlm_db.LCAT_tlmPacket.bitDict[i] = VSC_tlm_db.LCAT_tlmPacket.bitDict[i]

---