Enum.py

download

# @(#) $Id: Enum.py,v 1.1 2004/10/29 21:56:25 dfugate Exp $
#
# Copyright (C) 2001
# Associated Universities, Inc. Washington DC, USA.
#
# Produced for the ALMA project
#
# This library is free software; you can redistribute it and/or modify it under
# the terms of the GNU Library General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option) any
# later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more
# details.
#
# You should have received a copy of the GNU Library General Public License
# along with this library; if not, write to the Free Software Foundation, Inc.,
# 675 Massachusetts Ave, Cambridge, MA 02139, USA. Correspondence concerning
# ALMA should be addressed as follows:
#
# Internet email: alma-sw-admin@nrao.edu
# "@(#) $Id: Enum.py,v 1.1 2004/10/29 21:56:25 dfugate Exp $"
#
# who when what
# -------- ---------- ----------------------------------------------
# dfugate 2004/07/21 Created.
#------------------------------------------------------------------------------

'''
This module provides an implementation of the Ppattern IDL interface:

TODO:
- Testing
'''

__version__ = "$Id: Enum.py,v 1.1 2004/10/29 21:56:25 dfugate Exp $"

#--REGULAR IMPORTS-------------------------------------------------------------
#--CORBA STUBS-----------------------------------------------------------------
import ACS__POA
from ACS import RED, YELLOW, GREEN, GREY #Condition enumerations
from ACS import CBDescOut
import CORBA
#--ACS Imports-----------------------------------------------------------------
from ACSImpl.GenericProperty import GenericProperty
from ACSImpl.Monitors import Monitorpattern
from Acspy.Util.ACSCorba import interfaceRepository
#--GLOBALS---------------------------------------------------------------------

#------------------------------------------------------------------------------
#--P property------------------------------------------------------------------
#------------------------------------------------------------------------------
class Penum(GenericProperty):
'''
Properties can be derived from Penum only if their IDL derives from
ACS::Penum.
'''
#--------------------------------------------------------------------------
def __init__(self, name, charCompRef, devIORef):
'''
Constructor

Params:
- name is the quite literally the name of the property
- charCompRef is the characteristic component object which contains this
property
- devIORef is a reference to a DevIO to be used with this property

Returns: Nothing

Raises: Nothing.
'''
#------------------------------------------------------------------------------
#determine the real type of the BACI property using the IFR
#First determine the property's name
self.enumIFRName = None

IR = interfaceRepository()
interf = IR.lookup_id(self._NP_RepositoryId)

#determine the underlying enumeration type
for attr in interf._narrow(CORBA.InterfaceDef).describe_interface().attributes:
#we can use the default_value to get at an instance of the real enumeration
if attr.name == "default_value":
#ifr name of the enum
self.enumIFRName = attr.type.id()
break

#determine the IDL type
self.enumName = self.enumIFRName.split(':')[1].split('/')[1:]
self.modName = self.enumName[0]

#convert the list to a stringified Python package structure which can
#be used with eval
self.enumName = reduce((lambda x, y : str(x) + '.' + str(y)), self.enumName)

#now comes the complicated part...importing the correct CORBA stub
#without polluting the local namespace...
self.tGlobals = {}
self.tLocals = {}
exec "import " + self.modName in self.tGlobals, self.tLocals

GenericProperty.__init__(self, name, charCompRef, devIORef)

return
#--------------------------------------------------------------------------
def coerceToPropertyType(self, value=None):
'''
Overriden.
'''
#something went wrong. Return default value
if value==None:
#just use the first enum!
return eval(self.enumName + "._items[0]", self.tGlobals, self.tLocals)

try:
#coerce into an enum type
return eval(self.enumName + "." + value, self.tGlobals, self.tLocals)
except:
#warn them about CDB access
self.getLogger().logAlert("Unble to coerce '" + str(value) + "' into the correct type!", self)
#return an acceptable default value instead...an empty sequence
return eval(self.enumName + "._items[0]", self.tGlobals, self.tLocals)
#--------------------------------------------------------------------------
def getMonitorObject(self, scheduler, timeoutID):
'''
Helper method.
'''
return Monitorpattern(scheduler, timeoutID)
#--------------------------------------------------------------------------
def monitorRunner(self, cb, desc):
'''
Helper method
'''
allList = self._get_allStates()
(value, compl) = self.get_sync()
cb.working(long(allList.index(value)), compl, CBDescOut(0L, desc.id_tag))
return
#--------------------------------------------------------------------------
def _get_statesDescription(self):
'''
Implementation of the IDL attribute.

readonly attribute stringSeq bitDescription;
'''
try:
#should be in the form "abc, xyz, nypd, etc"
retVal = str(self.getCDBDict()['statesDescription'])
#now in the form of "abc,xyz,nypd,etc" (e.g., spaced removed)
retVal = retVal.replace(" ", "").replace("\t","")
#now it has been transformed into ["abc", "xyz", "nypd", "etc"]
retVal = retVal.split(",")
return retVal

