KJ135
/
ISCE_INSAR
mirror of http://172.16.0.12:5000/WBJY/ISCE_INSAR.git
4
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
ISCE_INSAR/components/iscesys/Component/TraitSeq.py

504 lines
15 KiB
Python
Executable File
Raw Blame History

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# Copyright 2012 California Institute of Technology. ALL RIGHTS RESERVED.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# United States Government Sponsorship acknowledged. This software is subject to
# U.S. export control laws and regulations and has been classified as 'EAR99 NLR'
# (No [Export] License Required except when exporting to an embargoed country,
# end user, or in support of a prohibited end use). By downloading this software,
# the user agrees to comply with all applicable U.S. export laws and regulations.
# The user has the responsibility to obtain export licenses, or other export
# authority as may be required before exporting this software to any 'EAR99'
# embargoed foreign country or citizen of those countries.
#
# Author: Ravi Lanka
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
from collections import MutableSequence
from iscesys.Component.Component import Component
import numpy as N
import re
# Factor or Parameter
FACTORY = Component.Parameter(
'_factory',
public_name='_factory',
default=None,
type=bool,
mandatory=False,
doc='Flag - Factory/Parameter'
)
# Factory Related
FACTORY_NAME = Component.Parameter(
'_factory_name',
public_name='_factorname',
default=None,
type=str,
mandatory=False,
doc='Factory Name used in the Trait Sequence'
)
MODULE_NAME = Component.Parameter(
'_module_name',
public_name='_modulename',
default=None,
type=str,
mandatory=False,
doc='Module name used in Trait Sequence'
)
# Parameter Related
CONTAINER = Component.Parameter(
'_container',
public_name='_container',
default=None,
type=str,
mandatory=False,
doc='Container Name of the Factory used in the Trait Sequence'
)
TYPE = Component.Parameter(
'_intent',
public_name='_intent',
default=None,
type=str,
mandatory=False,
doc='intent of the parameter used in the Trait Sequence'
)
TYPE = Component.Parameter(
'_type',
public_name='_type',
default=None,
type=str,
mandatory=False,
doc='Type of the parameter used in the Trait Sequence'
)
# Common Parameters
MANDATORY = Component.Parameter(
'_mandatory',
public_name='_mandatory',
default=False,
type=bool,
mandatory=False,
doc='Mandatory Field of the module used in Trait Sequence'
)
PRIVATE = Component.Parameter(
'_private',
public_name='_private',
default=True,
type=bool,
mandatory=False,
doc='Private Field of the module used in Trait Sequence'
)
NAME = Component.Parameter(
'_name',
public_name='NAME',
default=[],
container=list,
type=str,
mandatory=False,
doc='Holds the sequence of names'
)
class TraitSeq(Component, MutableSequence):
family = 'TraitSeq'
parameter_list = (FACTORY,
FACTORY_NAME,
MODULE_NAME,
CONTAINER,
TYPE,
MANDATORY,
PRIVATE,
NAME)
facility_list = ()
def __init__(self, name = ''):
super().__init__(family=self.__class__.family, name=name if name else self.__class__.family)
self.configure()
self.list = list()
self.objid = list()
self.facility_list = ()
return
def _instantiate_(self, obj):
from iscesys.Component.Configurable import Configurable
self._factory = isinstance(obj, Configurable)
if self._factory:
# Flag for element
self._factory = True
# Parse module name and factory
module_name, factory_name = TraitSeq._extractTraits_(obj)
# Setting Factory to default
self._factory_name = 'default'
self._module_name = module_name
else:
# Parameter
raise Exception("Yet to be supported")
self._factory = False
self._container = obj.container
self._intent = obj.intent
self.type = obj.type
return
def set_aux(self, obj):
if self._factory is None:
# Called for the first time to set
# objects of the class
self._instantiate_(obj)
if self._factory is True:
self._createFacility_(obj)
else:
self._createParameter_(obj)
return
def _createParameter_(self, obj):
"""
Creates Parameter class object and updates Dictionary
"""
objn = self.__getName__(obj.name)
self.objid.append(id(obj))
self._name.append(objn)
self.parameter_list += (objn,)
self.dictionaryOfVariables[objn] = {
'attrname' : objn,
'container': self._container,
'type' : self._type,
'intent' : self._intent}
setattr(self, objn, obj)
def _updateDict_(self, objn):
self._dictionaryOfFacilities[objn] = {
'attrname' : objn,
'public_name' : objn,
'factorymodule': self._module_name,
'factoryname' : self._factory_name,
'mandatory' : self._mandatory,
'private' : self._private,
'args' : (),
'kwargs' : None,
'doc' : ''}
self.dictionaryOfVariables[objn] = {
'attrname' : objn,
'type' : 'component',
'mandatory': self._mandatory,
'private' : self._private}
return
def _createFacility_(self, obj):
"""
Creates Facility class object and updates dictionary
"""
objn = self.__getName__(obj.name)
self.objid.append(id(obj))
self._name.append(objn)
self.facility_list += (objn,)
self._updateDict_(objn)
setattr(self, objn, obj)
return
def updateDict(self, obj, i, objn):
print(objn)
self.objid[i] = id(obj)
self._name[i] = objn
self._updateDict_(objn)
# Handle facility list differently as it is a tuple
cFacility = list(self.facility_list)
cFacility = objn
self.facility_list = tuple(cFacility)
return
def _copyFacility(self):
"""
Fixes the Variable of Variables to contain Facilities
"""
facility_list = list(self._dictionaryOfFacilities.keys())
variable_list = list(self.dictionaryOfVariables.keys())
for name in facility_list:
if name not in variable_list:
self.dictionaryOfVariables[name] = {
'attrname' : name,
'type' : 'component',
'mandatory': self._mandatory,
'private' : self._private}
return
def __getName__(self, name, _next_=0):
if name.lower() != 'traitseq_name':
objn = name.lower()
else:
objn = '{}{}'.format(self.name, len(self.list) + _next_)
objn = '{}{}'.format(self.name, len(self.list) + _next_)
return objn
@staticmethod
def _extractTraits_(obj):
# Parse module name and factory
module = re.findall("'([^']*)'", str(type(obj)))[0]
module_name = module.split('.')[-1]
factory_name = '.'.join(module.split('.')[:-1])
return (module_name, factory_name)
def _checkTrait_(self, obj):
'''
Checks if the element added is of the same type
as in the list
'''
#Set the ith element of self.list to value object
if self._factory is not None:
# Already the first element is added to the list
if self._factory:
module_name, factory_name = TraitSeq._extractTraits_(obj)
if (self._module_name != module_name):
raise Exception("""Incorrect object type added \
TraitSeq currently supports only objects of single type""")
else:
raise Exception('Not Yet supported')
###################
# fixes on basic methods because Configurability used properties to fetch
# some details on about facilities
###################
def renderToDictionary(self,obj,propDict,factDict,miscDict):
'''
Overloading rendering to preprocess before writting
'''
self._copyFacility()
super(Component, self).renderToDictionary(obj,propDict,factDict,miscDict)
return
def initRecursive(self,dictProp,dictFact):
'''
Fixing Properties dictionary before initializing
'''
self._copyFacility()
super(Component, self).initRecursive(dictProp,dictFact)
try:
# Fixing object ID and the list
if len(self._name) != len(self.objid):
self.objid = []
self.list = []
for name in self._name:
obj = getattr(self, name.lower())
cid = id(obj)
self.objid.append(cid)
self.list.append(obj)
except:
# Elements not initialized from xml
pass
##################
# List Methods
##################
def __add__(self, other):
#Add lists contained in other TraitSeq object
if self._checkEQ_(other):
for i in range(len(other)):
self.append(other.list[i])
else:
raise Exception("""Object are of different types
TraitSeq currently supports only objects of a single type""")
return self
def __contains__(self, x):
#Check if x is contained in self.list
return x in self.list
def __delitem__(self, i, flag=True):
#Delete item at index i from self.list
#Update the Component dictionaries and facility_list
if flag:
del self.list[i]
del self.dictionaryOfVariables[self._name[i]]
del self._dictionaryOfFacilities[self._name[i]]
del self._name[i]
del self.objid[i]
# Handle facility list differently as it is a tuple
cFacility = list(self.facility_list)
del cFacility[i]
self.facility_list = tuple(cFacility)
return
def __getitem__(self, i):
#Return the item in self.list at index i
return self.list[i]
def __len__(self):
#Return the length of self.list
return len(self.list)
def __str__(self):
#Return a string version of self.list
return str(self.list)
def __setitem__(self, i, obj):
self._checkTrait_(obj)
self.list[i] = obj
name = self.__getName__(obj.name, _next_=1)
setattr(self, name, obj)
self.objid = id(obj)
if self._name[i] != name:
# Update Facility List
cFacility = list(self.facility_list)
cFacility[i] = name
self.facility_list = tuple(cFacility)
# Remove old
del self.dictionaryOfVariables[self._name[i]]
del self._dictionaryOfFacilities[self._name[i]]
self._updateDict_(name)
self._name[i] = name
return
def append(self, obj):
#Append an element to self.list
self._checkTrait_(obj)
self.list.append(obj)
self.set_aux(obj)
def clear(self):
#Clear all items from self.list
self.list.clear()
self.dictionaryOfVariables.clear()
self._dictionaryOfFacilities.clear()
self._name.clear()
self.objid.clear()
# Handle facility list differently as it is a tuple
self.facility_list = ()
return
def copy(self):
#Return a copy of self.list
return self.copy()
def count(self, x):
#return count of how many times x occurs in self.list
return self.list.count(x)
def extend(self, other):
#Extend self.list with other list
raise Exception('Not Yet supported')
self.list.extend(other)
def index(self, x):
#return the index of x in self.list;
return self.list.index(x)
def insert(self, i, v):
self._checkTrait_(v)
self.list.insert(i, v)
objn = self.__getName__(v.name)
setattr(self, objn, v)
self._updateDict_(objn)
# Update Facility List
self._name.insert(i, objn)
self.objid.insert(i, id(v))
cFacility = list(self.facility_list)
cFacility.insert(i, objn)
self.facility_list = tuple(cFacility)
return
def pop(self, i=None):
#pop item off the specified index if given, else off the end of list
self.__delitem__(i if i else len(self)-1)
return
def remove(self, x):
#remove item x from the list
self.list.remove(x)
flag = False
# Update bookmark list
cidx = [id(x) for x in self.list]
setdiff = [obj for obj in self.objid + cidx if obj not in cidx]
if (len(setdiff) == 1):
self.__delitem__(self.objid.index(setdiff[0]), flag)
else:
raise Exception('Not Yet supported')
return
def reverse(self):
#reverse the items in the list
self.list.reverse()
self.facility_list = self.facility_list[::-1]
self._name.reverse()
self.objid.reverse()
return
@staticmethod
def _orderSeq_(x, idx):
if len(x) != len(idx):
raise Exception('Index of different length')
x = N.array(x)
return list(x[N.array(idx, dtype=int)])
def sort(self, key=None):
#Sort self.list according to the ordering relations (lt, gt, eq) of
#the type of elements in self.list.
self.list.sort(key=key)
# Find the order to update dictionary
pid = N.array(self.objid)
cid = N.empty((len(self.list)))
for i, obj in enumerate(self.list):
cid[i] = N.where(pid == id(obj))[0][0]
# Update internal list for proper sequencing
self._name = self._orderSeq_(self._name, cid)
self.objid = self._orderSeq_(self.objid, cid)
self.facility_list = tuple(self._orderSeq_(self.facility_list, cid))
return
def __eq__(self, other):
return self.list == other.list
def _checkEQ_(self, other):
if self._factory:
return ((self._module_name, self._factory_name, self._mandatory, self._private) ==
(other._module_name, other._factory_name, other._mandatory, other._private))
else:
return ((self._container, self._type, self._intent) == \
(other._container, other._type, other._intent))
Reference in New Issue View Git Blame Copy Permalink