Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor: Mode.from_name, tests for __main__ #254

Merged
merged 8 commits into from
Apr 14, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions src/wakepy/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

from wakepy import ModeExit
from wakepy.core.constants import ModeName
from wakepy.core.mode import create_mode
from wakepy.core.mode import Mode

if typing.TYPE_CHECKING:
from typing import List
Expand All @@ -33,7 +33,7 @@
\ \ /\ / // _` || |/ // _ \| '_ \ | | | |
\ V V /| (_| || <| __/| |_) || |_| |
\_/\_/ \__,_||_|\_\\___|| .__/ \__, |
{VERSION_STRING} | | __/ |
{VERSION_STRING}| | __/ |
|_| |___/ """

WAKEPY_TICKBOXES_TEMPLATE = """
Expand All @@ -44,7 +44,7 @@

def main() -> None:
modename = parse_arguments(sys.argv[1:])
mode = create_mode(modename=modename, on_fail=handle_activation_error)
mode = Mode.from_name(modename, on_fail=handle_activation_error)
print(get_startup_text(mode=modename))
with mode:
if not mode.active:
Expand All @@ -57,6 +57,10 @@ def main() -> None:


def handle_activation_error(result: ActivationResult) -> None:
print(_get_activation_error_text(result))


def _get_activation_error_text(result: ActivationResult) -> str:
from wakepy import __version__

error_text = f"""
Expand All @@ -74,8 +78,12 @@ def handle_activation_error(result: ActivationResult) -> None:

Thank you!
""" # noqa 501

out = []
for block in dedent(error_text.strip("\n")).split("\n"):
print(fill(block, 80))
out.append(fill(block, 80))

return "\n".join(out)


