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: Keeps only features the belong to the latent variable's Markov blanket +
  groups and counts identical rows +
  multiplies non missing data counts by a factor
  
  Args:
      df: Raw data
  
  Returns:
      Aggregated data, as well as data partition for each node
docstr_tok: ["Keeps","only","features","the","belong","to","the","latent","variable","'s","Markov","blanket","+","groups","and","counts","identical","rows","+","multiplies","non","missing","data","counts","by","a","factor","Args",":","df",":","Raw","data","Returns",":","Aggregated","data",",","as","well","as","data","partition","for","each","node"]
code: def _get_markov_blanket_data(
      self,
      df: pd.DataFrame,
  ) -> Tuple[pd.DataFrame, Dict[str, pd.DataFrame]]:
      """
      Keeps only features the belong to the latent variable's Markov blanket +
      groups and counts identical rows +
      multiplies non missing data counts by a factor
  
      Args:
          df: Raw data
  
      Returns:
          Aggregated data, as well as data partition for each node
      """
      # Add a column for latent variable if not already present
      if self.lv_name not in df.columns:
          df[self.lv_name] = np.nan
  
      # Get the counts of each unique record
      valid_cols = set()
  
      for node in self.valid_nodes:
          valid_cols.add(node)
          valid_cols.update(self.sm.predecessors(node))
  
      mb_data = count_unique_rows(df[list(valid_cols)])
      indices = ~mb_data.isna().any(axis=1)
      mb_data.loc[indices, "count"] *= self.non_missing_data_factor
  
      # Partition data based on the Markov blanket of each node
      mb_product = cpd_multiplication(
          [self.cpds[node] for node in self.valid_nodes]
      )
      mb_partitions = {}
  
      for node in self.valid_nodes:
          valid_cols = list(
              set(
                  [node, "count"]
                  + list(self.sm.predecessors(node))
                  + mb_product.columns.names
              )
          )
          mb_partitions[node] = count_unique_rows(mb_data[valid_cols])
  
      return mb_data, mb_partitions
  
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 15 for calling the EMSingleLatentVariable obj's underlying member method code with example usage: obj._get_markov_blanket_data(df) and returns: mb_data,  mb_partitions