except:
#warn them about CDB access
self.getLogger().logInfo("Some problem occurred when attempting to retrieve data from the ACS CDB", self)
#return an acceptable default value instead.
return []
#--------------------------------------------------------------------------
def getConditionSeq(self, attrName):
'''
Helper method which gets a list of ConditionSeq enumerations from the CDB.

Parameters: attrName is the name of the ConditionSeq attribute within the
CDB

Returns: a ConditionSeq

Raises: Nothing
'''
try:
#should be in the form "RED, YELLOW"
retVal = str(self.getCDBDict()[attrName])
#now in the form of "RED,YELLOW" (e.g., spaced removed)
retVal = retVal.replace(" ", "").replace("\t","")
#now it has been transformed into ["RED", "YELLOW"]
retVal = retVal.split(",")

#convert the strings to real enumerations!
for i in range(0, len(retVal)):
#sanity check first
if (retVal[i]!="0")and(retVal[i]!="1")and(retVal[i]!="2")and(retVal[i]!="3"):
if (retVal[i].upper()!="RED")and(retVal[i].upper()!="YELLOW")and(retVal[i].upper()!="GREEN")and(retVal[i].upper()!="GREY"):
self.getLogger().logCritical("The '" + str(retVal[i]) +
"' value retrieved from the CDB is invalid. Returning an empty ConditionSeq!", self)
return []
elif (retVal[i]=="0"):
retVal[i] = "RED"
elif (retVal[i]=="1"):
retVal[i] = "YELLOW"
elif (retVal[i]=="2"):
retVal[i] = "GREEN"
else:
retVal[i] = "GREY"

retVal[i] = eval(retVal[i])

return retVal

except:
#warn them about CDB access
self.getLogger().logInfo("Some problem occurred when attempting to retrieve data from the ACS CDB", self)
#return an acceptable default value instead.
return []
#--------------------------------------------------------------------------
def _get_condition(self):
'''
Implementation of IDL attribute.

readonly attribute ACS::ConditionSeq condition;
'''
return self.getConditionSeq("condition")
#--------------------------------------------------------------------------
def _get_allStates(self):
'''
Implementation of IDL attribute.

readonly attribute ConditionSeq whenCleared;
'''
return eval("list(" + self.enumName + "._items)", self.tGlobals, self.tLocals)
#--------------------------------------------------------------------------
def _get_alarm_on(self):
'''
Implementation of the IDL attribute.

readonly attribute (unknown type) alarm_on;
'''
try:
return [self.coerceToPropertyType(str(self.getCDBDict()['alarm_on']))]
except:
#warn them about CDB access
self.getLogger().logInfo("Some problem occurred when attempting to retrieve data from the ACS CDB", self)
#return an acceptable default value instead.
return [self.coerceToPropertyType(None)]
#--------------------------------------------------------------------------
def _get_alarm_off(self):
'''
Implementation of the IDL attribute.

readonly attribute (unknown type) alarm_off;
'''
try:
return [self.coerceToPropertyType(str(self.getCDBDict()['alarm_off']))]
except:
#warn them about CDB access
self.getLogger().logInfo("Some problem occurred when attempting to retrieve data from the ACS CDB", self)
#return an acceptable default value instead.
return [self.coerceToPropertyType(None)]
#--------------------------------------------------------------------------
def get_async(self, cb, desc):
'''
Implementation of the IDL method.

void get_async (in CBstringSeq cb, in CBDescIn desc);
'''
self.getLogger().logInfo('Not really asynchronous for now...', self)
retVal, compl = self.get_sync()
allList = self._get_allStates()

cb.done(long(allList.index(retVal)),
compl,
CBDescOut(0L, desc.id_tag))
return
#------------------------------------------------------------------------------
_enumClassDict = {}

def getEnumClass(repositoryID):
'''
'''
global _enumClassDict

if _enumClassDict.has_key(repositoryID):
return _enumClassDict[repositoryID]

#first convert the repos ID to the name of a Python class
enumName = repositoryID.split(':')[1].split('/')[1:]
enumName[0] = enumName[0] + "__POA"
modName = enumName[0]
shortEnumName = enumName[len(enumName) -1]
enumName = reduce((lambda x, y : str(x) + '.' + str(y)), enumName)

#next import the correct CORBA stub module
tGlobals = {}
tLocals = {}
exec "import " + modName in tGlobals, tLocals

class tClass(eval(enumName, tGlobals, tLocals), Penum):
'''
'''
pass

_enumClassDict[repositoryID] = tClass

return tClass
#------------------------------------------------------------------------------
if __name__ == "__main__":
joe = getEnumClass("IDL:alma/acsexmplCalendar/RWDaysEnum:1.0")
bar = joe(None, None, None)
print "DWF"

download