From f77ffbd3033d122a375c3be63624ea80507d0541 Mon Sep 17 00:00:00 2001 From: Benjamin Klimczak Date: Mon, 7 Nov 2022 12:04:06 +0000 Subject: MLIA-275 Remove support for CSV output Change-Id: I212c9cad5f2ac28b75b1d9e2fb8f1f8b444b8397 --- src/mlia/cli/options.py | 3 +- src/mlia/core/reporting.py | 126 ++------------------------------------------- src/mlia/core/typing.py | 2 +- 3 files changed, 7 insertions(+), 124 deletions(-) (limited to 'src/mlia') diff --git a/src/mlia/cli/options.py b/src/mlia/cli/options.py index b28fa8f..f6dcf75 100644 --- a/src/mlia/cli/options.py +++ b/src/mlia/cli/options.py @@ -11,6 +11,7 @@ from typing import Callable from mlia.cli.config import get_available_backends from mlia.cli.config import get_default_backends from mlia.cli.config import is_corstone_backend +from mlia.core.reporting import OUTPUT_FORMATS from mlia.utils.filesystem import get_supported_profile_names from mlia.utils.types import is_number @@ -77,7 +78,7 @@ def add_tflite_model_options(parser: argparse.ArgumentParser) -> None: def add_output_options(parser: argparse.ArgumentParser) -> None: """Add output specific options.""" - valid_extensions = ["csv", "json"] + valid_extensions = OUTPUT_FORMATS def check_extension(filename: str) -> str: """Check extension of the provided file.""" diff --git a/src/mlia/core/reporting.py b/src/mlia/core/reporting.py index 0c8fabc..b96a6b5 100644 --- a/src/mlia/core/reporting.py +++ b/src/mlia/core/reporting.py @@ -3,7 +3,6 @@ """Reporting module.""" from __future__ import annotations -import csv import json import logging from abc import ABC @@ -36,6 +35,8 @@ from mlia.utils.types import is_list_of logger = logging.getLogger(__name__) +OUTPUT_FORMATS = ("json",) + class Report(ABC): """Abstract class for the report.""" @@ -44,10 +45,6 @@ class Report(ABC): def to_json(self, **kwargs: Any) -> Any: """Convert to json serializible format.""" - @abstractmethod - def to_csv(self, **kwargs: Any) -> list[Any]: - """Convert to csv serializible format.""" - @abstractmethod def to_plain_text(self, **kwargs: Any) -> str: """Convert to human readable format.""" @@ -134,10 +131,6 @@ class Cell: val = self._get_value() return self._apply_style(val) - def to_csv(self) -> Any: - """Cell definition for csv.""" - return self.value - def to_json(self) -> Any: """Cell definition for json.""" return self.value @@ -168,10 +161,6 @@ class CountAwareCell(Cell): super().__init__(value, Format(str_fmt=format_value)) - def to_csv(self) -> Any: - """Cell definition for csv.""" - return {"value": self.value, "unit": self.unit} - def to_json(self) -> Any: """Cell definition for json.""" return {"value": self.value, "unit": self.unit} @@ -239,41 +228,6 @@ class NestedReport(Report): self.alias = alias self.items = items - def to_csv(self, **kwargs: Any) -> list[Any]: - """Convert to csv serializible format.""" - result = {} - - def collect_item_values( - item: ReportItem, - _parent: ReportItem | None, - _prev: ReportItem | None, - _level: int, - ) -> None: - """Collect item values into a dictionary..""" - if item.value is None: - return - - if not isinstance(item.value, Cell): - result[item.alias] = item.raw_value - return - - csv_value = item.value.to_csv() - if isinstance(csv_value, dict): - csv_value = { - f"{item.alias}_{key}": value for key, value in csv_value.items() - } - else: - csv_value = {item.alias: csv_value} - - result.update(csv_value) - - self._traverse(self.items, collect_item_values) - - # make list out of the result dictionary - # first element - keys of the dictionary as headers - # second element - list of the dictionary values - return list(zip(*result.items())) - def to_json(self, **kwargs: Any) -> Any: """Convert to json serializible format.""" per_parent: dict[ReportItem | None, dict] = defaultdict(dict) @@ -474,32 +428,6 @@ class Table(Report): return title + formatted_table + footer - def to_csv(self, **kwargs: Any) -> list[Any]: - """Convert table to csv format.""" - headers = [[c.header for c in self.columns if c.supports_format("csv")]] - - def item_data(item: Any) -> Any: - if isinstance(item, Cell): - return item.value - - if isinstance(item, Table): - return ";".join( - str(item_data(cell)) for row in item.rows for cell in row - ) - - return item - - rows = [ - [ - item_data(item) - for (item, col) in zip(row, self.columns) - if col.supports_format("csv") - ] - for row in self.rows - ] - - return headers + rows - class SingleRow(Table): """Table with a single row.""" @@ -541,46 +469,6 @@ class CompoundReport(Report): return result - def to_csv(self, **kwargs: Any) -> list[Any]: - """Convert to csv serializible format. - - CSV format does support only one table. In order to be able to export - multiply tables they should be merged before that. This method tries to - do next: - - - if all tables have the same length then just concatenate them - - if one table has many rows and other just one (two with headers), then - for each row in table with many rows duplicate values from other tables - """ - csv_data = [item.to_csv() for item in self.reports] - lengths = [len(csv_item_data) for csv_item_data in csv_data] - - same_length = len(set(lengths)) == 1 - if same_length: - # all lists are of the same length, merge them into one - return [[cell for item in row for cell in item] for row in zip(*csv_data)] - - main_obj_indexes = [i for i, item in enumerate(csv_data) if len(item) > 2] - one_main_obj = len(main_obj_indexes) == 1 - - reference_obj_indexes = [i for i, item in enumerate(csv_data) if len(item) == 2] - other_only_ref_objs = len(reference_obj_indexes) == len(csv_data) - 1 - - if one_main_obj and other_only_ref_objs: - main_obj = csv_data[main_obj_indexes[0]] - return [ - item - + [ - ref_item - for ref_table_index in reference_obj_indexes - for ref_item in csv_data[ref_table_index][0 if i == 0 else 1] - ] - for i, item in enumerate(main_obj) - ] - - # write tables one after another if there is no other options - return [row for item in csv_data for row in item] - def to_plain_text(self, **kwargs: Any) -> str: """Convert to human readable format.""" return "\n".join(item.to_plain_text(**kwargs) for item in self.reports) @@ -624,12 +512,6 @@ def text_reporter(report: Report, output: FileLike, **kwargs: Any) -> None: print(report.to_plain_text(**kwargs), file=output) -def csv_reporter(report: Report, output: FileLike, **kwargs: Any) -> None: - """Produce report in csv format.""" - csv_writer = csv.writer(output) - csv_writer.writerows(report.to_csv(**kwargs)) - - def produce_report( data: Any, formatter: Callable[[Any], Report], @@ -639,7 +521,7 @@ def produce_report( ) -> None: """Produce report based on provided data.""" # check if provided format value is supported - formats = {"json": json_reporter, "plain_text": text_reporter, "csv": csv_reporter} + formats = {"json": json_reporter, "plain_text": text_reporter} if fmt not in formats: raise Exception(f"Unknown format {fmt}") @@ -764,7 +646,7 @@ def resolve_output_format(output: PathOrFileLike | None) -> OutputFormat: if isinstance(output, (str, Path)): format_from_filename = Path(output).suffix.lstrip(".") - if format_from_filename in ["json", "csv"]: + if format_from_filename in OUTPUT_FORMATS: return cast(OutputFormat, format_from_filename) return "plain_text" diff --git a/src/mlia/core/typing.py b/src/mlia/core/typing.py index bda995c..8244ff5 100644 --- a/src/mlia/core/typing.py +++ b/src/mlia/core/typing.py @@ -9,4 +9,4 @@ from typing import Union FileLike = TextIO PathOrFileLike = Union[str, Path, FileLike] -OutputFormat = Literal["plain_text", "csv", "json"] +OutputFormat = Literal["plain_text", "json"] -- cgit v1.2.1