Function _load_or_prompt has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def _load_or_prompt(
        self,
        env_name: EnvVariableName,
        config_name: ConfigVariableName,
        persist: bool = False,

• Read up
  Read up

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

Open

    def write(self, path: Optional[str] = None, **kwargs: Any) -> Tuple[bool, str]:
        file_to_write = kwargs.get("file")
        path = file_to_write.name if file_to_write else path or self.get_default_path()
        if self._metrics and _MODEL_PERFORMANCE in self._metrics:
            from whylogs.migration.converters import v1_to_dataset_profile_message_v0

• Read up
  Read up

Function log_debug_event has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def log_debug_event(
        self,
        debug_event: Optional[Dict[str, Any]] = None,
        *,
        trace_id: str,

• Read up
  Read up

Function initialize_logger has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def initialize_logger() -> None:
    # Initialize session
    n_attempts = 3
    while n_attempts > 0:
        # Initialize logger

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        with tempfile.NamedTemporaryFile() as seg_file:
                            view.write(file=seg_file, use_v0=use_v0)
                            seg_file.flush()
                            seg_file.seek(0)
                            zip_file.write(seg_file.name, seg_file.name.split("/")[-1])

Function column_is_nullable_datatype has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def column_is_nullable_datatype(column_name: str, datatype: str) -> MetricConstraint:
    """Check if column contains only records of specific datatype.
    Datatypes can be: integral, fractional, boolean, string, object.

    Returns True if there is at least one record of type datatype and there is no records of remaining types.

• Read up
  Read up

Function relation has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def relation(op: Relation, value: Union[str, int, float]) -> Callable[[Any], bool]:  # type: ignore
    if op == Relation.match:  # type: ignore
        return lambda x: re.compile(value).match(x)  # type: ignore
    if op == Relation.fullmatch:  # type: ignore
        return lambda x: re.compile(value).fullmatch(x)  # type: ignore

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if isinstance(arr, pd.Series):
                            first = welford_online_variance_m2(existing=first, new_value=arr.iloc[0])
                        else:
                            first = welford_online_variance_m2(existing=first, new_value=arr[0])

Avoid deeply nested control flow statements.

Open

                        if validate_with_row_id:
                            action(self.name, cond_name, x, identity_list[index])  # type: ignore
                        else:
                            action(self.name, cond_name, x)  # type: ignore

Consider simplifying this complex logical expression.

Open

            if col_name == name or (col_name is None and isinstance(why_type, col_type)):  # type: ignore
                for spec in resolver_spec.metrics:
                    cfg = spec.config or self._default_config or config or MetricConfig()
                    if self._allowed_metric(name, why_type, cfg, spec.metric):
                        if spec.metric.get_namespace() in result:

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

Open

def _log_segment(
    partition: SegmentationPartition,
    schema: DatasetSchema,
    obj: Any = None,
    pandas: Optional[pd.DataFrame] = 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 calculate has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def calculate(
        self, target_column_view: ColumnProfileView, reference_column_view: ColumnProfileView, with_thresholds=False
    ) -> Optional[DriftAlgorithmScore]:
        """Calculates drift score for a given column.

• 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 merge has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def merge(self, other: "EmbeddingMetric") -> "EmbeddingMetric":
        if self.references.value.shape != other.references.value.shape:
            if other.references.value.shape == (1, 1):
                # TODO: handle merging with other.serialize_references==False better
                # The (1, 1) shape indicates the other metric was created without a reference matrix.

• Read up
  Read up

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 _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 columnar_update has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def columnar_update(self, data: PreprocessedColumn) -> OperationResult:
        vectors = data.list.tensors if data.list.tensors else []
        vectors = vectors + (data.pandas.tensors.tolist() if data.pandas.tensors else [])

        if not vectors:

• 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 _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