Skip to main content

Data sampling (DBN) in Python

# __author__ = 'Bayes Server'
# __version__= '0.1'

import pandas as pd
import jpype # pip install jpype1 (version 1.2.1 or later)
import jpype.imports
from jpype.types import *

classpath = "lib/bayesserver-10.8.jar" # TODO download the Bayes Server Java API, and adjust the path

# Launch the JVM

import data_frame_utils as dfu

# import the Java modules
from com.bayesserver import *
from com.bayesserver.inference import *
from com.bayesserver.learning.parameters import *
from import *
from import *
from jpype import java, JImplements, JOverride

# Uncomment the following line and change the license key, if you are using a licensed version
# License.validate("xxx")

def nullable_int(x):
""" Helper function to convert an integer to a java nullable integer (java.lang.Integer) """
return java.lang.Integer(x)

def value_as_text(variable, time, evidence):

time_nullable = nullable_int(time)

if evidence.getEvidenceType(variable, time_nullable) == EvidenceType.NONE:
return "(null)"

if variable.getValueType() == VariableValueType.CONTINUOUS:
return f"{evidence.get(variable, time_nullable)}"
return variable.getStates().get(evidence.getState(variable, time_nullable)).getName()

# we manually construct the network here, but it could be loaded from a file
network = Network()

# TODO download network from the Bayes Server User Interface (or Bayes Server Online)
# and adjust the following path
network.load("Tutorial 3 - Time series network.bayes")

x1 = network.getVariables().get("X1", True)
x2 = network.getVariables().get("X2", True)

fixedEvidence = DefaultEvidence(network)

# You can optionally set evidence on 'fixedEvidence' if you wish to fix
# certain variables, as follows
fixedEvidence.set(x2, 3.4, nullable_int(3))

# prepare to sample data from the Bayesian network
sampler = DataSampler(network, fixedEvidence)
options = DataSamplingOptions()

# If you want to simulate missing data, you can use the following line of code...
# options.setMissingDataProbability(0.05) # set 5% of the data to missing

random = java.util.Random(0)
sample = DefaultEvidence(network) # acts like a buffer to receive each sample

for i in range(10):
print(f"Sample {i}")

sequence_length = 5 # This can be varied per sample if required
sampler.takeSample(sample, random, options)

for t in range(sequence_length):
print(f"\t{t}\t{value_as_text(x1, t, sample)}\t{value_as_text(x2, t, sample)}")

except InconsistentEvidenceException:
print("Inconsistent evidence exception was raised.")