10488746911

Committed 21 Aug 2024 11:10AM UTC coverage: 67.43% (-0.02%) from 67.453%

Build # 10488746911

Build Type

Pull #1145

github

Committed by

joshuaunity

Commit Message

docs: fixed typo with wrong url linking

Pull Request Pull Request #1145: Feat-dynamic-logo

Run Details

6091 of 9882 branches covered (61.64%)

Branch coverage included in aggregate %.

8 of 14 new or added lines in 5 files covered. (57.14%)

1 existing line in 1 file now uncovered.

7724 of 10606 relevant lines covered (72.83%)

5.77 hits per line

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

60.0

/flexmeasures/cli/utils.py

"""
Utils for FlexMeasures CLI
"""

from __future__ import annotations

from typing import Any
from datetime import datetime, timedelta

import re
import click
import pytz
from click_default_group import DefaultGroup

from flexmeasures.utils.time_utils import get_most_recent_hour, get_timezone
from flexmeasures import Sensor


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)


def path_to_str(path: list, separator: str = ">") -> str:
    """
    Converts a list representing a path to a string format, using a specified separator.
    """

    return separator.join(path)


def are_all_equal(paths: list[list[str]]) -> bool:
    """
    Checks if all given entity paths represent the same path.
    """
    return len(set(path_to_str(p) for p in paths)) == 1


def reduce_entity_paths(asset_paths: list[list[str]]) -> list[list[str]]:
    """
    Simplifies a list of entity paths by trimming their common ancestor.

    Examples:
    >>> reduce_entity_paths([["Account1", "Asset1"], ["Account2", "Asset2"]])
    [["Account1", "Asset1"], ["Account2", "Asset2"]]

    >>> reduce_entity_paths([["Asset1"], ["Asset2"]])
    [["Asset1"], ["Asset2"]]

    >>> reduce_entity_paths([["Account1", "Asset1"], ["Account1", "Asset2"]])
    [["Asset1"], ["Asset2"]]

    >>> reduce_entity_paths([["Asset1", "Asset2"], ["Asset1"]])
    [["Asset1"], ["Asset1", "Asset2"]]

    >>> reduce_entity_paths([["Account1", "Asset", "Asset1"], ["Account1", "Asset", "Asset2"]])
    [["Asset1"], ["Asset2"]]
    """
    reduced_entities = 0

    # At least we need to leave one entity in each list
    max_reduced_entities = min([len(p) - 1 for p in asset_paths])

    # Find the common path
    while (
        are_all_equal([p[:reduced_entities] for p in asset_paths])
        and reduced_entities <= max_reduced_entities
    ):
        reduced_entities += 1

    return [p[reduced_entities - 1 :] for p in asset_paths]


def get_sensor_aliases(
    sensors: list[Sensor],
    reduce_paths: bool = True,
    separator: str = "/",
) -> dict:
    """
    Generates aliases for all sensors by appending a unique path to each sensor's name.

    Parameters:
    :param sensors:         A list of Sensor objects.
    :param reduce_paths:    Flag indicating whether to reduce each sensor's entity path. Defaults to True.
    :param separator:       Character or string used to separate entities within each sensor's path. Defaults to "/".

    :return: A dictionary mapping sensor IDs to their generated aliases.
    """

    entity_paths = [
        s.generic_asset.get_path(separator=separator).split(separator) for s in sensors
    ]
    if reduce_paths:
        entity_paths = reduce_entity_paths(entity_paths)
    entity_paths = [path_to_str(p, separator=separator) for p in entity_paths]

    aliases = {
        sensor.id: f"{sensor.name} ({path})"
        for path, sensor in zip(entity_paths, sensors)
    }

    return aliases


def validate_color_hex(ctx, param, value):
    """
    Validates that a given value is a valid hex color code.

    Parameters:
    :param ctx:     Click context.
    :param param:   Click parameter. Hex value.
    """
    if value is None:
        return value
    hex_pattern = re.compile(r"^#?([A-Fa-f0-9]{6}|[A-Fa-f0-9]{3})$")
    if re.match(hex_pattern, value):
        return value
    else:
        raise click.BadParameter(f"{param.name} must be a valid hex color code.")


