[docs]classVerosDiagnostic(metaclass=abc.ABCMeta):"""Base class for diagnostics. Provides an interface and wrappers for common I/O. Any diagnostic needs to implement the 5 interface methods and set some attributes. """name=None#: Name that identifies the current diagnosticsampling_frequency=0.0output_frequency=0.0output_path=Noneoutput_variables=Nonevar_meta=None#: Metadata of internal variablesextra_dimensions=None#: Dict of extra dimensions used in var_metadef__init__(self,state):pass
[docs]@abc.abstractmethoddefinitialize(self,state):"""Called at the end of setup. Use this to process user settings and handle setup."""pass
[docs]@abc.abstractmethoddefdiagnose(self,state):"""Called with frequency ``sampling_frequency``."""pass
[docs]@abc.abstractmethoddefoutput(self,state):"""Called with frequency ``output_frequency``."""pass
definitialize_variables(self,state):ifself.var_metaisNone:self.variables=Nonereturndimensions=dict(state.dimensions)ifself.extra_dimensionsisnotNone:dimensions.update(self.extra_dimensions)self.variables=VerosVariables(self.var_meta,dimensions)# we leave diagnostic variables unlockedself.variables.__locked__=Falsedefget_output_file_name(self,state):statedict=dict(state.variables.items())statedict.update(state.settings.items())returnself.output_path.format(**statedict)@do_not_disturbdefinitialize_output(self,state):inactive=notself.output_frequencyandnotself.sampling_frequencyno_output=notself.output_pathornotself.output_variablesifruntime_settings.diskless_modeorinactiveorno_output:returnoutput_path=self.get_output_file_name(state)ifos.path.isfile(output_path)andnotruntime_settings.force_overwrite:raiseIOError(f'output file {output_path} for diagnostic "{self.name}" exists '"(change output path or enable force_overwrite runtime setting)")# possible race condition ahead!distributed.barrier()withnctools.threaded_io(output_path,"w")asoutfile:nctools.initialize_file(state,outfile,extra_dimensions=self.extra_dimensions)forkeyinself.output_variables:var=self.var_meta[key]ifkeynotinoutfile.variables:nctools.initialize_variable(state,key,var,outfile)ifnotvar.time_dependent:var_data=self.variables.get(key)nctools.write_variable(state,key,var,var_data,outfile)@do_not_disturbdefwrite_output(self,state):vs=state.variablesifruntime_settings.diskless_mode:returnwithnctools.threaded_io(self.get_output_file_name(state),"r+")asoutfile:current_days=time.convert_time(vs.time,"seconds","days")nctools.advance_time(current_days,outfile)forkeyinself.output_variables:var=self.var_meta[key]var_data=self.variables.get(key)nctools.write_variable(state,key,var,var_data,outfile)