class_docstr: This class uses Expectation-Maximization to learn parameters of a single latent variable in a bayesian network.
  We do so by also allowing the user to CONSTRAINT the optimisation
  These are elements that help the algorithm find a local optimal point closer to the point we think
  
  The setting is:
  Input:
      - a StructureModel representing the whole network or any sub-graph containing the Markov Blanket of the LV
      - data as a dataframe. The LV must be in the dataframe, with missing values represented by `np.nan`s
      - constraints:
          - Box - A hard constraint; forbids the solution to be outside of certain boundaries
          - Priors - establishes Dirichlet priors to every parameter
  run:
      - using the method `run` or manually alternating over E and M steps)
  Result:
      - CPTs involving the latent variable, learnt by EM, found in the attribute `cpds`
      - CPTs not involving the LV not learned (They must be learned separately by MLE.
      This is faster and the result is the same)
  
  Example:
  >>> em = EMSingleLatentVariable(sm=sm, data=data, lv_name=lv_name, node_states=node_states)
  >>> em.run() # run EM until convergence
  >>> # or run E and M steps separately
  >>> for i in range(10): # Run EM 10 times
  >>>     em.e_step()
  >>>     em.m_step()
docstr: Args:
      sm: structure. Only requirement is: must contain all edges in the Markov Blanket of the latent variable.
          Note: all variable names must be non empty strings
      data: dataframe, must contain all variables in the Markov Blanket of the latent variable. Include one column
          with the latent variable name, filled with np.nan for missing info about LV.
          If some data is present about the LV, create complete columns.
      lv_name: name of latent variable
      node_states: dictionary mapping variable name and list of states
      initial_params: way to initialise parameters. Can be:
          - "random": random values (default)
          - if a dictionary of dataframes is provided, this will be used as the initialisation
      seed: seed for the random generator (used if iitialise parameters randomly)
      box_constraints: minimum and maximum values for each model parameter. Specified with a dictionary mapping:
          - Node
          - two dataframes, in order: Min(P(Node|Par(Node))) and Max(P(Node|Par(Node)))
      priors: priors, provided as a mapping Node -> dataframe with Dirichilet priors for P(Node|Par(Node))
      non_missing_data_factor:
          This is a weight added to the non-missing data samples. The effect is as if the amount of data provided
          was bigger. Empirically, it helps to set the factor to 10 if the non missing data is ~1% of the dataset
      n_jobs:
          If -1 all CPUs are used. If 1 is given, no parallel computing code is used at all, which is useful
          for debugging. For n_jobs below -1, (n_cpus + 1 + n_jobs) are used.
          Thus for n_jobs = -2, all CPUs but one are used.
