Function get_pil_image_statistics has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def get_pil_image_statistics(
    img: ImageType, channels: List[str] = _IMAGE_HSV_CHANNELS, image_stats: List[str] = _STATS_PROPERTIES
) -> Dict:
    """
    Compute statistics data for a PIL Image

• Read up
  Read up

Function _get_segments has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _get_segments(self, reference_results: SegmentedResultSet, target_results: SegmentedResultSet):
        if len(reference_results.partitions) > 1 and len(target_results.partitions) > 1:
            logger.warning("More than one partition found. Only the first partition will be used for the estimation.")
        if len(reference_results.partitions) != len(target_results.partitions):
            raise ValueError("The number of partitions in the reference and target results must be the same.")

• Read up
  Read up

Function generate_profile_summary has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def generate_profile_summary(
    target_view: DatasetProfileView, config: Optional[SummaryConfig]
) -> Optional[Dict[str, Any]]:
    if config is None:
        config = SummaryConfig()

• Read up
  Read up

Function _convert_to_int_if_bool has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def _convert_to_int_if_bool(data: pd.core.frame.DataFrame, *columns: str) -> pd.core.frame.DataFrame:
    for col in columns:
        if all(isinstance(x, bool) for x in data[col]):
            data[col] = data[col].apply(lambda x: 1 if x else 0)
    return data

• Read up
  Read up

Function condition_validator has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def condition_validator(
    col_names: Union[str, List[str]],
    condition_name: Optional[str] = None,
    actions: Union[
        List[Callable[[str, str, Any, Optional[Any]], None]], Callable[[str, str, Any, Optional[Any]], None]

• Read up
  Read up

Function get has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def get(name: str, writable: Optional[_Writable] = None, **kwargs) -> Union[Writer, WriterWrapper]:
        if name == "local":
            from whylogs.api.writer.local import LocalWriter

            writer = LocalWriter(**kwargs)  # type: ignore

• Read up
  Read up

Function resolve has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def resolve(
        self,
        *,
        pandas: Optional[pd.DataFrame] = None,
        row: Optional[Mapping[str, Any]] = None,

• Read up
  Read up

Function _calculate_descriptive_statistics has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def _calculate_descriptive_statistics(
    column_view: Union[ColumnProfileView, None]
) -> Union[None, DescriptiveStatistics]:
    if column_view is None:
        return None

• Read up
  Read up

Function _generate_segment_tags_metadata has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def _generate_segment_tags_metadata(
    segment: Segment, partition: SegmentationPartition
) -> Tuple[Dict[str, str], List[SegmentTag], Dict[str, str]]:
    segment_metadata: Optional[Dict[str, str]] = None
    segment_tags: Optional[List[SegmentTag]] = None

• Read up
  Read up

Function __init__ has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def __init__(self, resolvers: List[ResolverSpec], default_config: Optional[MetricConfig] = None) -> None:
        super().__init__(resolvers, default_config)
        for resolver in resolvers:
            for metric_spec in resolver.metrics:
                if issubclass(metric_spec.metric, MultiMetric) and not resolver.exclude:

• Read up
  Read up

Function write has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def write(
        self, file: _Writable, dest: Optional[str] = None, **kwargs: Any
    ) -> Tuple[bool, Union[str, List[Tuple[bool, str]]]]:
        self._whylabs_client = self._whylabs_client.option(**kwargs)  # type: ignore
        if isinstance(file, FeatureWeights):

• Read up
  Read up

Function get_cardinality_estimate has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def get_cardinality_estimate(column_profile: ColumnProfileView) -> CardinalityEstimate:
    cardinality = column_profile.get_metric("cardinality")
    counts = column_profile.get_metric("counts")
    est_value: Optional[float] = None
    est_ratio: Optional[float] = None

• Read up
  Read up

Function _process_flush_message has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _process_flush_message(self, message: FlushMessage) -> None:
        for dataset_timestamp, container in self._cache.items():
            self._logger.debug(f"Generating result set for dataset timestamp {dataset_timestamp}")

            result_set = container.to_result_set()

• Read up
  Read up

Function to_summary_dict has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def to_summary_dict(self, cfg: Optional[SummaryConfig] = None) -> Dict[str, Any]:
        cfg = cfg or SummaryConfig()
        summary = {}
        for sub_name, metrics in self.submetrics.items():
            for namespace, metric in metrics.items():

• Read up
  Read up

Function _do_match has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _do_match(cls, dtype_or_type: Any, maybe_type: Optional[Any]) -> bool:
        # Pandas Categorical is Strings
        if CategoricalDtype is not None and isinstance(dtype_or_type, CategoricalDtype):
            return True

• Read up
  Read up

Function merge_submetrics has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def merge_submetrics(self: MULTI_METRIC, other: MULTI_METRIC) -> Dict[str, Dict[str, Metric]]:
        if self.namespace != other.namespace:
            raise ValueError(f"Attempt to merge MultiMetrics {self.namespace} and {other.namespace}")

        submetric_names = set(self.submetrics.keys())

• Read up
  Read up

Function _write has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _write(
        self, path: Optional[str] = None, filename: Optional[str] = None, **kwargs: Any
    ) -> Tuple[bool, Union[str, List[str]]]:
        file_to_write = kwargs.get("file")
        if file_to_write is None:

• Read up
  Read up

Function to_protobuf has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def to_protobuf(
        self,
    ) -> ModelProfileMessage:
        model_type = ModelType.UNKNOWN
        if not self.model_type:

• Read up
  Read up

Avoid too many return statements within this method.

Open

      return (T) result;

Avoid too many return statements within this function.

Open

        return self._get_writer(WhyLabsBatchWriter).write(file, dest, **kwargs)