7843823363

Committed 09 Feb 2024 12:18PM UTC coverage: 66.465% (-0.3%) from 66.762%

Build # 7843823363

Build Type

Pull #975

github

Committed by

web-flow

Commit Message

Merge d3438222e into 50d0abf01

Pull Request Pull Request #975: Feature/delete beliefs

Run Details

3872 of 6527 branches covered (0.0%)

Branch coverage included in aggregate %.

15 of 59 new or added lines in 3 files covered. (25.42%)

4 existing lines in 2 files now uncovered.

6914 of 9701 relevant lines covered (71.27%)

2.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.14

/flexmeasures/cli/utils.py

"""
Utils for FlexMeasures CLI
"""

from __future__ import annotations

from typing import Any
from datetime import datetime, timedelta

import click
import pytz
from click_default_group import DefaultGroup

from flexmeasures.utils.time_utils import get_most_recent_hour, get_timezone


class MsgStyle(object):
    """Stores the text styles for the different events

    Styles options are the attributes of the `click.style` which can be found
    [here](https://click.palletsprojects.com/en/8.1.x/api/#click.style).

    """

    SUCCESS: dict[str, Any] = {"fg": "green"}
    WARN: dict[str, Any] = {"fg": "yellow"}
    ERROR: dict[str, Any] = {"fg": "red"}


class DeprecatedOption(click.Option):
    """A custom option that can be used to mark an option as deprecated.

    References
    ----------------

    Copied from  https://stackoverflow.com/a/50402799/13775459
    """

    def __init__(self, *args, **kwargs):
        self.deprecated = kwargs.pop("deprecated", ())
        self.preferred = kwargs.pop("preferred", args[0][-1])
        super(DeprecatedOption, self).__init__(*args, **kwargs)


class DeprecatedOptionsCommand(click.Command):
    """A custom command that can be used to mark options as deprecated.

    References
    ----------------

    Adapted from  https://stackoverflow.com/a/50402799/13775459
    """

    def make_parser(self, ctx):
        """Hook 'make_parser' and during processing check the name
        used to invoke the option to see if it is preferred"""

        parser = super(DeprecatedOptionsCommand, self).make_parser(ctx)

        # get the parser options
        options = set(parser._short_opt.values())
        options |= set(parser._long_opt.values())

        for option in options:
            if not isinstance(option.obj, DeprecatedOption):
                continue

            def make_process(an_option):
                """Construct a closure to the parser option processor"""

                orig_process = an_option.process
                deprecated = getattr(an_option.obj, "deprecated", None)
                preferred = getattr(an_option.obj, "preferred", None)
                msg = "Expected `deprecated` value for `{}`"
                assert deprecated is not None, msg.format(an_option.obj.name)

                def process(value, state):
                    """The function above us on the stack used 'opt' to
                    pick option from a dict, see if it is deprecated"""

                    # reach up the stack and get 'opt'
                    import inspect

                    frame = inspect.currentframe()
                    try:
                        opt = frame.f_back.f_locals.get("opt")
                    finally:
                        del frame

                    if opt in deprecated:
                        click.secho(
                            f"Option '{opt}' will be replaced by '{preferred}'.",
                            **MsgStyle.WARN,
                        )
                    return orig_process(value, state)

                return process

            option.process = make_process(option)

        return parser


class DeprecatedDefaultGroup(DefaultGroup):
    """Invokes a default subcommand, *and* shows a deprecation message.

    Also adds the `invoked_default` boolean attribute to the context.
    A group callback can use this information to figure out if it's being executed directly
    (invoking the default subcommand) or because the execution flow passes onwards to a subcommand.
    By default it's None, but it can be the name of the default subcommand to execute.

    .. sourcecode:: python

        import click
        from flexmeasures.cli.utils import DeprecatedDefaultGroup

        @click.group(cls=DeprecatedDefaultGroup, default="bar", deprecation_message="renamed to `foo bar`.")
        def foo(ctx):
            if ctx.invoked_default:
                click.echo("foo")

        @foo.command()
        def bar():
            click.echo("bar")

    .. sourcecode:: console

        $ flexmeasures foo
        DeprecationWarning: renamed to `foo bar`.
        foo
        bar
        $ flexmeasures foo bar
        bar
    """

    def __init__(self, *args, **kwargs):
        self.deprecation_message = "DeprecationWarning: " + kwargs.pop(
            "deprecation_message", ""
        )
        super().__init__(*args, **kwargs)

    def get_command(self, ctx, cmd_name):
        ctx.invoked_default = None
        if cmd_name not in self.commands:
            click.echo(click.style(self.deprecation_message, fg="red"), err=True)
            ctx.invoked_default = self.default_cmd_name
        return super().get_command(ctx, cmd_name)


