class_docstr: Contains helper methods for managing EventBridge rules for a feature processor.
class_docstr_tok: ["Contains","helper","methods","for","managing","EventBridge","rules","for","a","feature","processor","."]
docstr: Generates the event pattern json string from the pipeline events.
  
  Args:
      pipeline_events: List of pipeline events.
  Returns:
      str: The event pattern json string.
  
  Raises:
      ValueError: If pipeline events contain duplicate pipeline names.
docstr_tok: ["Generates","the","event","pattern","json","string","from","the","pipeline","events",".","Args",":","pipeline_events",":","List","of","pipeline","events",".","Returns",":","str",":","The","event","pattern","json","string",".","Raises",":","ValueError",":","If","pipeline","events","contain","duplicate","pipeline","names","."]
code: def _generate_event_pattern_from_feature_processor_pipeline_events(
      self, pipeline_events: List[FeatureProcessorPipelineEvents]
  ) -> str:
      """Generates the event pattern json string from the pipeline events.
  
      Args:
          pipeline_events: List of pipeline events.
      Returns:
          str: The event pattern json string.
  
      Raises:
          ValueError: If pipeline events contain duplicate pipeline names.
      """
  
      result_event_pattern = {
          "detail-type": [
              "SageMaker Model Building Pipeline Execution Status Change"
          ],
      }
      filters = []
      desired_status_to_pipeline_names_map = (
          self._aggregate_pipeline_events_with_same_desired_status(
              pipeline_events
          )
      )
      for desired_status in desired_status_to_pipeline_names_map:
          pipeline_arns = [
              self._generate_pipeline_arn_and_name(pipeline_name)[
                  "pipeline_arn"
              ]
              for pipeline_name in desired_status_to_pipeline_names_map[
                  desired_status
              ]
          ]
          curr_filter = BASE_EVENT_PATTERN.copy()
          curr_filter["detail"]["pipelineArn"] = pipeline_arns
          curr_filter["detail"]["currentPipelineExecutionStatus"] = [
              status_enum.value for status_enum in desired_status
          ]
          filters.append(curr_filter)
      if len(filters) > 1:
          result_event_pattern["$or"] = filters
      else:
          result_event_pattern.update(filters[0])
      return json.dumps(result_event_pattern)
  
imports: from __future__ import absolute_import
  import json
  import logging
  import re
  from typing import Dict, List, Tuple, Optional, Any
  import attr
  from botocore.exceptions import ClientError
  from botocore.paginate import PageIterator
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._feature_processor_pipeline_events import FeatureProcessorPipelineEvents
  from sagemaker.feature_store.feature_processor._constants import RESOURCE_NOT_FOUND_EXCEPTION, PIPELINE_ARN_REGEX_PATTERN, BASE_EVENT_PATTERN
  from sagemaker.feature_store.feature_processor._enums import FeatureProcessorPipelineExecutionStatus
  from sagemaker.utils import TagsDict
label_desc: Use image node_id 10 for calling the EventBridgeRuleHelper obj's underlying member method code with example usage: obj._generate_event_pattern_from_feature_processor_pipeline_events(pipeline_events) and returns: json