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: Performs the Expectation step.
  This boils down to computing the expected sufficient statistics M[X, U]
  for every "valid" node X, where U = Par(X)
  
  Returns:
      The expected sufficient statistics of each node X
docstr_tok: ["Performs","the","Expectation","step",".","This","boils","down","to","computing","the","expected","sufficient","statistics","M","[","X",",","U","]","for","every","``","valid","''","node","X",",","where","U","=","Par","(","X",")","Returns",":","The","expected","sufficient","statistics","of","each","node","X"]
code: def e_step(self) -> Dict[str, pd.DataFrame]:
      """
      Performs the Expectation step.
      This boils down to computing the expected sufficient statistics M[X, U]
      for every "valid" node X, where U = Par(X)
  
      Returns:
          The expected sufficient statistics of each node X
      """
      # This is a product of elements in the Markov Blanket of the latent variable.
      # NOTE: Convert product dataframe to dictionary to speed up the E-step
      self._mb_product = cpd_multiplication(
          [self.cpds[node] for node in self.valid_nodes],
          normalize=True,
      ).to_dict(orient="dict")
  
      # Get M[X, U] for X being each valid node and U being its parents (Daphne Koller's notation)
      for node in self.valid_nodes:
          node_mb_data = self._mb_partitions[node]
  
          # Initialize ESS with zeros (or prior values) and then increase from data
          sufficient_stats_df = self._initialize_sufficient_stats(node)
          sufficient_stats = sufficient_stats_df.to_dict(orient="dict")
  
          # Update ESS based on all data records (observations)
          if self.n_jobs == 1:
              results = self._update_sufficient_stats(
                  node_mb_data["_lookup_"]
              )
  
              for updates in results:
                  for idx, cols, val in updates:
                      sufficient_stats[cols][idx] += val
          else:
              results = Parallel(n_jobs=self.n_jobs, prefer="threads")(
                  delayed(self._update_sufficient_stats)(
                      chunk_df["_lookup_"]
                  )
                  for chunk_df in chunk_data(
                      node_mb_data, self.n_jobs * 2
                  )
              )
              for chunk_results in results:
                  for updates in chunk_results:
                      for idx, cols, val in updates:
                          sufficient_stats[cols][idx] += val
  
          # Register sufficient statistics as Pandas dataframe
          self._sufficient_stats[node] = pd.DataFrame(
              sufficient_stats,
              index=sufficient_stats_df.index,
              columns=sufficient_stats_df.columns,
          )
  
      return self._sufficient_stats
  
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 4 for calling the EMSingleLatentVariable obj's underlying member method code with example usage: obj.e_step() and returns: self