def parse_arguments(
Expand Down Expand Up @@ -143,7 +151,7 @@ def get_startup_text(mode: ModeName) -> str:
from wakepy import __version__

wakepy_text = WAKEPY_TEXT_TEMPLATE.format(
VERSION_STRING=f"{' v.'+__version__: <20}"
VERSION_STRING=f"{' v.'+__version__: <28}"
)
options_txt = WAKEPY_TICKBOXES_TEMPLATE.strip("\n").format(
no_auto_suspend="x",
Expand Down
117 changes: 60 additions & 57 deletions src/wakepy/core/mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -267,65 +267,68 @@ def __exit__(

return False

@classmethod
def from_name(
cls,
modename: ModeName,
methods: Optional[StrCollection] = None,
omit: Optional[StrCollection] = None,
methods_priority: Optional[MethodsPriorityOrder] = None,
on_fail: OnFail = "error",
dbus_adapter: Type[DBusAdapter] | DBusAdapterTypeSeq | None = None,
) -> Mode:
"""
Creates and returns a Mode based on a mode name.

Parameters
----------
modename:
The name of the mode to create. Used for debugging, logging,
warning and error messages. Could be basically any string.
methods: list, tuple or set of str
The names of Methods to select from the mode defined with
`modename`; a "whitelist" filter. Means "use these and only these
Methods". Any Methods in `methods` but not in the selected mode
will raise a ValueError. Cannot be used same time with `omit`.
Optional.
omit: list, tuple or set of str or None
The names of Methods to remove from the mode defined with
`modename`; a "blacklist" filter. Any Method in `omit` but not in
the selected mode will be silently ignored. Cannot be used same
time with `methods`. Optional.
on_fail: "error" | "warn" | "pass" | Callable
Determines what to do in case mode activation fails. Valid options
are: "error", "warn", "pass" and a callable. If the option is
"error", raises wakepy.ActivationError. Is selected "warn", issues
warning. If "pass", does nothing. If `on_fail` is a callable, it
must take one positional argument: result, which is an instance of
ActivationResult. The ActivationResult contains more detailed
information about the activation process.
methods_priority: list[str | set[str]]
The priority order, which is a list of method names or asterisk
('*'). The asterisk means "all rest methods" and may occur only
once in the priority order, and cannot be part of a set. All method
names must be unique and must be part of the `methods`.
dbus_adapter:
For using a custom dbus-adapter. Optional. If not given, the
default dbus adapter is used, which is :class:`~wakepy.\\
dbus_adapters.jeepney.JeepneyDBusAdapter`

def create_mode(
modename: ModeName,
methods: Optional[StrCollection] = None,
omit: Optional[StrCollection] = None,
methods_priority: Optional[MethodsPriorityOrder] = None,
on_fail: OnFail = "error",
dbus_adapter: Type[DBusAdapter] | DBusAdapterTypeSeq | None = None,
) -> Mode:
"""
Creates and returns a Mode (a context manager).

Parameters
----------
modename:
The name of the mode to create. Used for debugging, logging, warning
and error messaged. Could be basically any string.
methods: list, tuple or set of str
The names of Methods to select from the mode defined with `modename`;
a "whitelist" filter. Means "use these and only these Methods". Any
Methods in `methods` but not in the selected mode will raise a
ValueError. Cannot be used same time with `omit`. Optional.
omit: list, tuple or set of str or None
The names of Methods to remove from the mode defined with `modename`;
a "blacklist" filter. Any Method in `omit` but not in the selected mode
will be silently ignored. Cannot be used same time with `methods`.
Optional.
on_fail: "error" | "warn" | "pass" | Callable
Determines what to do in case mode activation fails. Valid options
are: "error", "warn", "pass" and a callable. If the option is
"error", raises wakepy.ActivationError. Is selected "warn", issues
warning. If "pass", does nothing. If `on_fail` is a callable, it
must take one positional argument: result, which is an instance of
ActivationResult. The ActivationResult contains more detailed
information about the activation process.
methods_priority: list[str | set[str]]
The priority order, which is a list of method names or asterisk
('*'). The asterisk means "all rest methods" and may occur only
once in the priority order, and cannot be part of a set. All method
names must be unique and must be part of the `methods`.
dbus_adapter:
For using a custom dbus-adapter. Optional. If not given, the
default dbus adapter is used, which is :class:`~wakepy.dbus_adapters.\\
jeepney.JeepneyDBusAdapter`
Returns
-------
mode: Mode
The context manager for the selected mode.

Returns
-------
mode: Mode
The context manager for the selected mode.
"""
methods_for_mode = get_methods_for_mode(modename)
selected_methods = select_methods(methods_for_mode, use_only=methods, omit=omit)
return Mode(
name=modename,
methods=selected_methods,
methods_priority=methods_priority,
on_fail=on_fail,
dbus_adapter=dbus_adapter,
)
"""
methods_for_mode = get_methods_for_mode(modename)
selected_methods = select_methods(methods_for_mode, use_only=methods, omit=omit)
return cls(
name=modename,
methods=selected_methods,
methods_priority=methods_priority,
on_fail=on_fail,
dbus_adapter=dbus_adapter,
)


def handle_activation_fail(on_fail: OnFail, result: ActivationResult) -> None:
Expand Down
11 changes: 9 additions & 2 deletions src/wakepy/core/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from __future__ import annotations

import logging
import typing
from typing import overload

Expand Down Expand Up @@ -47,6 +48,7 @@
"""A name -> Method class mapping. Updated automatically; when python loads
a module with a subclass of Method, the Method class is added to this registry.
"""
logger = logging.getLogger(__name__)


class MethodRegistryError(RuntimeError):
Expand All @@ -58,8 +60,13 @@ def register_method(method_class: Type[Method]) -> None:

if method_class.is_unnamed():
# Methods without a name will not be registered
logging.debug(
"Not registering Method %s as it does not have a name set.", method_class
)
return

logging.debug("Registering Method %s (name: %s)", method_class, method_class.name)

method_dict: MethodDict = _method_registry.get(method_class.mode, dict())

if method_class.name in method_dict:
Expand Down Expand Up @@ -188,8 +195,8 @@ def get_methods(
def get_methods_for_mode(
mode: ModeName | str,
) -> List[MethodCls]:
"""Get the Method classes belonging to a Mode; Methods with
Method.mode = `mode`.
"""Get the Method classes belonging to a Mode; Methods with Method.mode =
`mode`.

Parameters
----------
Expand Down
10 changes: 5 additions & 5 deletions src/wakepy/modes/keep.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import typing

from ..core.constants import ModeName
from ..core.mode import create_mode
from ..core.mode import Mode

if typing.TYPE_CHECKING:
from typing import Optional, Type
Expand Down Expand Up @@ -77,8 +77,8 @@ def running(
>>> with keep.running() as k:
>>> # do something that takes a long time.
"""
return create_mode(
modename=ModeName.KEEP_RUNNING,
return Mode.from_name(
ModeName.KEEP_RUNNING,
omit=omit,
methods=methods,
methods_priority=methods_priority,
Expand Down Expand Up @@ -151,8 +151,8 @@ def presenting(
>>> # do something that takes a long time.

"""
return create_mode(
modename=ModeName.KEEP_PRESENTING,
return Mode.from_name(
ModeName.KEEP_PRESENTING,
methods=methods,
omit=omit,
methods_priority=methods_priority,
Expand Down
Loading