# Licensed under a 3-clause BSD style license - see LICENSE.rstimportjsonimportloggingfromcollectionsimportdefaultdictfromenumimportEnumfrompathlibimportPathfromtypingimportListfromastropy.coordinatesimportAnglefromastropy.timeimportTimefromastropy.unitsimportQuantityimportyamlfrompydanticimportBaseModelfrompydantic.utilsimportdeep_updatefromgammapy.makersimportMapDatasetMakerfromgammapy.utils.scriptsimportmake_path,read_yaml__all__=["AnalysisConfig"]CONFIG_PATH=Path(__file__).resolve().parent/"config"DOCS_FILE=CONFIG_PATH/"docs.yaml"log=logging.getLogger(__name__)classAngleType(Angle):@classmethoddef__get_validators__(cls):yieldcls.validate@classmethoddefvalidate(cls,v):returnAngle(v)classEnergyType(Quantity):@classmethoddef__get_validators__(cls):yieldcls.validate@classmethoddefvalidate(cls,v):v=Quantity(v)ifv.unit.physical_type!="energy":raiseValueError(f"Invalid unit for energy: {v.unit!r}")returnvclassTimeType(Time):@classmethoddef__get_validators__(cls):yieldcls.validate@classmethoddefvalidate(cls,v):returnTime(v)classReductionTypeEnum(str,Enum):spectrum="1d"cube="3d"classFrameEnum(str,Enum):icrs="icrs"galactic="galactic"classRequiredHDUEnum(str,Enum):events="events"gti="gti"aeff="aeff"bkg="bkg"edisp="edisp"psf="psf"rad_max="rad_max"classBackgroundMethodEnum(str,Enum):reflected="reflected"fov="fov_background"ring="ring"classSafeMaskMethodsEnum(str,Enum):aeff_default="aeff-default"aeff_max="aeff-max"edisp_bias="edisp-bias"offset_max="offset-max"bkg_peak="bkg-peak"classMapSelectionEnum(str,Enum):counts="counts"exposure="exposure"background="background"psf="psf"edisp="edisp"classGammapyBaseConfig(BaseModel):classConfig:validate_all=Truevalidate_assignment=Trueextra="forbid"json_encoders={Angle:lambdav:f"{v.value}{v.unit}",Quantity:lambdav:f"{v.value}{v.unit}",Time:lambdav:f"{v.value}",}classSkyCoordConfig(GammapyBaseConfig):frame:FrameEnum=Nonelon:AngleType=Nonelat:AngleType=NoneclassEnergyAxisConfig(GammapyBaseConfig):min:EnergyType=Nonemax:EnergyType=Nonenbins:int=NoneclassSpatialCircleConfig(GammapyBaseConfig):frame:FrameEnum=Nonelon:AngleType=Nonelat:AngleType=Noneradius:AngleType=NoneclassEnergyRangeConfig(GammapyBaseConfig):min:EnergyType=Nonemax:EnergyType=NoneclassTimeRangeConfig(GammapyBaseConfig):start:TimeType=Nonestop:TimeType=NoneclassFluxPointsConfig(GammapyBaseConfig):energy:EnergyAxisConfig=EnergyAxisConfig()source:str="source"parameters:dict={"selection_optional":"all"}classLightCurveConfig(GammapyBaseConfig):time_intervals:TimeRangeConfig=TimeRangeConfig()energy_edges:EnergyAxisConfig=EnergyAxisConfig()source:str="source"parameters:dict={"selection_optional":"all"}classFitConfig(GammapyBaseConfig):fit_range:EnergyRangeConfig=EnergyRangeConfig()classExcessMapConfig(GammapyBaseConfig):correlation_radius:AngleType="0.1 deg"parameters:dict={}energy_edges:EnergyAxisConfig=EnergyAxisConfig()classBackgroundConfig(GammapyBaseConfig):method:BackgroundMethodEnum=Noneexclusion:Path=Noneparameters:dict={}classSafeMaskConfig(GammapyBaseConfig):methods:List[SafeMaskMethodsEnum]=[SafeMaskMethodsEnum.aeff_default]parameters:dict={}classEnergyAxesConfig(GammapyBaseConfig):energy:EnergyAxisConfig=EnergyAxisConfig(min="1 TeV",max="10 TeV",nbins=5)energy_true:EnergyAxisConfig=EnergyAxisConfig(min="0.5 TeV",max="20 TeV",nbins=16)classSelectionConfig(GammapyBaseConfig):offset_max:AngleType="2.5 deg"classWidthConfig(GammapyBaseConfig):width:AngleType="5 deg"height:AngleType="5 deg"classWcsConfig(GammapyBaseConfig):skydir:SkyCoordConfig=SkyCoordConfig()binsize:AngleType="0.02 deg"width:WidthConfig=WidthConfig()binsize_irf:AngleType="0.2 deg"classGeomConfig(GammapyBaseConfig):wcs:WcsConfig=WcsConfig()selection:SelectionConfig=SelectionConfig()axes:EnergyAxesConfig=EnergyAxesConfig()classDatasetsConfig(GammapyBaseConfig):type:ReductionTypeEnum=ReductionTypeEnum.spectrumstack:bool=Truegeom:GeomConfig=GeomConfig()map_selection:List[MapSelectionEnum]=MapDatasetMaker.available_selectionbackground:BackgroundConfig=BackgroundConfig()safe_mask:SafeMaskConfig=SafeMaskConfig()on_region:SpatialCircleConfig=SpatialCircleConfig()containment_correction:bool=TrueclassObservationsConfig(GammapyBaseConfig):datastore:Path=Path("$GAMMAPY_DATA/hess-dl3-dr1/")obs_ids:List[int]=[]obs_file:Path=Noneobs_cone:SpatialCircleConfig=SpatialCircleConfig()obs_time:TimeRangeConfig=TimeRangeConfig()required_irf:List[RequiredHDUEnum]=["aeff","edisp","psf","bkg"]classLogConfig(GammapyBaseConfig):level:str="info"filename:Path=Nonefilemode:str=Noneformat:str=Nonedatefmt:str=NoneclassGeneralConfig(GammapyBaseConfig):log:LogConfig=LogConfig()outdir:str="."n_jobs:int=1datasets_file:Path=Nonemodels_file:Path=None
[docs]classAnalysisConfig(GammapyBaseConfig):"""Gammapy analysis configuration."""general:GeneralConfig=GeneralConfig()observations:ObservationsConfig=ObservationsConfig()datasets:DatasetsConfig=DatasetsConfig()fit:FitConfig=FitConfig()flux_points:FluxPointsConfig=FluxPointsConfig()excess_map:ExcessMapConfig=ExcessMapConfig()light_curve:LightCurveConfig=LightCurveConfig()def__str__(self):"""Display settings in pretty YAML format."""info=self.__class__.__name__+"\n\n\t"data=self.to_yaml()data=data.replace("\n","\n\t")info+=datareturninfo.expandtabs(tabsize=4)
[docs]@classmethoddefread(cls,path):"""Reads from YAML file."""config=read_yaml(path)returnAnalysisConfig(**config)
[docs]@classmethoddeffrom_yaml(cls,config_str):"""Create from YAML string."""settings=yaml.safe_load(config_str)returnAnalysisConfig(**settings)
[docs]defwrite(self,path,overwrite=False):"""Write to YAML file."""path=make_path(path)ifpath.exists()andnotoverwrite:raiseIOError(f"File exists already: {path}")path.write_text(self.to_yaml())
[docs]defto_yaml(self):"""Convert to YAML string."""# Here using `dict()` instead of `json()` would be more natural.# We should change this once pydantic adds support for custom encoders# to `dict()`. See https://github.com/samuelcolvin/pydantic/issues/1043config=json.loads(self.json())returnyaml.dump(config,sort_keys=False,indent=4,width=80,default_flow_style=None)
[docs]defset_logging(self):"""Set logging config. Calls ``logging.basicConfig``, i.e. adjusts global logging state. """self.general.log.level=self.general.log.level.upper()logging.basicConfig(**self.general.log.dict())log.info("Setting logging config: {!r}".format(self.general.log.dict()))
[docs]defupdate(self,config=None):"""Update config with provided settings. Parameters ---------- config : string dict or `AnalysisConfig` object Configuration settings provided in dict() syntax. """ifisinstance(config,str):other=AnalysisConfig.from_yaml(config)elifisinstance(config,AnalysisConfig):other=configelse:raiseTypeError(f"Invalid type: {config}")config_new=deep_update(self.dict(exclude_defaults=True),other.dict(exclude_defaults=True))returnAnalysisConfig(**config_new)
@staticmethoddef_get_doc_sections():"""Returns dict with commented docs from docs file"""doc=defaultdict(str)withopen(DOCS_FILE)asf:forlineinfilter(lambdaline:notline.startswith("---"),f):line=line.strip("\n")ifline.startswith("# Section: "):keyword=line.replace("# Section: ","")doc[keyword]+=line+"\n"returndoc