Skip to content

Commit

Permalink
Add ClassVar
Browse files Browse the repository at this point in the history
  • Loading branch information
InvincibleRMC committed Jul 30, 2024
1 parent 0a68da4 commit 13a0469
Showing 1 changed file with 13 additions and 8 deletions.
21 changes: 13 additions & 8 deletions rclpy/rclpy/impl/rcutils_logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,15 @@
import sys
from types import FrameType
from typing import cast
from typing import ClassVar
from typing import Dict
from typing import List
from typing import Literal
from typing import NamedTuple
from typing import Optional
from typing import Tuple
from typing import Type
from typing import TypeAlias
from typing import TypedDict
from typing import Union

Expand All @@ -39,6 +42,8 @@
from typing_extensions import Unpack


SupportedFiltersKeys: TypeAlias = Literal['throttle', 'skip_first', 'once']

# Known filenames from which logging methods can be called (will be ignored in `_find_caller`).
_internal_callers: List[str] = []
# This will cause rclpy filenames to be registered in `_internal_callers` on first logging call.
Expand Down Expand Up @@ -97,7 +102,7 @@ def __new__(cls, frame: Optional[FrameType] = None) -> 'CallerId':
class RcutilsLoggerContext(TypedDict):
name: str
severity: LoggingSeverity
filters: List[str]
filters: List[SupportedFiltersKeys]


class OnceContext(RcutilsLoggerContext):
Expand Down Expand Up @@ -140,7 +145,7 @@ class LoggingFilter:
A default value of None makes a parameter required.
"""
params: LoggingFilterParams = {}
params: ClassVar[LoggingFilterParams] = {}

"""
Initialize the context of a logging call, e.g. declare variables needed for
Expand Down Expand Up @@ -169,7 +174,7 @@ def should_log(context: RcutilsLoggerContext) -> bool:
class Once(LoggingFilter):
"""Ignore all log calls except the first one."""

params: LoggingFilterParams = {
params: ClassVar[LoggingFilterParams] = {
'once': None,
}

Expand All @@ -193,7 +198,7 @@ def should_log(context: RcutilsLoggerContext) -> bool:
class Throttle(LoggingFilter):
"""Ignore log calls if the last call is not longer ago than the specified duration."""

params: LoggingFilterParams = {
params: ClassVar[LoggingFilterParams] = {
'throttle_duration_sec': None,
'throttle_time_source_type': Clock(),
}
Expand Down Expand Up @@ -225,7 +230,7 @@ def should_log(context: RcutilsLoggerContext) -> bool:
class SkipFirst(LoggingFilter):
"""Ignore the first log call but process all subsequent calls."""

params: LoggingFilterParams = {
params: ClassVar[LoggingFilterParams] = {
'skip_first': None,
}

Expand All @@ -247,19 +252,19 @@ def should_log(context: RcutilsLoggerContext) -> bool:


# The ordering of this dictionary defines the order in which filters will be processed.
supported_filters: OrderedDict[str, Type[LoggingFilter]] = OrderedDict()
supported_filters: OrderedDict[SupportedFiltersKeys, Type[LoggingFilter]] = OrderedDict()
supported_filters['throttle'] = Throttle
supported_filters['skip_first'] = SkipFirst
supported_filters['once'] = Once


def get_filters_from_kwargs(**kwargs: 'Unpack[LoggingFilterArgs]') -> List[str]:
def get_filters_from_kwargs(**kwargs: 'Unpack[LoggingFilterArgs]') -> List[SupportedFiltersKeys]:
"""
Determine which filters have had parameters specified in the given keyword arguments.
Returns the list of filters using the order specified by `supported_filters`.
"""
detected_filters: List[str] = []
detected_filters: List[SupportedFiltersKeys] = []
all_supported_params: List[str] = []
for supported_filter, filter_class in supported_filters.items():
filter_params = filter_class.params.keys()
Expand Down

0 comments on commit 13a0469

Please sign in to comment.