def validate_url(ctx, param, value):
    """
    Validates that a given value is a valid URL format using regex.

    Parameters:
    :param ctx:     Click context.
    :param param:   Click parameter. URL value.
    :param value:   The URL to validate.
    """
    url_regex = re.compile(
        r"^(https?|ftp)://"  # Protocol: http, https, or ftp
        r"((([A-Za-z0-9-]+\.)+[A-Za-z]{2,6})|"  # Domain name
        r"(\d{1,3}\.){3}\d{1,3})"  # OR IPv4
        r"(:\d+)?"  # Port
        r"(/([A-Za-z0-9$_.+!*\'(),;?&=-]|%[0-9A-Fa-f]{2})*)*"  # Path
        r"(\?([A-Za-z0-9$_.+!*\'(),;?&=-]|%[0-9A-Fa-f]{2})*)?"  # Query string
    )

    if not url_regex.match(value):
        raise click.BadParameter(f"'{value}' is not a valid URL.")

    # check if more than 255 characters
    if len(value) > 255:
        raise click.BadParameter(
            "provided logo-url is too long. Maximum length is 255 characters."
        )

    return value

1	"""	4✔
2	Utils for FlexMeasures CLI
3	"""
4
5	from __future__ import annotations	8✔
6
7	from typing import Any	8✔
8	from datetime import datetime, timedelta	8✔
9
10	import re	8✔
11	import click	8✔
12	import pytz	8✔
13	from click_default_group import DefaultGroup	8✔
14
15	from flexmeasures.utils.time_utils import get_most_recent_hour, get_timezone	8✔
16	from flexmeasures import Sensor	8✔
17
18
19	class MsgStyle(object):	8✔
20	"""Stores the text styles for the different events	4✔
21
22	Styles options are the attributes of the `click.style` which can be found
23	[here](https://click.palletsprojects.com/en/8.1.x/api/#click.style).
24
25	"""
26
27	SUCCESS: dict[str, Any] = {"fg": "green"}	8✔
28	WARN: dict[str, Any] = {"fg": "yellow"}	8✔
29	ERROR: dict[str, Any] = {"fg": "red"}	8✔
30
31
32	class DeprecatedOption(click.Option):	8✔
33	"""A custom option that can be used to mark an option as deprecated.	4✔
34
35	References
36	----------------
37
38	Copied from https://stackoverflow.com/a/50402799/13775459
39	"""
40
41	def __init__(self, args, *kwargs):	8✔
42	self.deprecated = kwargs.pop("deprecated", ())	8✔
43	self.preferred = kwargs.pop("preferred", args[0][-1])	8✔
44	super(DeprecatedOption, self).__init__(args, *kwargs)	8✔
45
46
47	class DeprecatedOptionsCommand(click.Command):	8✔
48	"""A custom command that can be used to mark options as deprecated.	4✔
49
50	References
51	----------------
52
53	Adapted from https://stackoverflow.com/a/50402799/13775459
54	"""
55
56	def make_parser(self, ctx):	8✔
57	"""Hook 'make_parser' and during processing check the name
58	used to invoke the option to see if it is preferred"""
59
60	parser = super(DeprecatedOptionsCommand, self).make_parser(ctx)	×
61
62	# get the parser options
63	options = set(parser._short_opt.values())	×
64	options \|= set(parser._long_opt.values())	×
65
66	for option in options:	×
67	if not isinstance(option.obj, DeprecatedOption):	×
68	continue	×
69
70	def make_process(an_option):	×
71	"""Construct a closure to the parser option processor"""
72
73	orig_process = an_option.process	×
74	deprecated = getattr(an_option.obj, "deprecated", None)	×
75	preferred = getattr(an_option.obj, "preferred", None)	×
76	msg = "Expected `deprecated` value for `{}`"	×
77	assert deprecated is not None, msg.format(an_option.obj.name)	×
78
79	def process(value, state):	×
80	"""The function above us on the stack used 'opt' to
81	pick option from a dict, see if it is deprecated"""
82
83	# reach up the stack and get 'opt'
84	import inspect	×
85
86	frame = inspect.currentframe()	×
87	try:	×
88	opt = frame.f_back.f_locals.get("opt")	×
89	finally:
90	del frame	×
91
92	if opt in deprecated:	×
93	click.secho(	×
94	f"Option '{opt}' will be replaced by '{preferred}'.",
95	**MsgStyle.WARN,
96	)
97	return orig_process(value, state)	×
98
99	return process	×
100
101	option.process = make_process(option)	×
102
103	return parser	×
104
105
106	class DeprecatedDefaultGroup(DefaultGroup):	8✔
107	"""Invokes a default subcommand, and shows a deprecation message.	4✔
108
109	Also adds the `invoked_default` boolean attribute to the context.
110	A group callback can use this information to figure out if it's being executed directly
111	(invoking the default subcommand) or because the execution flow passes onwards to a subcommand.
112	By default it's None, but it can be the name of the default subcommand to execute.
113
114	.. sourcecode:: python
115
116	import click
117	from flexmeasures.cli.utils import DeprecatedDefaultGroup
118
119	@click.group(cls=DeprecatedDefaultGroup, default="bar", deprecation_message="renamed to `foo bar`.")
120	def foo(ctx):
121	if ctx.invoked_default:
122	click.echo("foo")
123
124	@foo.command()
125	def bar():
126	click.echo("bar")
127
128	.. sourcecode:: console
129
130	$ flexmeasures foo
131	DeprecationWarning: renamed to `foo bar`.
132	foo
133	bar
134	$ flexmeasures foo bar
135	bar
136	"""
137
138	def __init__(self, args, *kwargs):	8✔
139	self.deprecation_message = "DeprecationWarning: " + kwargs.pop(	8✔
140	"deprecation_message", ""
141	)
142	super().__init__(args, *kwargs)	8✔
143
144	def get_command(self, ctx, cmd_name):	8✔
145	ctx.invoked_default = None	×
146	if cmd_name not in self.commands:	×
147	click.echo(click.style(self.deprecation_message, fg="red"), err=True)	×
148	ctx.invoked_default = self.default_cmd_name	×
149	return super().get_command(ctx, cmd_name)	×
150
151
152	def get_timerange_from_flag(	8✔
153	last_hour: bool = False,
154	last_day: bool = False,
155	last_7_days: bool = False,
156	last_month: bool = False,
157	last_year: bool = False,
158	timezone: pytz.BaseTzInfo = get_timezone(),
159	) -> tuple[datetime, datetime]:
160	"""This function returns a time range [start,end] of the last-X period.
161	See input parameters for more details.
162
163	:param bool last_hour: flag to get the time range of the last finished hour.
164	:param bool last_day: flag to get the time range for yesterday.
165	:param bool last_7_days: flag to get the time range of the previous 7 days.
166	:param bool last_month: flag to get the time range of last calendar month
167	:param bool last_year: flag to get the last completed calendar year
168	:param timezone: timezone object to represent
169	:returns: start:datetime, end:datetime
170	"""
171
172	current_hour = get_most_recent_hour().astimezone(timezone)	8✔
173
174	if last_hour: # last finished hour	8✔
175	end = current_hour	8✔
176	start = current_hour - timedelta(hours=1)	8✔
177
178	if last_day: # yesterday	8✔
179	end = current_hour.replace(hour=0)	8✔
180	start = end - timedelta(days=1)	8✔
181
182	if last_7_days: # last finished 7 day period.	8✔
183	end = current_hour.replace(hour=0)	8✔
184	start = end - timedelta(days=7)	8✔
185
186	if last_month:	8✔
187	end = current_hour.replace(	8✔
188	hour=0, day=1
189	) # get the first day of the current month
190	start = (end - timedelta(days=1)).replace(	8✔
191	day=1
192	) # get first day of the previous month
193
194	if last_year: # last calendar year	8✔
195	end = current_hour.replace(	8✔
196	month=1, day=1, hour=0
197	) # get first day of current year
198	start = (end - timedelta(days=1)).replace(	8✔
199	day=1, month=1
200	) # get first day of previous year
201
202	return start, end	8✔
203
204
205	def validate_unique(ctx, param, value):	8✔
206	"""Callback function to ensure multiple values are unique."""
207	if value is not None:	×
208	# Check if all values are unique
209	if len(value) != len(set(value)):	×
210	raise click.BadParameter("Values must be unique.")	×
211	return value	×
212
213
214	def abort(message: str):	8✔
215	click.secho(message, **MsgStyle.ERROR)	×
216	raise click.Abort()	×
217
218
219	def done(message: str):	8✔
220	click.secho(message, **MsgStyle.SUCCESS)	8✔
221
222
223	def path_to_str(path: list, separator: str = ">") -> str:	8✔
224	"""
225	Converts a list representing a path to a string format, using a specified separator.
226	"""
227
228	return separator.join(path)	8✔
229
230
231	def are_all_equal(paths: list[list[str]]) -> bool:	8✔
232	"""
233	Checks if all given entity paths represent the same path.
234	"""
235	return len(set(path_to_str(p) for p in paths)) == 1	8✔
236
237
238	def reduce_entity_paths(asset_paths: list[list[str]]) -> list[list[str]]:	8✔
239	"""
240	Simplifies a list of entity paths by trimming their common ancestor.
241
242	Examples:
243	>>> reduce_entity_paths([["Account1", "Asset1"], ["Account2", "Asset2"]])
244	[["Account1", "Asset1"], ["Account2", "Asset2"]]
245
246	>>> reduce_entity_paths([["Asset1"], ["Asset2"]])
247	[["Asset1"], ["Asset2"]]
248
249	>>> reduce_entity_paths([["Account1", "Asset1"], ["Account1", "Asset2"]])
250	[["Asset1"], ["Asset2"]]
251
252	>>> reduce_entity_paths([["Asset1", "Asset2"], ["Asset1"]])
253	[["Asset1"], ["Asset1", "Asset2"]]
254
255	>>> reduce_entity_paths([["Account1", "Asset", "Asset1"], ["Account1", "Asset", "Asset2"]])
256	[["Asset1"], ["Asset2"]]
257	"""
258	reduced_entities = 0	8✔
259
260	# At least we need to leave one entity in each list
261	max_reduced_entities = min([len(p) - 1 for p in asset_paths])	8✔
262
263	# Find the common path
264	while (	8✔
265	are_all_equal([p[:reduced_entities] for p in asset_paths])
266	and reduced_entities <= max_reduced_entities
267	):
268	reduced_entities += 1	8✔
269
270	return [p[reduced_entities - 1 :] for p in asset_paths]	8✔
271
272
273	def get_sensor_aliases(	8✔
274	sensors: list[Sensor],
275	reduce_paths: bool = True,
276	separator: str = "/",
277	) -> dict:
278	"""
279	Generates aliases for all sensors by appending a unique path to each sensor's name.
280
281	Parameters:
282	:param sensors: A list of Sensor objects.
283	:param reduce_paths: Flag indicating whether to reduce each sensor's entity path. Defaults to True.
284	:param separator: Character or string used to separate entities within each sensor's path. Defaults to "/".
285
286	:return: A dictionary mapping sensor IDs to their generated aliases.
287	"""
288
289	entity_paths = [	8✔
290	s.generic_asset.get_path(separator=separator).split(separator) for s in sensors
291	]
292	if reduce_paths:	8✔
293	entity_paths = reduce_entity_paths(entity_paths)	8✔
294	entity_paths = [path_to_str(p, separator=separator) for p in entity_paths]	8!
295
296	aliases = {	8✔
297	sensor.id: f"{sensor.name} ({path})"
298	for path, sensor in zip(entity_paths, sensors)
299	}
300
301	return aliases	8✔
302
303
304	def validate_color_hex(ctx, param, value):	8✔
305	"""
306	Validates that a given value is a valid hex color code.
307
308	Parameters:
309	:param ctx: Click context.
310	:param param: Click parameter. Hex value.
311	"""
312	if value is None:	×
313	return value	×
314	hex_pattern = re.compile(r"^#?([A-Fa-f0-9]{6}\|[A-Fa-f0-9]{3})$")	×
315	if re.match(hex_pattern, value):	×
316	return value	×
317	else:
318	raise click.BadParameter(f"{param.name} must be a valid hex color code.")	×
319
320
321	def validate_url(ctx, param, value):	8✔
322	"""
323	Validates that a given value is a valid URL format using regex.
324
325	Parameters:
326	:param ctx: Click context.
327	:param param: Click parameter. URL value.
328	:param value: The URL to validate.
329	"""
NEW 330	url_regex = re.compile(	×
331	r"^(https?\|ftp)://" # Protocol: http, https, or ftp
332	r"((([A-Za-z0-9-]+\.)+[A-Za-z]{2,6})\|" # Domain name
333	r"(\d{1,3}\.){3}\d{1,3})" # OR IPv4
334	r"(:\d+)?" # Port
335	r"(/([A-Za-z0-9$_.+!\'(),;?&=-]\|%[0-9A-Fa-f]{2}))*" # Path
336	r"(\?([A-Za-z0-9$_.+!\'(),;?&=-]\|%[0-9A-Fa-f]{2}))?" # Query string
337	)
338
NEW 339	if not url_regex.match(value):	×
NEW 340	raise click.BadParameter(f"'{value}' is not a valid URL.")	×
341
342	# check if more than 255 characters
NEW 343	if len(value) > 255:	×
NEW 344	raise click.BadParameter(	×
345	"provided logo-url is too long. Maximum length is 255 characters."
346	)
347
NEW 348	return value	×

FlexMeasures / flexmeasures / 10488746911

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