def get_timerange_from_flag(
    last_hour: bool = False,
    last_day: bool = False,
    last_7_days: bool = False,
    last_month: bool = False,
    last_year: bool = False,
    timezone: pytz.BaseTzInfo = get_timezone(),
) -> tuple[datetime, datetime]:
    """This function returns a time range [start,end] of the last-X period.
    See input parameters for more details.

    :param bool last_hour: flag to get the time range of the last finished hour.
    :param bool last_day: flag to get the time range for yesterday.
    :param bool last_7_days: flag to get the time range of the previous 7 days.
    :param bool last_month: flag to get the time range of last calendar month
    :param bool last_year: flag to get the last completed calendar year
    :param timezone: timezone object to represent
    :returns: start:datetime, end:datetime
    """

    current_hour = get_most_recent_hour().astimezone(timezone)

    if last_hour:  # last finished hour
        end = current_hour
        start = current_hour - timedelta(hours=1)

    if last_day:  # yesterday
        end = current_hour.replace(hour=0)
        start = end - timedelta(days=1)

    if last_7_days:  # last finished 7 day period.
        end = current_hour.replace(hour=0)
        start = end - timedelta(days=7)

    if last_month:
        end = current_hour.replace(
            hour=0, day=1
        )  # get the first day of the current month
        start = (end - timedelta(days=1)).replace(
            day=1
        )  # get first day of the previous month

    if last_year:  # last calendar year
        end = current_hour.replace(
            month=1, day=1, hour=0
        )  # get first day of current year
        start = (end - timedelta(days=1)).replace(
            day=1, month=1
        )  # get first day of previous year

    return start, end


def validate_unique(ctx, param, value):
    """Callback function to ensure multiple values are unique."""
    if value is not None:
        # Check if all values are unique
        if len(value) != len(set(value)):
            raise click.BadParameter("Values must be unique.")
    return value


def abort(message: str):
    click.secho(message, **MsgStyle.ERROR)
    raise click.Abort()


def done(message: str):
    click.secho(message, **MsgStyle.SUCCESS)

