Extract handling logic from EventHandler and introduce dependencies

This commit is contained in:
Christophe Bedard 2019-08-02 10:31:11 +02:00
parent 54c581f54d
commit b1d2b16b6d
5 changed files with 190 additions and 142 deletions

View file

@ -11,3 +11,187 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Base processor module."""
from collections import defaultdict
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Type
from tracetools_read.utils import get_event_name
from tracetools_read.utils import get_field
from tracetools_read.utils import DictEvent
class EventMetadata():
"""Container for event metadata."""
def __init__(
self,
event_name: str,
timestamp: int,
cpu_id: int,
procname: str = None,
pid: int = None,
tid: int = None,
) -> None:
"""
Constructor.
Parameters with a default value of `None` are not mandatory,
since they are not always present.
"""
self._event_name = event_name
self._timestamp = timestamp
self._cpu_id = cpu_id
self._procname = procname
self._pid = pid
self._tid = tid
@property
def event_name(self):
return self._event_name
@property
def timestamp(self):
return self._timestamp
@property
def cpu_id(self):
return self._cpu_id
@property
def procname(self):
return self._procname
@property
def pid(self):
return self._pid
@property
def tid(self):
return self._tid
HandlerMap = Dict[str, Callable[[DictEvent, EventMetadata], None]]
HandlerMultimap = Dict[str, List[Callable[[DictEvent, EventMetadata], None]]]
class EventHandler():
"""
Base event handling class.
Provides handling functions for some events, depending on the name.
"""
def __init__(self, handler_map: HandlerMap) -> None:
"""
Constructor.
:param handler_map: the mapping from event name to handling method
"""
assert handler_map is not None and len(handler_map) > 0, f'empty map: {handler_map}'
self._handler_map = handler_map
@property
def handler_map(self) -> HandlerMap:
return self._handler_map
def get_dependants(self) -> List[Type['EventHandler']]:
# Default: no dependants
# Subclasses should override this method id they want to declare dependants
return []
@classmethod
def process(cls, events: List[DictEvent]) -> 'EventHandler':
"""
Util method that creates a `Processor` calls `process()` with an instance of the class.
:param events: the list of events
:return: the processor object after processing
"""
assert cls != EventHandler, 'only call process() from inheriting classes'
handler_object = cls() # pylint: disable=no-value-for-parameter
processor = Processor(handler_object)
processor.process(events)
return handler_object
class Processor():
"""Base processor class."""
def __init__(self, *handlers: EventHandler) -> None:
"""
Constructor.
:param
"""
print('num handlers before:', len(handlers))
self._add_dependant_handlers(handlers)
self._handlers = handlers
print('num handlers after:', len(handlers))
def _add_dependant_handlers(self, handlers: List[EventHandler]) -> None:
"""Checks handlers and add dependant handlers if not included. Ordered."""
# TODO
# For each handler object, check if its dependants are included
# If not, add them _before_
pass
def _get_handler_maps(self) -> HandlerMultimap:
"""Collects and merges `HandlerMap`s from all events handlers into a `HandlerMultimap`."""
handler_multimap = defaultdict(list)
for handler in self._handlers:
handler_map = handler.handler_map
print(f'{handler}:: {handler_map}')
for event_name, handler in handler_map.items():
handler_multimap[event_name].append(handler)
return handler_multimap
def process(self, events: List[DictEvent]) -> None:
"""
Process all events.
:param events: the events to process
"""
self._handler_multimap = self._get_handler_maps()
print(f'multimap: {self._handler_multimap}')
for event in events:
self._process_event(event)
def _process_event(self, event: DictEvent) -> None:
event_name = get_event_name(event)
print(f"event name: {event_name}")
handler_functions = self._handler_multimap.get(event_name, None)
if handler_functions is not None:
print(f'\thandler functions: {handler_functions}')
for handler_function in handler_functions:
print(f'\t\thandler function: {handler_function}')
timestamp = get_field(event, '_timestamp')
cpu_id = get_field(event, 'cpu_id')
# TODO perhaps validate fields depending on the type of event,
# i.e. all UST events should have procname, (v)pid and (v)tid
# context info, since analyses might not work otherwise
procname = get_field(event, 'procname', raise_if_not_found=False)
pid = get_field(
event,
'vpid',
default=get_field(
event,
'pid',
raise_if_not_found=False),
raise_if_not_found=False)
tid = get_field(
event,
'vtid',
default=get_field(
event,
'tid',
raise_if_not_found=False),
raise_if_not_found=False)
metadata = EventMetadata(event_name, timestamp, cpu_id, procname, pid, tid)
handler_function(event, metadata)
input()

View file

@ -18,8 +18,8 @@ from typing import Dict
from tracetools_read.utils import get_field
from .handler import EventHandler
from .handler import EventMetadata
from . import EventHandler
from . import EventMetadata
from ..data_model.cpu_time import CpuTimeDataModel

View file

@ -1,136 +0,0 @@
# Copyright 2019 Robert Bosch GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module for event handling."""
from typing import Callable
from typing import Dict
from typing import List
from tracetools_read.utils import get_event_name
from tracetools_read.utils import get_field
class EventMetadata():
"""Container for event metadata."""
def __init__(
self,
event_name: str,
timestamp: int,
cpu_id: int,
procname: str = None,
pid: int = None,
tid: int = None,
) -> None:
"""
Constructor.
Parameters with a default value of `None` are not mandatory,
since they are not always present.
"""
self._event_name = event_name
self._timestamp = timestamp
self._cpu_id = cpu_id
self._procname = procname
self._pid = pid
self._tid = tid
@property
def event_name(self):
return self._event_name
@property
def timestamp(self):
return self._timestamp
@property
def cpu_id(self):
return self._cpu_id
@property
def procname(self):
return self._procname
@property
def pid(self):
return self._pid
@property
def tid(self):
return self._tid
class EventHandler():
"""Base event handling class."""
def __init__(self, handler_map: Dict[str, Callable[[Dict, EventMetadata], None]]) -> None:
"""
Constructor.
:param handler_map: the mapping from event name to handling method
"""
assert handler_map is not None and len(handler_map) > 0, f'empty map: {handler_map}'
self._handler_map = handler_map
def handle_events(self, events: List[Dict[str, str]]) -> None:
"""
Handle events by calling their handlers.
:param events: the events to process
"""
for event in events:
self._handle(event)
def _handle(self, event: Dict[str, str]) -> None:
event_name = get_event_name(event)
handler_function = self._handler_map.get(event_name, None)
if handler_function is not None:
timestamp = get_field(event, '_timestamp')
cpu_id = get_field(event, 'cpu_id')
# TODO perhaps validate fields depending on the type of event,
# i.e. all UST events should have procname, (v)pid and (v)tid
# context info, since analyses might not work otherwise
procname = get_field(event, 'procname', raise_if_not_found=False)
pid = get_field(
event,
'vpid',
default=get_field(
event,
'pid',
raise_if_not_found=False),
raise_if_not_found=False)
tid = get_field(
event,
'vtid',
default=get_field(
event,
'tid',
raise_if_not_found=False),
raise_if_not_found=False)
metadata = EventMetadata(event_name, timestamp, cpu_id, procname, pid, tid)
handler_function(event, metadata)
@classmethod
def process(cls, events: List[Dict[str, str]]) -> 'EventHandler':
"""
Create processor and process unpickled events to create model.
:param events: the list of events
:return: the processor object after processing
"""
assert cls != EventHandler, 'only call process() from inheriting classes'
processor = cls() # pylint: disable=no-value-for-parameter
processor.handle_events(events)
return processor

View file

@ -22,8 +22,8 @@ from typing import Union
from tracetools_read.utils import get_field
from .handler import EventHandler
from .handler import EventMetadata
from . import EventHandler
from . import EventMetadata
from ..data_model.profile import ProfileDataModel

View file

@ -18,8 +18,8 @@ from typing import Dict
from tracetools_read.utils import get_field
from .handler import EventHandler
from .handler import EventMetadata
from . import EventHandler
from . import EventMetadata
from ..data_model.ros import RosDataModel