docstr_tok: ["Args",":","sm",":","structure",".","Only","requirement","is",":","must","contain","all","edges","in","the","Markov","Blanket","of","the","latent","variable",".","Note",":","all","variable","names","must","be","non","empty","strings","data",":","dataframe",",","must","contain","all","variables","in","the","Markov","Blanket","of","the","latent","variable",".","Include","one","column","with","the","latent","variable","name",",","filled","with","np.nan","for","missing","info","about","LV",".","If","some","data","is","present","about","the","LV",",","create","complete","columns",".","lv_name",":","name","of","latent","variable","node_states",":","dictionary","mapping","variable","name","and","list","of","states","initial_params",":","way","to","initialise","parameters",".","Can","be",":","-","``","random","''",":","random","values","(","default",")","-","if","a","dictionary","of","dataframes","is","provided",",","this","will","be","used","as","the","initialisation","seed",":","seed","for","the","random","generator","(","used","if","iitialise","parameters","randomly",")","box_constraints",":","minimum","and","maximum","values","for","each","model","parameter",".","Specified","with","a","dictionary","mapping",":","-","Node","-","two","dataframes",",","in","order",":","Min","(","P","(","Node|Par","(","Node",")",")",")","and","Max","(","P","(","Node|Par","(","Node",")",")",")","priors",":","priors",",","provided","as","a","mapping","Node","-",">","dataframe","with","Dirichilet","priors","for","P","(","Node|Par","(","Node",")",")","non_missing_data_factor",":","This","is","a","weight","added","to","the","non-missing","data","samples",".","The","effect","is","as","if","the","amount","of","data","provided","was","bigger",".","Empirically",",","it","helps","to","set","the","factor","to","10","if","the","non","missing","data","is","~1","%","of","the","dataset","n_jobs",":","If","-1","all","CPUs","are","used",".","If","1","is","given",",","no","parallel","computing","code","is","used","at","all",",","which","is","useful","for","debugging",".","For","n_jobs","below","-1",",","(","n_cpus","+","1","+","n_jobs",")","are","used",".","Thus","for","n_jobs","=","-2",",","all","CPUs","but","one","are","used","."]
code: def __init__(
      self,
      sm: StructureModel,
      data: pd.DataFrame,
      lv_name: str,
      node_states: Dict[str, list],
      initial_params: Union[str, Dict[str, pd.DataFrame]] = "random",
      seed: int = 22,
      box_constraints: Dict[
          str, Tuple[pd.DataFrame, pd.DataFrame]
      ] = None,
      priors: Dict[str, pd.DataFrame] = None,
      non_missing_data_factor: int = 1,
      n_jobs: int = 1,
  ):
      """
      Args:
          sm: structure. Only requirement is: must contain all edges in the Markov Blanket of the latent variable.
              Note: all variable names must be non empty strings
          data: dataframe, must contain all variables in the Markov Blanket of the latent variable. Include one column
              with the latent variable name, filled with np.nan for missing info about LV.
              If some data is present about the LV, create complete columns.
          lv_name: name of latent variable
          node_states: dictionary mapping variable name and list of states
          initial_params: way to initialise parameters. Can be:
              - "random": random values (default)
              - if a dictionary of dataframes is provided, this will be used as the initialisation
          seed: seed for the random generator (used if iitialise parameters randomly)
          box_constraints: minimum and maximum values for each model parameter. Specified with a dictionary mapping:
              - Node
              - two dataframes, in order: Min(P(Node|Par(Node))) and Max(P(Node|Par(Node)))
          priors: priors, provided as a mapping Node -> dataframe with Dirichilet priors for P(Node|Par(Node))
          non_missing_data_factor:
              This is a weight added to the non-missing data samples. The effect is as if the amount of data provided
              was bigger. Empirically, it helps to set the factor to 10 if the non missing data is ~1% of the dataset
          n_jobs:
              If -1 all CPUs are used. If 1 is given, no parallel computing code is used at all, which is useful
              for debugging. For n_jobs below -1, (n_cpus + 1 + n_jobs) are used.
              Thus for n_jobs = -2, all CPUs but one are used.
      """
      np.random.seed(seed)
  
      self.sm = sm
      self.lv_name = lv_name
      self.node_states = node_states
      self.initial_params = initial_params
      self.seed = seed
      self.non_missing_data_factor = non_missing_data_factor
      self.n_jobs = max(
          1, n_jobs if n_jobs >= 0 else (os.cpu_count() + 1 + n_jobs)
      )
  
      self.box_constraints = self._check_box_constraints(
          box_constraints
      )
      self.priors = self._check_priors(priors)
  
      # These are the nodes for which we compute CPDs. We do not care about CPDs of parents of the latent variable or
      # parents of children of the LV. This is because these CPDs do not depend on the LV, and are not affected by
      # the fact that we do not observe it
      # **IMPORTANT**: The first name in the list MUST BE lv_name (for the multiplication to work correctly)
      self.valid_nodes = [self.lv_name] + list(
          self.sm.successors(self.lv_name)
      )
  
      # Initialise CPDs
      self.cpds = self._initialise_network_cpds()
      self._old_cpds = None
      self._mb_partitions = None
      self._sufficient_stats = {}
      self._index_columns_lookup = {}
  
      # Compute aggregated data based on Markov blanket
      (
          self._mb_data,
          self._mb_partitions,
      ) = self._get_markov_blanket_data(data)
  
      # Build index columns lookup for each valid node
      self._lv_states = self.node_states[self.lv_name]
      self._mb_product = cpd_multiplication(
          [self.cpds[node] for node in self.valid_nodes]
      )
      for node in self.valid_nodes:
          self._mb_partitions[node]["_lookup_"] = self._mb_partitions[
              node
          ].apply(
              lambda record: self._build_lookup(
                  node, record  # pylint: disable=cell-var-from-loop
              ),
              axis=1,
          )
  
imports: import logging
  import os
  from time import time
  from typing import Dict, List, Tuple, Union
  import numpy
  import pandas
  from joblib import Parallel, delayed
  from causalnex.structure import StructureModel
  from causalnex.utils.data_utils import chunk_data, count_unique_rows
  from causalnex.utils.pgmpy_utils import cpd_multiplication
label_desc: Use image node_id 1 to create a new EMSingleLatentVariable object  with example: obj = EMSingleLatentVariable(sm, data, lv_name, node_states, initial_params, seed, box_constraints, priors, non_missing_data_factor, n_jobs)