1	"""	4✔
2	Utils for FlexMeasures CLI
3	"""
4
5	from __future__ import annotations	4✔
6
7	from typing import Any	4✔
8	from datetime import datetime, timedelta	4✔
9
10	import click	4✔
11	import pytz	4✔
12	from click_default_group import DefaultGroup	4✔
13
14	from flexmeasures.utils.time_utils import get_most_recent_hour, get_timezone	4✔
15
16
17	class MsgStyle(object):	4✔
18	"""Stores the text styles for the different events	4✔
19
20	Styles options are the attributes of the `click.style` which can be found
21	[here](https://click.palletsprojects.com/en/8.1.x/api/#click.style).
22
23	"""
24
25	SUCCESS: dict[str, Any] = {"fg": "green"}	4✔
26	WARN: dict[str, Any] = {"fg": "yellow"}	4✔
27	ERROR: dict[str, Any] = {"fg": "red"}	4✔
28
29
30	class DeprecatedOption(click.Option):	4✔
31	"""A custom option that can be used to mark an option as deprecated.	4✔
32
33	References
34	----------------
35
36	Copied from https://stackoverflow.com/a/50402799/13775459
37	"""
38
39	def __init__(self, args, *kwargs):	4✔
40	self.deprecated = kwargs.pop("deprecated", ())	4✔
41	self.preferred = kwargs.pop("preferred", args[0][-1])	4✔
42	super(DeprecatedOption, self).__init__(args, *kwargs)	4✔
43
44
45	class DeprecatedOptionsCommand(click.Command):	4✔
46	"""A custom command that can be used to mark options as deprecated.	4✔
47
48	References
49	----------------
50
51	Adapted from https://stackoverflow.com/a/50402799/13775459
52	"""
53
54	def make_parser(self, ctx):	4✔
55	"""Hook 'make_parser' and during processing check the name
56	used to invoke the option to see if it is preferred"""
57
58	parser = super(DeprecatedOptionsCommand, self).make_parser(ctx)	×
59
60	# get the parser options
61	options = set(parser._short_opt.values())	×
62	options \|= set(parser._long_opt.values())	×
63
64	for option in options:	×
65	if not isinstance(option.obj, DeprecatedOption):	×
66	continue	×
67
68	def make_process(an_option):	×
69	"""Construct a closure to the parser option processor"""
70
71	orig_process = an_option.process	×
72	deprecated = getattr(an_option.obj, "deprecated", None)	×
73	preferred = getattr(an_option.obj, "preferred", None)	×
74	msg = "Expected `deprecated` value for `{}`"	×
75	assert deprecated is not None, msg.format(an_option.obj.name)	×
76
77	def process(value, state):	×
78	"""The function above us on the stack used 'opt' to
79	pick option from a dict, see if it is deprecated"""
80
81	# reach up the stack and get 'opt'
82	import inspect	×
83
84	frame = inspect.currentframe()	×
85	try:	×
86	opt = frame.f_back.f_locals.get("opt")	×
87	finally:
88	del frame	×
89
90	if opt in deprecated:	×
91	click.secho(	×
92	f"Option '{opt}' will be replaced by '{preferred}'.",
93	**MsgStyle.WARN,
94	)
95	return orig_process(value, state)	×
96
97	return process	×
98
99	option.process = make_process(option)	×
100
101	return parser	×
102
103
104	class DeprecatedDefaultGroup(DefaultGroup):	4✔
105	"""Invokes a default subcommand, and shows a deprecation message.	4✔
106
107	Also adds the `invoked_default` boolean attribute to the context.
108	A group callback can use this information to figure out if it's being executed directly
109	(invoking the default subcommand) or because the execution flow passes onwards to a subcommand.
110	By default it's None, but it can be the name of the default subcommand to execute.
111
112	.. sourcecode:: python
113
114	import click
115	from flexmeasures.cli.utils import DeprecatedDefaultGroup
116
117	@click.group(cls=DeprecatedDefaultGroup, default="bar", deprecation_message="renamed to `foo bar`.")
118	def foo(ctx):
119	if ctx.invoked_default:
120	click.echo("foo")
121
122	@foo.command()
123	def bar():
124	click.echo("bar")
125
126	.. sourcecode:: console
127
128	$ flexmeasures foo
129	DeprecationWarning: renamed to `foo bar`.
130	foo
131	bar
132	$ flexmeasures foo bar
133	bar
134	"""
135
136	def __init__(self, args, *kwargs):	4✔
137	self.deprecation_message = "DeprecationWarning: " + kwargs.pop(	4✔
138	"deprecation_message", ""
139	)
140	super().__init__(args, *kwargs)	4✔
141
142	def get_command(self, ctx, cmd_name):	4✔
143	ctx.invoked_default = None	×
144	if cmd_name not in self.commands:	×
145	click.echo(click.style(self.deprecation_message, fg="red"), err=True)	×
146	ctx.invoked_default = self.default_cmd_name	×
147	return super().get_command(ctx, cmd_name)	×
148
149
150	def get_timerange_from_flag(	4✔
151	last_hour: bool = False,
152	last_day: bool = False,
153	last_7_days: bool = False,
154	last_month: bool = False,
155	last_year: bool = False,
156	timezone: pytz.BaseTzInfo = get_timezone(),
157	) -> tuple[datetime, datetime]:
158	"""This function returns a time range [start,end] of the last-X period.
159	See input parameters for more details.
160
161	:param bool last_hour: flag to get the time range of the last finished hour.
162	:param bool last_day: flag to get the time range for yesterday.
163	:param bool last_7_days: flag to get the time range of the previous 7 days.
164	:param bool last_month: flag to get the time range of last calendar month
165	:param bool last_year: flag to get the last completed calendar year
166	:param timezone: timezone object to represent
167	:returns: start:datetime, end:datetime
168	"""
169
170	current_hour = get_most_recent_hour().astimezone(timezone)	4✔
171
172	if last_hour: # last finished hour	4✔
173	end = current_hour	4✔
174	start = current_hour - timedelta(hours=1)	4✔
175
176	if last_day: # yesterday	4✔
177	end = current_hour.replace(hour=0)	4✔
178	start = end - timedelta(days=1)	4✔
179
180	if last_7_days: # last finished 7 day period.	4✔
181	end = current_hour.replace(hour=0)	4✔
182	start = end - timedelta(days=7)	4✔
183
184	if last_month:	4✔
185	end = current_hour.replace(	4✔
186	hour=0, day=1
187	) # get the first day of the current month
188	start = (end - timedelta(days=1)).replace(	4✔
189	day=1
190	) # get first day of the previous month
191
192	if last_year: # last calendar year	4✔
193	end = current_hour.replace(	4✔
194	month=1, day=1, hour=0
195	) # get first day of current year
196	start = (end - timedelta(days=1)).replace(	4✔
197	day=1, month=1
198	) # get first day of previous year
199
200	return start, end	4✔
201
202
203	def validate_unique(ctx, param, value):	4✔
204	"""Callback function to ensure multiple values are unique."""
205	if value is not None:	×
206	# Check if all values are unique
207	if len(value) != len(set(value)):	×
208	raise click.BadParameter("Values must be unique.")	×
209	return value	×
210
211
212	def abort(message: str):	4✔
NEW 213	click.secho(message, **MsgStyle.ERROR)	×
NEW 214	raise click.Abort()	×
215
216
217	def done(message: str):	4✔
NEW 218	click.secho(message, **MsgStyle.SUCCESS)	×

FlexMeasures / flexmeasures / 7843823363

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous