1 # The MIT License (MIT)
3 # Copyright (c) 2015-2020 Philippe Proulx <pproulx@efficios.com>
5 # Permission is hereby granted, free of charge, to any person obtaining
6 # a copy of this software and associated documentation files (the
7 # "Software"), to deal in the Software without restriction, including
8 # without limitation the rights to use, copy, modify, merge, publish,
9 # distribute, sublicense, and/or sell copies of the Software, and to
10 # permit persons to whom the Software is furnished to do so, subject to
11 # the following conditions:
13 # The above copyright notice and this permission notice shall be
14 # included in all copies or substantial portions of the Software.
16 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
17 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
18 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
19 # IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
20 # CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
21 # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
22 # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
24 import barectf
.version
as barectf_version
25 from typing
import Optional
, Any
, FrozenSet
, Mapping
, Iterator
, Set
, Union
27 from barectf
.typing
import Count
, Alignment
, _OptStr
, Id
28 import collections
.abc
36 class ByteOrder(enum
.Enum
):
43 def alignment(self
) -> Alignment
:
44 raise NotImplementedError
47 class _BitArrayFieldType(_FieldType
):
48 def __init__(self
, size
: Count
, byte_order
: Optional
[ByteOrder
] = None,
49 alignment
: Alignment
= Alignment(1)):
51 self
._byte
_order
= byte_order
52 self
._alignment
= alignment
55 def size(self
) -> Count
:
59 def byte_order(self
) -> Optional
[ByteOrder
]:
60 return self
._byte
_order
63 def alignment(self
) -> Alignment
:
64 return self
._alignment
67 class DisplayBase(enum
.Enum
):
74 class _IntegerFieldType(_BitArrayFieldType
):
75 def __init__(self
, size
: Count
, byte_order
: Optional
[ByteOrder
] = None,
76 alignment
: Optional
[Alignment
] = None,
77 preferred_display_base
: DisplayBase
= DisplayBase
.DECIMAL
):
78 effective_alignment
= 1
80 if alignment
is None and size
% 8 == 0:
81 effective_alignment
= 8
83 super().__init
__(size
, byte_order
, Alignment(effective_alignment
))
84 self
._preferred
_display
_base
= preferred_display_base
87 def preferred_display_base(self
) -> DisplayBase
:
88 return self
._preferred
_display
_base
91 class UnsignedIntegerFieldType(_IntegerFieldType
):
92 def __init__(self
, *args
):
93 super().__init
__(*args
)
94 self
._mapped
_clk
_type
_name
= None
97 class SignedIntegerFieldType(_IntegerFieldType
):
101 class EnumerationFieldTypeMappingRange
:
102 def __init__(self
, lower
: int, upper
: int):
107 def lower(self
) -> int:
111 def upper(self
) -> int:
114 def __eq__(self
, other
: Any
) -> bool:
115 if type(other
) is not type(self
):
118 return (self
._lower
, self
._upper
) == (other
._lower
, other
._upper
)
120 def __hash__(self
) -> int:
121 return hash((self
._lower
, self
._upper
))
123 def contains(self
, value
: int) -> bool:
124 return self
._lower
<= value
<= self
._upper
127 class EnumerationFieldTypeMapping
:
128 def __init__(self
, ranges
: Set
[EnumerationFieldTypeMappingRange
]):
129 self
._ranges
= frozenset(ranges
)
132 def ranges(self
) -> FrozenSet
[EnumerationFieldTypeMappingRange
]:
135 def ranges_contain_value(self
, value
: int) -> bool:
136 return any([rg
.contains(value
) for rg
in self
._ranges
])
139 _EnumFtMappings
= Mapping
[str, EnumerationFieldTypeMapping
]
142 class EnumerationFieldTypeMappings(collections
.abc
.Mapping
):
143 def __init__(self
, mappings
: _EnumFtMappings
):
144 self
._mappings
= {label
: mapping
for label
, mapping
in mappings
.items()}
146 def __getitem__(self
, key
: str) -> EnumerationFieldTypeMapping
:
147 return self
._mappings
[key
]
149 def __iter__(self
) -> Iterator
[str]:
150 return iter(self
._mappings
)
152 def __len__(self
) -> int:
153 return len(self
._mappings
)
156 class _EnumerationFieldType(_IntegerFieldType
):
157 def __init__(self
, size
: Count
, byte_order
: Optional
[ByteOrder
] = None,
158 alignment
: Optional
[Alignment
] = None,
159 preferred_display_base
: DisplayBase
= DisplayBase
.DECIMAL
,
160 mappings
: Optional
[_EnumFtMappings
] = None):
161 super().__init
__(size
, byte_order
, alignment
, preferred_display_base
)
162 self
._mappings
= EnumerationFieldTypeMappings({})
164 if mappings
is not None:
165 self
._mappings
= EnumerationFieldTypeMappings(mappings
)
168 def mappings(self
) -> EnumerationFieldTypeMappings
:
169 return self
._mappings
171 def labels_for_value(self
, value
: int) -> Set
[str]:
174 for label
, mapping
in self
._mappings
.items():
175 if mapping
.ranges_contain_value(value
):
181 class UnsignedEnumerationFieldType(_EnumerationFieldType
, UnsignedIntegerFieldType
):
185 class SignedEnumerationFieldType(_EnumerationFieldType
, SignedIntegerFieldType
):
189 class RealFieldType(_BitArrayFieldType
):
193 class StringFieldType(_FieldType
):
195 def alignment(self
) -> Alignment
:
199 class _ArrayFieldType(_FieldType
):
200 def __init__(self
, element_field_type
: _FieldType
):
201 self
._element
_field
_type
= element_field_type
204 def element_field_type(self
) -> _FieldType
:
205 return self
._element
_field
_type
208 def alignment(self
) -> Alignment
:
209 return self
._element
_field
_type
.alignment
212 class StaticArrayFieldType(_ArrayFieldType
):
213 def __init__(self
, length
: Count
, element_field_type
: _FieldType
):
214 super().__init
__(element_field_type
)
215 self
._length
= length
218 def length(self
) -> Count
:
222 class StructureFieldTypeMember
:
223 def __init__(self
, field_type
: _FieldType
):
224 self
._field
_type
= field_type
227 def field_type(self
) -> _FieldType
:
228 return self
._field
_type
231 _StructFtMembers
= Mapping
[str, StructureFieldTypeMember
]
234 class StructureFieldTypeMembers(collections
.abc
.Mapping
):
235 def __init__(self
, members
: _StructFtMembers
):
236 self
._members
= collections
.OrderedDict()
238 for name
, member
in members
.items():
239 assert type(member
) is StructureFieldTypeMember
240 self
._members
[name
] = member
242 def __getitem__(self
, key
: str) -> StructureFieldTypeMember
:
243 return self
._members
[key
]
245 def __iter__(self
) -> Iterator
[str]:
246 return iter(self
._members
)
248 def __len__(self
) -> int:
249 return len(self
._members
)
252 class StructureFieldType(_FieldType
):
253 def __init__(self
, minimum_alignment
: Alignment
= Alignment(1),
254 members
: Optional
[_StructFtMembers
] = None):
255 self
._minimum
_alignment
= minimum_alignment
256 self
._members
= StructureFieldTypeMembers({})
258 if members
is not None:
259 self
._members
= StructureFieldTypeMembers(members
)
261 self
._set
_alignment
()
263 def _set_alignment(self
):
264 self
._alignment
: Alignment
= self
._minimum
_alignment
266 for member
in self
._members
.values():
267 if member
.field_type
.alignment
> self
._alignment
:
268 self
._alignment
= member
.field_type
.alignment
271 def minimum_alignment(self
) -> Alignment
:
272 return self
._minimum
_alignment
275 def alignment(self
) -> Alignment
:
276 return self
._alignment
279 def members(self
) -> StructureFieldTypeMembers
:
286 def __eq__(self
, other
: Any
) -> bool:
287 if type(other
) is not type(self
):
290 return self
._name
== other
._name
292 def __lt__(self
, other
: '_UniqueByName'):
293 assert type(self
) is type(other
)
294 return self
._name
< other
._name
296 def __hash__(self
) -> int:
297 return hash(self
._name
)
300 _OptFt
= Optional
[_FieldType
]
301 _OptStructFt
= Optional
[StructureFieldType
]
302 LogLevel
= typing
.NewType('LogLevel', int)
305 class EventType(_UniqueByName
):
306 def __init__(self
, name
: str, log_level
: Optional
[LogLevel
] = None,
307 specific_context_field_type
: _OptStructFt
= None, payload_field_type
: _OptStructFt
= None):
308 self
._id
: Optional
[Id
] = None
310 self
._log
_level
= log_level
311 self
._specific
_context
_field
_type
= specific_context_field_type
312 self
._payload
_field
_type
= payload_field_type
315 def id(self
) -> Optional
[Id
]:
319 def name(self
) -> str:
323 def log_level(self
) -> Optional
[LogLevel
]:
324 return self
._log
_level
327 def specific_context_field_type(self
) -> _OptStructFt
:
328 return self
._specific
_context
_field
_type
331 def payload_field_type(self
) -> _OptStructFt
:
332 return self
._payload
_field
_type
335 class ClockTypeOffset
:
336 def __init__(self
, seconds
: int = 0, cycles
: Count
= Count(0)):
337 self
._seconds
= seconds
338 self
._cycles
= cycles
341 def seconds(self
) -> int:
345 def cycles(self
) -> Count
:
349 _OptUuid
= Optional
[uuidp
.UUID
]
352 class ClockType(_UniqueByName
):
353 def __init__(self
, name
: str, frequency
: Count
= Count(int(1e9
)), uuid
: _OptUuid
= None,
354 description
: _OptStr
= None, precision
: Count
= Count(0),
355 offset
: Optional
[ClockTypeOffset
] = None, origin_is_unix_epoch
: bool = False):
357 self
._frequency
= frequency
359 self
._description
= description
360 self
._precision
= precision
361 self
._offset
= ClockTypeOffset()
363 if offset
is not None:
364 self
._offset
= offset
366 self
._origin
_is
_unix
_epoch
= origin_is_unix_epoch
369 def name(self
) -> str:
373 def frequency(self
) -> Count
:
374 return self
._frequency
377 def uuid(self
) -> _OptUuid
:
381 def description(self
) -> _OptStr
:
382 return self
._description
385 def precision(self
) -> Count
:
386 return self
._precision
389 def offset(self
) -> ClockTypeOffset
:
393 def origin_is_unix_epoch(self
) -> bool:
394 return self
._origin
_is
_unix
_epoch
397 DEFAULT_FIELD_TYPE
= 'default'
398 _DefaultableUIntFt
= Union
[str, UnsignedIntegerFieldType
]
399 _OptDefaultableUIntFt
= Optional
[_DefaultableUIntFt
]
400 _OptUIntFt
= Optional
[UnsignedIntegerFieldType
]
403 class StreamTypePacketFeatures
:
404 def __init__(self
, total_size_field_type
: _DefaultableUIntFt
= DEFAULT_FIELD_TYPE
,
405 content_size_field_type
: _DefaultableUIntFt
= DEFAULT_FIELD_TYPE
,
406 beginning_time_field_type
: _OptDefaultableUIntFt
= None,
407 end_time_field_type
: _OptDefaultableUIntFt
= None,
408 discarded_events_counter_field_type
: _OptDefaultableUIntFt
= None):
409 def get_ft(user_ft
: _OptDefaultableUIntFt
) -> _OptUIntFt
:
410 if user_ft
== DEFAULT_FIELD_TYPE
:
411 return UnsignedIntegerFieldType(64)
413 return typing
.cast(_OptUIntFt
, user_ft
)
415 self
._total
_size
_field
_type
= get_ft(total_size_field_type
)
416 self
._content
_size
_field
_type
= get_ft(content_size_field_type
)
417 self
._beginning
_time
_field
_type
= get_ft(beginning_time_field_type
)
418 self
._end
_time
_field
_type
= get_ft(end_time_field_type
)
419 self
._discarded
_events
_counter
_field
_type
= get_ft(discarded_events_counter_field_type
)
422 def total_size_field_type(self
) -> _OptUIntFt
:
423 return self
._total
_size
_field
_type
426 def content_size_field_type(self
) -> _OptUIntFt
:
427 return self
._content
_size
_field
_type
430 def beginning_time_field_type(self
) -> _OptUIntFt
:
431 return self
._beginning
_time
_field
_type
434 def end_time_field_type(self
) -> _OptUIntFt
:
435 return self
._end
_time
_field
_type
438 def discarded_events_counter_field_type(self
) -> _OptUIntFt
:
439 return self
._discarded
_events
_counter
_field
_type
442 class StreamTypeEventFeatures
:
443 def __init__(self
, type_id_field_type
: _OptDefaultableUIntFt
= DEFAULT_FIELD_TYPE
,
444 time_field_type
: _OptDefaultableUIntFt
= None):
445 def get_ft(user_ft
: _OptDefaultableUIntFt
) -> _OptUIntFt
:
446 if user_ft
== DEFAULT_FIELD_TYPE
:
447 return UnsignedIntegerFieldType(64)
449 return typing
.cast(_OptUIntFt
, user_ft
)
451 self
._type
_id
_field
_type
= get_ft(type_id_field_type
)
452 self
._time
_field
_type
= get_ft(time_field_type
)
455 def type_id_field_type(self
) -> _OptUIntFt
:
456 return self
._type
_id
_field
_type
459 def time_field_type(self
) -> _OptUIntFt
:
460 return self
._time
_field
_type
463 class StreamTypeFeatures
:
464 def __init__(self
, packet_features
: Optional
[StreamTypePacketFeatures
] = None,
465 event_features
: Optional
[StreamTypeEventFeatures
] = None):
466 self
._packet
_features
= StreamTypePacketFeatures()
468 if packet_features
is not None:
469 self
._packet
_features
= packet_features
471 self
._event
_features
= StreamTypeEventFeatures()
473 if event_features
is not None:
474 self
._event
_features
= event_features
477 def packet_features(self
) -> StreamTypePacketFeatures
:
478 return self
._packet
_features
481 def event_features(self
) -> StreamTypeEventFeatures
:
482 return self
._event
_features
485 class StreamType(_UniqueByName
):
486 def __init__(self
, name
: str, event_types
: Set
[EventType
],
487 default_clock_type
: Optional
[ClockType
] = None,
488 features
: Optional
[StreamTypeFeatures
] = None,
489 packet_context_field_type_extra_members
: Optional
[_StructFtMembers
] = None,
490 event_common_context_field_type
: _OptStructFt
= None):
491 self
._id
: Optional
[Id
] = None
493 self
._default
_clock
_type
= default_clock_type
494 self
._event
_common
_context
_field
_type
= event_common_context_field_type
495 self
._event
_types
= frozenset(event_types
)
498 for index
, ev_type
in enumerate(sorted(self
._event
_types
, key
=lambda evt
: evt
.name
)):
499 assert ev_type
._id
is None
500 ev_type
._id
= Id(index
)
502 self
._set
_features
(features
)
503 self
._packet
_context
_field
_type
_extra
_members
= StructureFieldTypeMembers({})
505 if packet_context_field_type_extra_members
is not None:
506 self
._packet
_context
_field
_type
_extra
_members
= StructureFieldTypeMembers(packet_context_field_type_extra_members
)
508 self
._set
_pkt
_ctx
_ft
()
509 self
._set
_ev
_header
_ft
()
511 def _set_features(self
, features
: Optional
[StreamTypeFeatures
]):
512 if features
is not None:
513 self
._features
= features
517 pkt_beginning_time_ft
= None
518 pkt_end_time_ft
= None
520 if self
._default
_clock
_type
is not None:
521 # Automatic time field types because the stream type has a
522 # default clock type.
523 ev_time_ft
= DEFAULT_FIELD_TYPE
524 pkt_beginning_time_ft
= DEFAULT_FIELD_TYPE
525 pkt_end_time_ft
= DEFAULT_FIELD_TYPE
527 self
._features
= StreamTypeFeatures(StreamTypePacketFeatures(beginning_time_field_type
=pkt_beginning_time_ft
,
528 end_time_field_type
=pkt_end_time_ft
),
529 StreamTypeEventFeatures(time_field_type
=ev_time_ft
))
531 def _set_ft_mapped_clk_type_name(self
, ft
: Optional
[UnsignedIntegerFieldType
]):
535 if self
._default
_clock
_type
is not None:
536 assert isinstance(ft
, UnsignedIntegerFieldType
)
537 ft
._mapped
_clk
_type
_name
= self
._default
_clock
_type
.name
539 def _set_pkt_ctx_ft(self
):
542 def add_member_if_exists(name
: str, ft
: _FieldType
, set_mapped_clk_type_name
: bool = False):
546 if set_mapped_clk_type_name
:
547 self
._set
_ft
_mapped
_clk
_type
_name
(typing
.cast(UnsignedIntegerFieldType
, ft
))
549 members
[name
] = StructureFieldTypeMember(ft
)
551 members
= collections
.OrderedDict([
554 StructureFieldTypeMember(self
._features
.packet_features
.total_size_field_type
)
558 StructureFieldTypeMember(self
._features
.packet_features
.content_size_field_type
)
562 add_member_if_exists('timestamp_begin',
563 self
._features
.packet_features
.beginning_time_field_type
, True)
564 add_member_if_exists('timestamp_end', self
._features
.packet_features
.end_time_field_type
,
566 add_member_if_exists('events_discarded',
567 self
._features
.packet_features
.discarded_events_counter_field_type
)
569 if self
._packet
_context
_field
_type
_extra
_members
is not None:
570 for name
, field_type
in self
._packet
_context
_field
_type
_extra
_members
.items():
571 assert name
not in members
572 members
[name
] = field_type
574 self
._pkt
_ctx
_ft
= StructureFieldType(8, members
)
576 def _set_ev_header_ft(self
):
577 members
= collections
.OrderedDict()
579 if self
._features
.event_features
.type_id_field_type
is not None:
580 members
['id'] = StructureFieldTypeMember(self
._features
.event_features
.type_id_field_type
)
582 if self
._features
.event_features
.time_field_type
is not None:
583 ft
= self
._features
.event_features
.time_field_type
584 self
._set
_ft
_mapped
_clk
_type
_name
(ft
)
585 members
['timestamp'] = StructureFieldTypeMember(ft
)
587 self
._ev
_header
_ft
= StructureFieldType(8, members
)
590 def id(self
) -> Optional
[Id
]:
594 def name(self
) -> str:
598 def default_clock_type(self
) -> Optional
[ClockType
]:
599 return self
._default
_clock
_type
602 def features(self
) -> StreamTypeFeatures
:
603 return self
._features
606 def packet_context_field_type_extra_members(self
) -> StructureFieldTypeMembers
:
607 return self
._packet
_context
_field
_type
_extra
_members
610 def event_common_context_field_type(self
) -> _OptStructFt
:
611 return self
._event
_common
_context
_field
_type
614 def event_types(self
) -> FrozenSet
[EventType
]:
615 return self
._event
_types
618 _OptUuidFt
= Optional
[Union
[str, StaticArrayFieldType
]]
621 class TraceTypeFeatures
:
622 def __init__(self
, magic_field_type
: _OptDefaultableUIntFt
= DEFAULT_FIELD_TYPE
,
623 uuid_field_type
: _OptUuidFt
= None,
624 stream_type_id_field_type
: _OptDefaultableUIntFt
= DEFAULT_FIELD_TYPE
):
625 def get_field_type(user_ft
: Optional
[Union
[str, _FieldType
]], default_ft
: _FieldType
) -> _OptFt
:
626 if user_ft
== DEFAULT_FIELD_TYPE
:
629 return typing
.cast(_OptFt
, user_ft
)
631 self
._magic
_field
_type
= typing
.cast(_OptUIntFt
, get_field_type(magic_field_type
,
632 UnsignedIntegerFieldType(32)))
633 self
._uuid
_field
_type
= typing
.cast(Optional
[StaticArrayFieldType
], get_field_type(uuid_field_type
,
634 StaticArrayFieldType(Count(16),
635 UnsignedIntegerFieldType(8))))
636 self
._stream
_type
_id
_field
_type
= typing
.cast(_OptUIntFt
, get_field_type(stream_type_id_field_type
,
637 UnsignedIntegerFieldType(64)))
640 def magic_field_type(self
) -> _OptUIntFt
:
641 return self
._magic
_field
_type
644 def uuid_field_type(self
) -> Optional
[StaticArrayFieldType
]:
645 return self
._uuid
_field
_type
648 def stream_type_id_field_type(self
) -> _OptUIntFt
:
649 return self
._stream
_type
_id
_field
_type
653 def __init__(self
, stream_types
: Set
[StreamType
], default_byte_order
: ByteOrder
,
654 uuid
: _OptUuid
= None, features
: Optional
[TraceTypeFeatures
] = None):
655 self
._default
_byte
_order
= default_byte_order
656 self
._stream
_types
= frozenset(stream_types
)
659 for index
, stream_type
in enumerate(sorted(self
._stream
_types
, key
=lambda st
: st
.name
)):
660 assert stream_type
._id
is None
661 stream_type
._id
= Id(index
)
664 self
._set
_features
(features
)
665 self
._set
_pkt
_header
_ft
()
666 self
._set
_fts
_effective
_byte
_order
()
668 def _set_features(self
, features
: Optional
[TraceTypeFeatures
]):
669 if features
is not None:
670 self
._features
= features
673 # automatic UUID field type because the trace type has a UUID
674 uuid_ft
= None if self
._uuid
is None else DEFAULT_FIELD_TYPE
675 self
._features
= TraceTypeFeatures(uuid_field_type
=uuid_ft
)
677 def _set_pkt_header_ft(self
):
678 members
= collections
.OrderedDict()
680 def add_member_if_exists(name
: str, ft
: _OptFt
):
684 members
[name
] = StructureFieldTypeMember(ft
)
686 add_member_if_exists('magic', self
._features
.magic_field_type
)
687 add_member_if_exists('uuid', self
._features
.uuid_field_type
)
688 add_member_if_exists('stream_id', self
._features
.stream_type_id_field_type
)
689 self
._pkt
_header
_ft
= StructureFieldType(8, members
)
691 def _set_fts_effective_byte_order(self
):
692 def set_ft_effective_byte_order(ft
: _OptFt
):
696 if isinstance(ft
, _BitArrayFieldType
):
697 if ft
._byte
_order
is None:
698 assert self
._default
_byte
_order
is not None
699 ft
._byte
_order
= self
._default
_byte
_order
700 elif isinstance(ft
, StaticArrayFieldType
):
701 set_ft_effective_byte_order(ft
.element_field_type
)
702 elif isinstance(ft
, StructureFieldType
):
703 for member
in ft
.members
.values():
704 set_ft_effective_byte_order(member
.field_type
)
706 # packet header field type
707 set_ft_effective_byte_order(self
._pkt
_header
_ft
)
709 # stream type field types
710 for stream_type
in self
._stream
_types
:
711 set_ft_effective_byte_order(stream_type
._pkt
_ctx
_ft
)
712 set_ft_effective_byte_order(stream_type
._ev
_header
_ft
)
713 set_ft_effective_byte_order(stream_type
._event
_common
_context
_field
_type
)
715 # event type field types
716 for ev_type
in stream_type
.event_types
:
717 set_ft_effective_byte_order(ev_type
._specific
_context
_field
_type
)
718 set_ft_effective_byte_order(ev_type
._payload
_field
_type
)
721 def default_byte_order(self
) -> ByteOrder
:
722 return self
._default
_byte
_order
725 def uuid(self
) -> _OptUuid
:
729 def stream_types(self
) -> FrozenSet
[StreamType
]:
730 return self
._stream
_types
732 def stream_type(self
, name
: str) -> Optional
[StreamType
]:
733 for cand_stream_type
in self
._stream
_types
:
734 if cand_stream_type
.name
== name
:
735 return cand_stream_type
740 def features(self
) -> TraceTypeFeatures
:
741 return self
._features
744 _EnvEntry
= Union
[str, int]
745 _EnvEntries
= Mapping
[str, _EnvEntry
]
748 class TraceEnvironment(collections
.abc
.Mapping
):
749 def __init__(self
, environment
: _EnvEntries
):
750 self
._env
= {name
: value
for name
, value
in environment
.items()}
752 def __getitem__(self
, key
: str) -> _EnvEntry
:
753 return self
._env
[key
]
755 def __iter__(self
) -> Iterator
[str]:
756 return iter(self
._env
)
758 def __len__(self
) -> int:
759 return len(self
._env
)
763 def __init__(self
, type: TraceType
, environment
: Optional
[_EnvEntries
] = None):
765 self
._set
_env
(environment
)
767 def _set_env(self
, environment
: Optional
[_EnvEntries
]):
768 init_env
= collections
.OrderedDict([
770 ('tracer_name', 'barectf'),
771 ('tracer_major', barectf_version
.__major
_version
__),
772 ('tracer_minor', barectf_version
.__minor
_version
__),
773 ('tracer_patch', barectf_version
.__patch
_version
__),
774 ('barectf_gen_date', str(datetime
.datetime
.now().isoformat())),
777 if environment
is None:
780 init_env
.update(environment
)
781 self
._env
= TraceEnvironment(typing
.cast(_EnvEntries
, init_env
))
784 def type(self
) -> TraceType
:
788 def environment(self
) -> TraceEnvironment
:
792 _ClkTypeCTypes
= Mapping
[ClockType
, str]
795 class ClockTypeCTypes(collections
.abc
.Mapping
):
796 def __init__(self
, c_types
: _ClkTypeCTypes
):
797 self
._c
_types
= {clk_type
: c_type
for clk_type
, c_type
in c_types
.items()}
799 def __getitem__(self
, key
: ClockType
) -> str:
800 return self
._c
_types
[key
]
802 def __iter__(self
) -> Iterator
[ClockType
]:
803 return iter(self
._c
_types
)
805 def __len__(self
) -> int:
806 return len(self
._c
_types
)
809 class ConfigurationCodeGenerationHeaderOptions
:
810 def __init__(self
, identifier_prefix_definition
: bool = False,
811 default_stream_type_name_definition
: bool = False):
812 self
._identifier
_prefix
_definition
= identifier_prefix_definition
813 self
._default
_stream
_type
_name
_definition
= default_stream_type_name_definition
816 def identifier_prefix_definition(self
) -> bool:
817 return self
._identifier
_prefix
_definition
820 def default_stream_type_name_definition(self
) -> bool:
821 return self
._default
_stream
_type
_name
_definition
824 class ConfigurationCodeGenerationOptions
:
825 def __init__(self
, identifier_prefix
: str = 'barectf_', file_name_prefix
: str = 'barectf',
826 default_stream_type
: Optional
[StreamType
] = None,
827 header_options
: Optional
[ConfigurationCodeGenerationHeaderOptions
] = None,
828 clock_type_c_types
: Optional
[_ClkTypeCTypes
] = None):
829 self
._identifier
_prefix
= identifier_prefix
830 self
._file
_name
_prefix
= file_name_prefix
831 self
._default
_stream
_type
= default_stream_type
833 self
._header
_options
= ConfigurationCodeGenerationHeaderOptions()
835 if header_options
is not None:
836 self
._header
_options
= header_options
838 self
._clock
_type
_c
_types
= ClockTypeCTypes({})
840 if clock_type_c_types
is not None:
841 self
._clock
_type
_c
_types
= ClockTypeCTypes(clock_type_c_types
)
844 def identifier_prefix(self
) -> str:
845 return self
._identifier
_prefix
848 def file_name_prefix(self
) -> str:
849 return self
._file
_name
_prefix
852 def default_stream_type(self
) -> Optional
[StreamType
]:
853 return self
._default
_stream
_type
856 def header_options(self
) -> ConfigurationCodeGenerationHeaderOptions
:
857 return self
._header
_options
860 def clock_type_c_types(self
) -> ClockTypeCTypes
:
861 return self
._clock
_type
_c
_types
864 class ConfigurationOptions
:
866 code_generation_options
: Optional
[ConfigurationCodeGenerationOptions
] = None):
867 self
._code
_generation
_options
= ConfigurationCodeGenerationOptions()
869 if code_generation_options
is not None:
870 self
._code
_generation
_options
= code_generation_options
873 def code_generation_options(self
) -> ConfigurationCodeGenerationOptions
:
874 return self
._code
_generation
_options
878 def __init__(self
, trace
: Trace
, options
: Optional
[ConfigurationOptions
] = None):
880 self
._options
= ConfigurationOptions()
882 if options
is not None:
883 self
._options
= options
885 clk_type_c_types
= self
._options
.code_generation_options
.clock_type_c_types
887 for stream_type
in trace
.type.stream_types
:
888 def_clk_type
= stream_type
.default_clock_type
890 if def_clk_type
is None:
893 if def_clk_type
not in clk_type_c_types
:
894 clk_type_c_types
._c
_types
[def_clk_type
] = 'uint32_t'
897 def trace(self
) -> Trace
:
901 def options(self
) -> ConfigurationOptions
: