Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if len(init) != num_chains:
raise ValueError("Initial values must be provided for each chain.")
payloads = []
for chain in range(1, num_chains + 1):
payload = {"function": "stan::services::sample::hmc_nuts_diag_e_adapt"}
payload.update(kwargs)
payload["chain"] = chain
payload["data"] = self.data
payload["init"] = init.pop(0)
if self.random_seed is not None:
payload["random_seed"] = self.random_seed
# fit needs to know num_samples, num_warmup, num_thin, save_warmup
# progress bar needs to know some of these
num_warmup = payload.get("num_warmup", arguments.lookup_default(arguments.Method["SAMPLE"], "num_warmup"))
num_samples = payload.get(
"num_samples", arguments.lookup_default(arguments.Method["SAMPLE"], "num_samples"),
)
num_thin = payload.get("num_thin", arguments.lookup_default(arguments.Method["SAMPLE"], "num_thin"))
save_warmup = payload.get(
"save_warmup", arguments.lookup_default(arguments.Method["SAMPLE"], "save_warmup"),
)
payloads.append(payload)
def extract_protobuf_messages(fit_bytes):
varint_decoder = google.protobuf.internal.decoder._DecodeVarint32
next_pos, pos = 0, 0
while pos < len(fit_bytes):
msg = callbacks_writer_pb2.WriterMessage()
next_pos, pos = varint_decoder(fit_bytes, pos)
msg.ParseFromString(fit_bytes[pos : pos + next_pos])