Merge pull request #257 from alexander-lam/master

Cal-VanDoren · web-flow · commit 7ef746c7370c · 2024-12-23T08:57:46.000-07:00
Add Python plugin script examples
diff --git a/StkExtensionPlugins/AccessConstraints/Python/COM_PY_RangeLightingConstraint.py b/StkExtensionPlugins/AccessConstraints/Python/COM_PY_RangeLightingConstraint.py
@@ -0,0 +1,133 @@
+import typing
+import numpy as np
+from agi.stk12.plugins.accessconstraintplugin import IAgAccessConstraintPlugin, AgEAccessConstraintObjectType, AgEAccessConstraintDependencyFlags, AgEAccessApparentPositionType
+from agi.stk12.plugins.utplugin               import IAgUtPluginConfig, AgEUtLogMsgType, AgEUtFrame
+from agi.stk12.plugins.attrautomation         import AgEAttrAddFlags
+from agi.stk12.plugins.stkplugin              import AgStkPluginSite
+
+# This plugin determines the lighting condition of the background of an access object.
+# The range constraint can be dynamically set if the background is Earth and lit by direct sun,
+# background is Earth and unlit (penumbra or umbra), or if the background is deep space.
+# Other CBs are not considered. The plugin returns 1 if the range constraint for the specified
+# background is satisifed, and 0 if not.
+
+class CAgAccessConstraintPlugin(object):
+    def __init__(self):
+        self.scope = None
+        self.site = None
+        self.DayRange = 100000000
+        self.NightRange = 100000000
+        self.SpaceRange = 100000000
+        self.a = 6378137
+        self.b = 6378137
+        self.c = 6356752.31424
+        self.sunRadius = 695700000
+    
+    @property
+    def DisplayName(self) -> str:
+        '''
+        Triggered when the plugin is being registered. This is the name of the constraint used by STK.
+        The DisplayName property may alternatively be defined as a member attribute (see the STK Python API Programmer's Guide).
+        '''
+        return 'COM_PY_RangeLightingConstraint'
+
+    def Register(self, result:"IAgAccessConstraintPluginResultRegister") -> None:
+        '''
+        Triggered after application start-up, in order to register the constraint for specific STK object pairs for which this constraint is applicable.
+        '''
+        result.BaseObjectType = AgEAccessConstraintObjectType.eSensor
+        result.BaseDependency = AgEAccessConstraintDependencyFlags.eDependencyRelativePosVel + AgEAccessConstraintDependencyFlags.eDependencyPosVel + AgEAccessConstraintDependencyFlags.eDependencyRelSun
+        result.Dimension = "Unitless"
+        result.MinValue = 0.5
+        result.MaxValue = 1.0
+        result.TargetDependency = AgEAccessConstraintDependencyFlags.eDependencyRelativePosVel + AgEAccessConstraintDependencyFlags.eDependencyPosVel + AgEAccessConstraintDependencyFlags.eDependencyRelSun
+        result.AddTarget(AgEAccessConstraintObjectType.eSatellite)
+        result.AddTarget(AgEAccessConstraintObjectType.eAircraft)
+        result.Register()
+
+        result.Message(AgEUtLogMsgType.eUtLogMsgInfo, f'{self.DisplayName}: Register(Sensor to Satellite/Aircraft)')
+
+
+    def Init(self, site:"IAgUtPluginSite") -> bool:
+        '''
+        Triggered just before the first computational event trigger.
+        '''
+        self.site = AgStkPluginSite(site)
+        return True
+
+    def PreCompute(self, result:"IAgAccessConstraintPluginResultPreCompute") -> bool:
+        '''
+        Triggered prior to the calls to the Evaluate method, to allow for any required initialization.
+        '''
+        return True
+
+    def Evaluate(self, result:"IAgAccessConstraintPluginResultEval", baseData:"IAgAccessConstraintPluginObjectData", targetData:"IAgAccessConstraintPluginObjectData") -> bool:
+        '''
+        Triggered when the plugin is evaluated for an access constraint value
+        '''
+        # Compute Earth intersection point
+        posVec = np.array(baseData.Position_Array(AgEUtFrame.eUtFrameFixed))
+        losVec = np.array(baseData.RelativePosition_Array(AgEAccessApparentPositionType.eLightPathApparentPosition, AgEUtFrame.eUtFrameFixed))
+        scaleFactor = np.linalg.norm(posVec) / np.linalg.norm(losVec)
+        losVecScaled = losVec * scaleFactor
+        D = np.diag([1/self.a**2, 1/self.b**2, 1/self.c**2])
+        coeffA = np.dot(np.dot(losVecScaled, D), losVecScaled)
+        coeffB = np.dot(np.dot(posVec, D), losVecScaled)
+        coeffC = np.dot(np.dot(posVec, D), posVec)-1
+        discriminant = coeffB**2 - coeffA*coeffC
+        range = targetData.Range(AgEAccessApparentPositionType.eLightPathApparentPosition)
+        taus = [(-coeffB + np.sqrt(discriminant))/coeffA, (-coeffB - np.sqrt(discriminant))/coeffA]
+        if discriminant < 0:
+            # No intersection exists, we are looking at deep space
+            result.Value = float((range < self.SpaceRange))
+        else:
+            # Intersection exists, determine lighting condition
+            tau = min(n for n in taus if n>0)
+            earthIntersect = posVec + losVecScaled * tau
+            sens2SunVec = np.array(baseData.ApparentSunPosition_Array(AgEUtFrame.eUtFrameFixed))
+            intersect2SunVec = sens2SunVec + posVec - earthIntersect
+            intersect2SunVecHat = intersect2SunVec / np.linalg.norm(intersect2SunVec)
+            intersectNormalHat = np.dot(D,earthIntersect)/np.linalg.norm(np.dot(D,earthIntersect))
+            sunDistance = np.linalg.norm(intersect2SunVec)
+            sunDiskAngle = self.sunRadius/sunDistance
+            sunPosAngle = np.arctan2(np.dot(intersect2SunVecHat, intersectNormalHat), np.linalg.norm(np.cross(intersect2SunVecHat, intersectNormalHat)))
+            if sunPosAngle > -sunDiskAngle:
+                result.Value = float(range < self.DayRange)
+            else:
+                result.Value = float(range < self.NightRange)
+        return True
+
+    def PostCompute(self, result:"IAgAccessConstraintPluginResultPostCompute") -> bool:
+        '''
+        Triggered after the calls to the Evaluate method, to allow for any required clean up.
+        '''
+        return True
+
+    def Free(self) -> None:
+        '''
+        Triggered just before the plugin is destroyed.
+        '''
+        del(self.scope)
+        del(self.site)
+        del(self.DayRange)
+        del(self.NightRange)
+        del(self.a)
+        del(self.b)
+        del(self.c)
+        
+    def GetPluginConfig(self, pAttrBuilder:"IAgAttrBuilder") -> typing.Any:
+        '''
+        Get an attribute container of the configuration settings.
+        '''
+        if self.scope is None:
+            self.scope = pAttrBuilder.NewScope()
+            pAttrBuilder.AddQuantityMinDispatchProperty2(self.scope, "DayRange", "Range of sensor when background is lit Earth", "DayRange", "Distance", "km", "m", 0.0, AgEAttrAddFlags.eAddFlagNone)
+            pAttrBuilder.AddQuantityMinDispatchProperty2(self.scope, "NightRange", "Range of sensor when background is unlit Earth", "NightRange", "Distance", "km", "m", 0.0, AgEAttrAddFlags.eAddFlagNone)
+            pAttrBuilder.AddQuantityMinDispatchProperty2(self.scope, "SpaceRange", "Range of sensor when background is space", "SpaceRange", "Distance", "km", "m", 0.0, AgEAttrAddFlags.eAddFlagNone)
+        return self.scope
+        
+    def VerifyPluginConfig(self, pPluginCfgResult:"IAgUtPluginConfigVerifyResult") -> None:
+        '''
+        Verify the Plugin Config
+        '''
+        pass
diff --git a/StkExtensionPlugins/DensityModels/Python/VenusGRAMSuite_AtmosPlugin.py b/StkExtensionPlugins/DensityModels/Python/VenusGRAMSuite_AtmosPlugin.py
@@ -0,0 +1,240 @@
+# This implements the Venus GRAM in STK under the GRAM common framework.
+# This plugin was most recently tested with GRAM 2.1.0.
+# This plugin requires the built GRAMpy module to interface with the GRAM suite.
+# A copy of the GRAM Suite can be obtained through the NASA Technology Transfer Program.
+
+import typing, math
+from agi.stk12.plugins.utplugin         import AgEUtLogMsgType
+from agi.stk12.plugins.attrautomation   import AgEAttrAddFlags
+from agi.stk12.plugins.stkplugin        import AgStkPluginSite
+from agi.stk12.vgt                      import AgECrdnSmartEpochState
+from GRAMpy                             import gram
+
+class CAgAsDensityModelPlugin(object):
+    def __init__(self):
+        # Debug flags
+        self.DebugMode = False
+        self.PluginEnabled = True
+        self.MessageInterval = 1000
+        self.MsgCntr = -1
+        self.IntegSteps = 0
+
+        # INPUTS
+        self.CentralBodyName = 'Venus'
+        self.LowestValidAltKm = 0
+        self.SpicePath = r'\SPICE'
+        self.InitialRandomSeed = 1001
+        self.DensityPerturbationScale = 1.0
+        self.MinimumRelativeStepSize = 0.0
+        self.FastModeOn = 0
+        self.DensityType = 'Mean'
+
+        # STK specific, do not change
+        self.Scope = None
+        self.Site = None
+        self.Root = None
+        self.Density = None
+        self.DisplayName = 'VenusGRAM Suite Atmosphere Plugin'
+        self.OutputLowAltMsg = True
+        self.ComputingTemperature = True
+        self.ComputingPressure = True
+      
+        # GRAM specific, do not change
+        self.venus = None
+        self.position = None
+    
+    def Init(self, Site:"IAgUtPluginSite") -> bool:
+        self.Site = AgStkPluginSite(Site)
+        self.Root = self.Site.StkRootObject
+
+        # Parse scenario epoch
+        self.Root.UnitPreferences.SetCurrentUnit('DateFormat', 'YYYY:MM:DD') # YYYY:MM:DD:HH:MM:SS.sss Time
+        smartEpoch = self.Root.CurrentScenario.AnalysisEpoch
+        if smartEpoch.State == AgECrdnSmartEpochState.eCrdnSmartEpochStateExplicit:
+            epoch = smartEpoch.TimeInstant
+        else:
+            epoch = smartEpoch.ReferenceEvent.FindOccurrence().Epoch
+        epochSplit = epoch.split(':')
+        self.Root.UnitPreferences.SetCurrentUnit('DateFormat', 'UTCG')
+        
+        # Set up GRAM
+        inputParameters = gram.VenusInputParameters()
+        inputParameters.timeFrame = gram.PET
+        inputParameters.spicePath = self.SpicePath
+        inputParameters.initialRandomSeed = self.InitialRandomSeed
+        inputParameters.densityPerturbationScale = self.DensityPerturbationScale
+        inputParameters.minRelativeStepSize = self.MinimumRelativeStepSize
+        inputParameters.fastModeOn = self.FastModeOn
+        self.venus = gram.VenusAtmosphere()
+        self.venus.setInputParameters(inputParameters)
+
+        # Configure GRAM start time and position
+        self.position = gram.Position()
+        ttime = gram.GramTime()
+        ttime.setStartTime(int(epochSplit[0]), int(epochSplit[1]), int(epochSplit[2]), int(epochSplit[3]), int(epochSplit[4]), float(epochSplit[5]), gram.UTC, gram.PET)
+        self.venus.setStartTime(ttime)
+
+        if self.DebugMode:
+            self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, "VenusGRAMPlugin:Init()")
+            if self.PluginEnabled:
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgInfo, "VenusGRAMPlugin:Init() Enabled")
+            else:
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgInfo, "VenusGRAMPlugin:Init() Disabled because Enabled flag is False")
+        if not self.PluginEnabled:
+            self.Site.Message(AgEUtLogMsgType.eUtLogMsgAlarm, "VenusGRAMPlugin:Init() Disabled because Enabled flag is False")
+        return self.PluginEnabled
+
+    def Register(self, Result:"IAgAsDensityModelResultRegister"):
+        if self.DebugMode:
+            Result.Message(AgEUtLogMsgType.eUtLogMsgDebug, "VenusGRAMPlugin:Register()")
+    
+    def Free(self) -> bool:
+        self.Site = None
+        return True
+
+    # PreCompute is a member of the optional IAgAsDensityModelPluginExtended interface
+    def PreCompute(self, Result:"IAgAsDensityModelResult") -> bool:
+        self.IntegSteps = 0
+        self.MsgCntr = -1
+        if self.PluginEnabled:
+            self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, "PreCompute() called.")
+
+            if self.DebugMode:
+                self.TestResultInterface("PreCompute()", Result)
+            
+            return True
+        return False
+
+    # PreNextStep is a member of the optional IAgAsDensityModelPluginExtended interface
+    def PreNextStep(self, Result:"IAgAsDensityModelResult") -> bool:
+        if self.PluginEnabled:
+            RptCnt = self.MessageInterval
+            if self.DebugMode:
+                if self.IntegSteps % RptCnt == 0:
+                    self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, f'PreNextStep(): Integration Step: {self.IntegSteps}')
+                if self.IntegSteps == 0:
+                    self.TestResultInterface("PreNextStep()", Result)
+            
+            self.IntegSteps += 1
+
+            return True
+        return False
+
+    # PostCompute is a member of the optional IAgAsDensityModelPluginExtended interface
+    def PostCompute(self, Result:"IAgAsDensityModelResult") -> bool:
+        if self.PluginEnabled:
+            self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, "PostCompute() called.")
+            if self.DebugMode:
+                self.TestResultInterface("PostCompute()", Result)
+
+            return True
+        return False
+    
+    def Evaluate(self, ResultEval:"IAgAsDensityModelResultEval") -> bool:
+        self.MsgCntr += 1
+        if self.PluginEnabled:
+            self.PluginEnabled = self.SetDensity(ResultEval)
+            if self.DebugMode and self.MsgCntr == 0:
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, "Evaluate() called.")
+                self.TestResultEvalInterface("Evaluate()", ResultEval)
+
+        return self.PluginEnabled
+    
+    def CentralBody(self) -> str:
+        return self.CentralBodyName
+    
+    def ComputesTemperature(self) -> bool:
+        return self.ComputingTemperature
+    
+    def ComputesPressure(self) -> bool:
+        return self.ComputingPressure
+    
+    def UsesAugmentedSpaceWeather(self) -> bool:
+        return False
+    
+    def GetLowestValidAltitude(self) -> bool:
+        return self.LowestValidAltKm
+    
+    def OverrideAtmFluxLags(self, FluxLags: "IAgAsDensityModelPLuginAtmFluxLagsConfig") -> bool:
+        return True
+
+    def SetDensity(self, ResultEval:"IAgAsDensityModelResultEval") -> bool:
+        # Query STK for time and LLA
+        stkTime = float(ResultEval.DateString('EpSec'))
+        LLA = ResultEval.LatLonAlt_Array()
+        LatitudeDeg = math.degrees(LLA[0])
+        LongitudeDeg = math.degrees(LLA[1])
+        AltitudeKm = LLA[2] / 1000
+
+        # Validate altitude
+        if AltitudeKm < self.LowestValidAltKm:
+            if self.OutputLowAltMsg:
+                msg = f'setDensity: altitude {AltitudeKm:1.6f} is less than minimum valid altitude ( {self.LowestValidAltKm} km ). Keeping density constant below this height.'
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgWarning, f'{msg}')
+                self.OutputLowAltMsg = False
+            AltitudeKm = self.LowestValidAltKm
+
+        if self.DebugMode:
+            if self.MsgCntr % self.MessageInterval == 0:
+                msg = f'setDensity: time = {stkTime:02.6f} EpSec'
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, f'{msg}')
+                msg = f'setDensity: lat = {LatitudeDeg:1.6f}, lon = {LongitudeDeg:1.6f}, alt = {AltitudeKm:1.6f}'
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, f'{msg}')
+
+        # Set GRAM position
+        self.position.elapsedTime = stkTime        
+        self.position.latitude = LatitudeDeg
+        self.position.longitude = LongitudeDeg
+        self.position.height = AltitudeKm
+        self.venus.setPosition(self.position)
+
+        # Update GRAM atmosphere
+        self.venus.update()
+        atmos = self.venus.getAtmosphereState()
+        if self.DensityType == 'Mean':
+            self.Density = atmos.density
+        elif self.DensityType == 'Low':
+            self.Density = atmos.lowDensity
+        elif self.DensityType == 'High':
+            self.Density = atmos.highDensity
+        else:
+            self.Density = atmos.perturbedDensity
+
+        # Push result back to STK
+        ResultEval.SetDensity(self.Density)
+        ResultEval.SetPressure(atmos.pressure)
+        ResultEval.SetTemperature(atmos.temperature)
+
+        if self.DebugMode:
+            if self.MsgCntr % self.MessageInterval == 0:
+                msg = f'setDensity: density {self.Density}'
+                self.Site.Message(AgEUtLogMsgType.eUtLogMsgDebug, f'{msg}')        
+        return True
+
+    def TestResultInterface(self, Name:"str", Result:"IAgAsDensityModelResult"):
+        if not self.DebugMode:
+            return
+        return
+
+    def TestResultEvalInterface(self, Name:"str", ResultEval:"IAgAsDensityModelResultEval"):
+        if not self.DebugMode:
+            return
+        return
+
+    def GetPluginConfig(self, pAttrBuilder:"IAgAttrBuilder") -> typing.Any:
+        if self.Scope is None:
+            self.Scope = pAttrBuilder.NewScope()
+            pAttrBuilder.AddStringDispatchProperty( self.Scope, "CentralBodyName", "CentralBodyName", "CentralBodyName", AgEAttrAddFlags.eAddFlagReadOnly)
+            pAttrBuilder.AddStringDispatchProperty( self.Scope, "SpicePath", "SpicePath", "SpicePath", AgEAttrAddFlags.eAddFlagNone )
+            pAttrBuilder.AddIntMinMaxDispatchProperty( self.Scope, "InitialRandomSeed", "InitialRandomSeed", "InitialRandomSeed", 1, 29999, AgEAttrAddFlags.eAddFlagNone )
+            pAttrBuilder.AddDoubleMinMaxDispatchProperty( self.Scope, "DensityPerturbationScale", "DensityPerturbationScale", "DensityPerturbationScale", 0.0, 2.0, AgEAttrAddFlags.eAddFlagNone )
+            pAttrBuilder.AddDoubleMinMaxDispatchProperty( self.Scope, "MinimumRelativeStepSize", "MinimumRelativeStepSize", "MinimumRelativeStepSize", 0.0, 1.0, AgEAttrAddFlags.eAddFlagNone )
+            pAttrBuilder.AddBoolDispatchProperty( self.Scope, "FastModeOn", "FastModeOn", "FastModeOn", AgEAttrAddFlags.eAddFlagNone )
+            pAttrBuilder.AddChoicesDispatchProperty( self.Scope, "DensityType", "Density Type", "DensityType", ["Low", "Mean", "High", "Perturbed"] )
+        return self.Scope
+        
+    def VerifyPluginConfig(self, pPluginCfgResult:"IAgUtPluginConfigVerifyResult") -> bool:
+        pPluginCfgResult.Result = True
+        pPluginCfgResult.Message = "Ok"
+        return True
+        
