aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorBenjamin Klimczak <benjamin.klimczak@arm.com>2022-11-07 12:04:06 +0000
committerBenjamin Klimczak <benjamin.klimczak@arm.com>2022-11-09 11:23:32 +0000
commitf77ffbd3033d122a375c3be63624ea80507d0541 (patch)
tree0465aafdba081f15a769c5929ed5d3b5610bbed1 /src
parent006c95f3053c89af41726c582c7b4d0f9914553b (diff)
downloadmlia-f77ffbd3033d122a375c3be63624ea80507d0541.tar.gz
MLIA-275 Remove support for CSV output
Change-Id: I212c9cad5f2ac28b75b1d9e2fb8f1f8b444b8397
Diffstat (limited to 'src')
-rw-r--r--src/mlia/cli/options.py3
-rw-r--r--src/mlia/core/reporting.py126
-rw-r--r--src/mlia/core/typing.py2
3 files changed, 7 insertions, 124 deletions
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."""
@@ -45,10 +46,6 @@ class Report(ABC):
"""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"]