Unverified Commit b3b976b8 authored by Philipp Sommer's avatar Philipp Sommer
Browse files

make report settings public

and add NONE ShowReportMethod
parent bc283903
......@@ -36,9 +36,13 @@ class BaseReport(BaseModel):
_pulsar: Optional[PulsarMessageConsumer] = None
_request: Optional[Dict] = None
_response_properties: Optional[Dict] = None
_settings: ReportSettings = ReportSettings()
_window: Optional[curses.window] = None # type: ignore
# settings for this report
settings: ReportSettings = Field(
default_factory=ReportSettings, description="Settings for the report."
# selector for the report type. should be changed by subclasses to make
# sure we select the correct model when deserializing
report_type: Literal["basic"] = Field(
......@@ -72,17 +76,13 @@ class BaseReport(BaseModel):
if submit:
def window(self) -> curses.window: # type: ignore
"""A curses window for displaying the report.
See Also
if self._window is None:
self._window = curses.initscr()
return self._window
def _combine_subclasses(cls, combined_type: Any) -> Any:
"""Recurse subclasses and combine them with a type."""
for cls_ in cls.__subclasses__():
combined_type = Union[combined_type, cls_]
combined_type = cls_._combine_subclasses(combined_type)
return combined_type
def from_payload(cls, payload: str) -> BaseReport:
......@@ -91,20 +91,80 @@ class BaseReport(BaseModel):
This method takes into account all subclasses of the
:class:`BaseReport` to find the correct one.
t: Any = cls
for cls_ in cls.__subclasses__():
t = Union[t, cls_]
combined_type: Any = cls
combined_type = cls._combine_subclasses(combined_type)
print("*" * 100)
print("+" * 100)
Model: Type[BaseModel] = create_model(
"Report", __root__=(t, Field(description="Report type"))
__root__=(combined_type, Field(description="Report type")),
return Model.parse_raw(payload).__root__
def get_dummy_arguments(cls) -> Dict:
"""Get dummy argument to instantiate a report.
This class method is supposed to generate arguments that can be used
to create a report that shows nothing."""
return {"settings": {"show_method": ShowReportMethod.none}}
def from_arg(cls, arg: Any) -> BaseReport:
"""Convenience method to generate a report from a generic argument.
This method can be used to generate a report from a generic argument.
It is supposed to be used as a quick function to generate a report that
does nothing eventually.
arg: Any
The argument that shall be interpreted as an input to generate a
- If `arg` is an instance of this report class, it is returned
- If `arg` is an instance of :class:`BaseReport`, this method
returns a new report based using the `BaseReport` as an input
- If `arg` is a dictionary, this is used to instantiate a new
Report of this class
- for any other argument, we return a silent report, i.e. one with
:attr:`~deprogressapi.settings.ReportSettings.show_method` set
to :attr:`deprogressapi.settings.none`.
This function takes an optional report argument and makes sure it has
a report with the ``from_arg`` convenience function::
from typing import Optional
def reporting_function(report: Optional[BaseReport] = None):
# now report may be None, using ``from_arg``, we make sure
# that report is a real report class.
report = BaseReport.from_arg(report)
if isinstance(arg, cls):
ret = arg
elif isinstance(arg, BaseReport) or isinstance(arg, dict):
ret = cls.parse_obj(arg)
ret._pulsar = getattr(arg, "pulsar", None)
ret._request = getattr(arg, "request_msg", None)
dummy_args = cls.get_dummy_arguments()
ret = cls.parse_obj(dummy_args)
return ret
def submit(self) -> None:
"""Submit the report.
This method submits the report and submits it via the pulsar or
calls the :meth:`show` method."""
if self._pulsar is not None:
if self.settings.show_method is None:
elif self._pulsar is not None:
from demessaging.PulsarMessageConstants import (
......@@ -129,17 +189,19 @@ class BaseReport(BaseModel):
The default implementation just prints the json string of this report.
method = self._settings.get_show_method()
method = self.settings.get_show_method()
if method == ShowReportMethod.print_:
elif method == ShowReportMethod.curses:
elif method == ShowReportMethod.none:
def show_print(self):
def show_print(self) -> None:
"""Show the report using pythons built-in :func:`print` function."""
def show_curses(self):
def show_curses(self) -> None:
"""Show the report using pythons built-in :func:`curses` module."""
report = self.report_to_string()
height, width = self.window.getmaxyx()
......@@ -150,6 +212,22 @@ class BaseReport(BaseModel):
def show_none(self) -> None:
"""Dummy method that is called when the report_method is None"""
def window(self) -> curses.window: # type: ignore
"""A curses window for displaying the report.
See Also
if self._window is None:
self._window = curses.initscr()
return self._window
def report_to_string(self) -> str:
"""Render the report as a string."""
return self.json(indent=2)
......@@ -183,6 +261,7 @@ class BaseReport(BaseModel):
ret = dict(super().__repr_args__())
ret.pop("status", None)
ret.pop("report_type", None)
ret.pop("settings", None)
return list(ret.items())
......@@ -236,6 +315,15 @@ class ProgressReport(BaseReport):
return child
def get_dummy_arguments(cls) -> Dict:
# reimplemented to add a step_message
ret = super().get_dummy_arguments()
ret["step_message"] = ""
return ret
get_dummy_arguments.__doc__ = BaseReport.__doc__
......@@ -11,9 +11,19 @@ def is_running_in_notebook():
class ShowReportMethod(str, Enum):
"""Methods for reporting."""
# do not show any report
none = "NONE"
# automatically determine what to show
auto = "AUTO"
# use pythons built-in print function
print_ = "PRINT"
# use python built-in curses module
curses = "WINDOWED"
# use ipywidgets (not yet implemented)
jupyter = "JUPYTER"
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment