Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_set_path(self):
if 'CMDSTAN' in os.environ:
self.assertEqual(cmdstan_path(), os.environ['CMDSTAN'])
else:
install_dir = os.path.expanduser(os.path.join('~', '.cmdstanpy'))
install_version = os.path.expanduser(
os.path.join(install_dir, get_latest_cmdstan(install_dir))
)
set_cmdstan_path(install_version)
self.assertEqual(install_version, cmdstan_path())
self.assertEqual(install_version, os.environ['CMDSTAN'])
def test_set_path(self):
if 'CMDSTAN' in os.environ:
self.assertEqual(cmdstan_path(), os.environ['CMDSTAN'])
else:
install_dir = os.path.expanduser(os.path.join('~', '.cmdstanpy'))
install_version = os.path.expanduser(
os.path.join(install_dir, get_latest_cmdstan(install_dir))
)
set_cmdstan_path(install_version)
self.assertEqual(install_version, cmdstan_path())
self.assertEqual(install_version, os.environ['CMDSTAN'])
self.assertEqual(cmdstan_path(), os.environ['CMDSTAN'])
path = os.environ['CMDSTAN']
del os.environ['CMDSTAN']
self.assertFalse('CMDSTAN' in os.environ)
set_cmdstan_path(path)
self.assertEqual(cmdstan_path(), path)
self.assertTrue('CMDSTAN' in os.environ)
else:
install_dir = os.path.expanduser(
os.path.join('~', '.cmdstanpy')
)
install_version = os.path.expanduser(
os.path.join(install_dir, get_latest_cmdstan(install_dir))
)
self.assertTrue(
os.path.samefile(cmdstan_path(), install_version)
)
self.assertTrue('CMDSTAN' in os.environ)
finally:
if cur_value is not None:
os.environ['CMDSTAN'] = cur_value
else:
if 'CMDSTAN' in os.environ:
del os.environ['CMDSTAN']
def summary(runset: RunSet) -> pd.DataFrame:
"""
Run cmdstan/bin/stansummary over all output csv files.
Echo stansummary stdout/stderr to console.
Assemble csv tempfile contents into pandasDataFrame.
:param runset: record of completed run of NUTS sampler
"""
names = runset.column_names
cmd_path = os.path.join(cmdstan_path(), 'bin', 'stansummary')
tmp_csv_file = 'stansummary-{}-{}-chains-'.format(
runset.model, runset.chains
)
fd, tmp_csv_path = tempfile.mkstemp(
suffix='.csv', prefix=tmp_csv_file, dir=TMPDIR, text=True
)
cmd = '{} --csv_file={} {}'.format(
cmd_path, tmp_csv_path, ' '.join(runset.csv_files)
)
do_command(cmd.split()) # breaks on all whitespace
summary_data = pd.read_csv(
tmp_csv_path, delimiter=',', header=0, index_col=0, comment='#'
)
mask = [x == 'lp__' or not x.endswith('__') for x in summary_data.index]
return summary_data[mask]
"""
Run cmdstan/bin/diagnose over all output csv files.
Echo diagnose stdout/stderr to console.
The diagnose utility reads the outputs of all chains
and checks for the following potential problems:
+ Transitions that hit the maximum treedepth
+ Divergent transitions
+ Low E-BFMI values (sampler transitions HMC potential energy)
+ Low effective sample sizes
+ High R-hat values
:param runset: record of completed run of NUTS sampler
"""
cmd_path = os.path.join(cmdstan_path(), 'bin', 'diagnose')
csv_files = ' '.join(runset.csv_files)
cmd = '{} {} '.format(cmd_path, csv_files)
result = do_command(cmd=cmd.split())
if result is None:
print('No problems detected.')
else:
print(result)
cmd.append('--include_paths=' + ','.join(include_paths))
print('stan to c++: make args {}'.format(cmd))
do_command(cmd)
if not os.path.exists(hpp_file):
raise Exception('syntax error'.format(stan_file))
if platform.system().lower().startswith('win'):
exe_file += '.exe'
if not overwrite and os.path.exists(exe_file):
# print('model is up to date') # notify user or not?
return Model(stan_file, exe_file)
exe_file_path = Path(exe_file).as_posix()
cmd = ['make', 'O={}'.format(opt_lvl), exe_file_path]
print('compiling c++: make args {}'.format(cmd))
try:
do_command(cmd, cmdstan_path())
except Exception:
return Model(stan_file)
return Model(stan_file, exe_file)
:param overwrite: When True, existing executable will be overwritten.
Defaults to False.
:param include_paths: List of paths to directories where Stan should look
for files to include in compilation of the C++ executable.
"""
if stan_file is None:
raise Exception('must specify argument "stan_file"')
if not os.path.exists(stan_file):
raise Exception('no such stan_file {}'.format(stan_file))
program_name = os.path.basename(stan_file)
exe_file, _ = os.path.splitext(os.path.abspath(stan_file))
hpp_file = '.'.join([exe_file, 'hpp'])
if overwrite or not os.path.exists(hpp_file):
print('translating to {}'.format(hpp_file))
stanc_path = os.path.join(cmdstan_path(), 'bin', 'stanc')
cmd = [stanc_path, '--o={}'.format(hpp_file), stan_file]
if include_paths is not None:
bad_paths = [d for d in include_paths if not os.path.exists(d)]
if any(bad_paths):
raise Exception(
'invalid include paths: {}'.format(', '.join(bad_paths))
)
cmd.append('--include_paths=' + ','.join(include_paths))
print('stan to c++: make args {}'.format(cmd))
do_command(cmd)
if not os.path.exists(hpp_file):
raise Exception('syntax error'.format(stan_file))
if platform.system().lower().startswith('win'):
exe_file += '.exe'
if not overwrite and os.path.exists(exe_file):
def diagnose(self) -> str:
"""
Run cmdstan/bin/diagnose over all output csv files.
Returns output of diagnose (stdout/stderr).
The diagnose utility reads the outputs of all chains
and checks for the following potential problems:
+ Transitions that hit the maximum treedepth
+ Divergent transitions
+ Low E-BFMI values (sampler transitions HMC potential energy)
+ Low effective sample sizes
+ High R-hat values
"""
cmd_path = os.path.join(cmdstan_path(), 'bin', 'diagnose' + EXTENSION)
cmd = [cmd_path] + self.runset.csv_files
result = do_command(cmd=cmd, logger=self.runset._logger)
if result:
self.runset._logger.info(result)
return result
self._compiler_options.validate()
self._logger.info(
'compiler options: %s', self._compiler_options
)
make = os.getenv(
'MAKE',
'make'
if platform.system() != 'Windows'
else 'mingw32-make',
)
cmd = [make]
if self._compiler_options is not None:
cmd.extend(self._compiler_options.compose())
cmd.append(Path(exe_file).as_posix())
try:
do_command(cmd, cmdstan_path(), logger=self._logger)
except RuntimeError as e:
self._logger.error(
'file %s, exception %s', stan_file, str(e)
)
compilation_failed = True
if not compilation_failed:
if is_copied:
original_target_dir = os.path.dirname(
os.path.abspath(self._stan_file)
)
new_exec_name = (
os.path.basename(os.path.splitext(self._stan_file)[0])
+ EXTENSION
)
self._exe_file = os.path.join(