cli.py: fix Pylint errors
[deliverable/barectf.git] / barectf / config.py
1 # The MIT License (MIT)
2 #
3 # Copyright (c) 2015-2020 Philippe Proulx <pproulx@efficios.com>
4 #
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:
12 #
13 # The above copyright notice and this permission notice shall be
14 # included in all copies or substantial portions of the Software.
15 #
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.
23
24 import barectf.version as barectf_version
25 from typing import Optional, Any, FrozenSet, Mapping, Iterator, Set, Union
26 import typing
27 from barectf.typing import Count, Alignment, _OptStr, Id
28 import collections.abc
29 import collections
30 import datetime
31 import enum
32 import uuid as uuidp
33
34
35 @enum.unique
36 class ByteOrder(enum.Enum):
37 LITTLE_ENDIAN = 'le'
38 BIG_ENDIAN = 'be'
39
40
41 class _FieldType:
42 @property
43 def alignment(self) -> Alignment:
44 raise NotImplementedError
45
46
47 class _BitArrayFieldType(_FieldType):
48 def __init__(self, size: Count, byte_order: Optional[ByteOrder] = None,
49 alignment: Alignment = Alignment(1)):
50 self._size = size
51 self._byte_order = byte_order
52 self._alignment = alignment
53
54 @property
55 def size(self) -> Count:
56 return self._size
57
58 @property
59 def byte_order(self) -> Optional[ByteOrder]:
60 return self._byte_order
61
62 @property
63 def alignment(self) -> Alignment:
64 return self._alignment
65
66
67 class DisplayBase(enum.Enum):
68 BINARY = 2
69 OCTAL = 8
70 DECIMAL = 10
71 HEXADECIMAL = 16
72
73
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
79
80 if alignment is None and size % 8 == 0:
81 effective_alignment = 8
82
83 super().__init__(size, byte_order, Alignment(effective_alignment))
84 self._preferred_display_base = preferred_display_base
85
86 @property
87 def preferred_display_base(self) -> DisplayBase:
88 return self._preferred_display_base
89
90
91 class UnsignedIntegerFieldType(_IntegerFieldType):
92 def __init__(self, *args):
93 super().__init__(*args)
94 self._mapped_clk_type_name = None
95
96
97 class SignedIntegerFieldType(_IntegerFieldType):
98 pass
99
100
101 class EnumerationFieldTypeMappingRange:
102 def __init__(self, lower: int, upper: int):
103 self._lower = lower
104 self._upper = upper
105
106 @property
107 def lower(self) -> int:
108 return self._lower
109
110 @property
111 def upper(self) -> int:
112 return self._upper
113
114 def __eq__(self, other: Any) -> bool:
115 if type(other) is not type(self):
116 return False
117
118 return (self._lower, self._upper) == (other._lower, other._upper)
119
120 def __hash__(self) -> int:
121 return hash((self._lower, self._upper))
122
123 def contains(self, value: int) -> bool:
124 return self._lower <= value <= self._upper
125
126
127 class EnumerationFieldTypeMapping:
128 def __init__(self, ranges: Set[EnumerationFieldTypeMappingRange]):
129 self._ranges = frozenset(ranges)
130
131 @property
132 def ranges(self) -> FrozenSet[EnumerationFieldTypeMappingRange]:
133 return self._ranges
134
135 def ranges_contain_value(self, value: int) -> bool:
136 return any([rg.contains(value) for rg in self._ranges])
137
138
139 _EnumFtMappings = Mapping[str, EnumerationFieldTypeMapping]
140
141
142 class EnumerationFieldTypeMappings(collections.abc.Mapping):
143 def __init__(self, mappings: _EnumFtMappings):
144 self._mappings = {label: mapping for label, mapping in mappings.items()}
145
146 def __getitem__(self, key: str) -> EnumerationFieldTypeMapping:
147 return self._mappings[key]
148
149 def __iter__(self) -> Iterator[str]:
150 return iter(self._mappings)
151
152 def __len__(self) -> int:
153 return len(self._mappings)
154
155
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({})
163
164 if mappings is not None:
165 self._mappings = EnumerationFieldTypeMappings(mappings)
166
167 @property
168 def mappings(self) -> EnumerationFieldTypeMappings:
169 return self._mappings
170
171 def labels_for_value(self, value: int) -> Set[str]:
172 labels = set()
173
174 for label, mapping in self._mappings.items():
175 if mapping.ranges_contain_value(value):
176 labels.add(label)
177
178 return labels
179
180
181 class UnsignedEnumerationFieldType(_EnumerationFieldType, UnsignedIntegerFieldType):
182 pass
183
184
185 class SignedEnumerationFieldType(_EnumerationFieldType, SignedIntegerFieldType):
186 pass
187
188
189 class RealFieldType(_BitArrayFieldType):
190 pass
191
192
193 class StringFieldType(_FieldType):
194 @property
195 def alignment(self) -> Alignment:
196 return Alignment(8)
197
198
199 class _ArrayFieldType(_FieldType):
200 def __init__(self, element_field_type: _FieldType):
201 self._element_field_type = element_field_type
202
203 @property
204 def element_field_type(self) -> _FieldType:
205 return self._element_field_type
206
207 @property
208 def alignment(self) -> Alignment:
209 return self._element_field_type.alignment
210
211
212 class StaticArrayFieldType(_ArrayFieldType):
213 def __init__(self, length: Count, element_field_type: _FieldType):
214 super().__init__(element_field_type)
215 self._length = length
216
217 @property
218 def length(self) -> Count:
219 return self._length
220
221
222 class StructureFieldTypeMember:
223 def __init__(self, field_type: _FieldType):
224 self._field_type = field_type
225
226 @property
227 def field_type(self) -> _FieldType:
228 return self._field_type
229
230
231 _StructFtMembers = Mapping[str, StructureFieldTypeMember]
232
233
234 class StructureFieldTypeMembers(collections.abc.Mapping):
235 def __init__(self, members: _StructFtMembers):
236 self._members = collections.OrderedDict()
237
238 for name, member in members.items():
239 assert type(member) is StructureFieldTypeMember
240 self._members[name] = member
241
242 def __getitem__(self, key: str) -> StructureFieldTypeMember:
243 return self._members[key]
244
245 def __iter__(self) -> Iterator[str]:
246 return iter(self._members)
247
248 def __len__(self) -> int:
249 return len(self._members)
250
251
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({})
257
258 if members is not None:
259 self._members = StructureFieldTypeMembers(members)
260
261 self._set_alignment()
262
263 def _set_alignment(self):
264 self._alignment: Alignment = self._minimum_alignment
265
266 for member in self._members.values():
267 if member.field_type.alignment > self._alignment:
268 self._alignment = member.field_type.alignment
269
270 @property
271 def minimum_alignment(self) -> Alignment:
272 return self._minimum_alignment
273
274 @property
275 def alignment(self) -> Alignment:
276 return self._alignment
277
278 @property
279 def members(self) -> StructureFieldTypeMembers:
280 return self._members
281
282
283 class _UniqueByName:
284 _name: str
285
286 def __eq__(self, other: Any) -> bool:
287 if type(other) is not type(self):
288 return False
289
290 return self._name == other._name
291
292 def __lt__(self, other: '_UniqueByName'):
293 assert type(self) is type(other)
294 return self._name < other._name
295
296 def __hash__(self) -> int:
297 return hash(self._name)
298
299
300 _OptFt = Optional[_FieldType]
301 _OptStructFt = Optional[StructureFieldType]
302 LogLevel = typing.NewType('LogLevel', int)
303
304
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
309 self._name = name
310 self._log_level = log_level
311 self._specific_context_field_type = specific_context_field_type
312 self._payload_field_type = payload_field_type
313
314 @property
315 def id(self) -> Optional[Id]:
316 return self._id
317
318 @property
319 def name(self) -> str:
320 return self._name
321
322 @property
323 def log_level(self) -> Optional[LogLevel]:
324 return self._log_level
325
326 @property
327 def specific_context_field_type(self) -> _OptStructFt:
328 return self._specific_context_field_type
329
330 @property
331 def payload_field_type(self) -> _OptStructFt:
332 return self._payload_field_type
333
334
335 class ClockTypeOffset:
336 def __init__(self, seconds: int = 0, cycles: Count = Count(0)):
337 self._seconds = seconds
338 self._cycles = cycles
339
340 @property
341 def seconds(self) -> int:
342 return self._seconds
343
344 @property
345 def cycles(self) -> Count:
346 return self._cycles
347
348
349 _OptUuid = Optional[uuidp.UUID]
350
351
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):
356 self._name = name
357 self._frequency = frequency
358 self._uuid = uuid
359 self._description = description
360 self._precision = precision
361 self._offset = ClockTypeOffset()
362
363 if offset is not None:
364 self._offset = offset
365
366 self._origin_is_unix_epoch = origin_is_unix_epoch
367
368 @property
369 def name(self) -> str:
370 return self._name
371
372 @property
373 def frequency(self) -> Count:
374 return self._frequency
375
376 @property
377 def uuid(self) -> _OptUuid:
378 return self._uuid
379
380 @property
381 def description(self) -> _OptStr:
382 return self._description
383
384 @property
385 def precision(self) -> Count:
386 return self._precision
387
388 @property
389 def offset(self) -> ClockTypeOffset:
390 return self._offset
391
392 @property
393 def origin_is_unix_epoch(self) -> bool:
394 return self._origin_is_unix_epoch
395
396
397 DEFAULT_FIELD_TYPE = 'default'
398 _DefaultableUIntFt = Union[str, UnsignedIntegerFieldType]
399 _OptDefaultableUIntFt = Optional[_DefaultableUIntFt]
400 _OptUIntFt = Optional[UnsignedIntegerFieldType]
401
402
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)
412
413 return typing.cast(_OptUIntFt, user_ft)
414
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)
420
421 @property
422 def total_size_field_type(self) -> _OptUIntFt:
423 return self._total_size_field_type
424
425 @property
426 def content_size_field_type(self) -> _OptUIntFt:
427 return self._content_size_field_type
428
429 @property
430 def beginning_time_field_type(self) -> _OptUIntFt:
431 return self._beginning_time_field_type
432
433 @property
434 def end_time_field_type(self) -> _OptUIntFt:
435 return self._end_time_field_type
436
437 @property
438 def discarded_events_counter_field_type(self) -> _OptUIntFt:
439 return self._discarded_events_counter_field_type
440
441
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)
448
449 return typing.cast(_OptUIntFt, user_ft)
450
451 self._type_id_field_type = get_ft(type_id_field_type)
452 self._time_field_type = get_ft(time_field_type)
453
454 @property
455 def type_id_field_type(self) -> _OptUIntFt:
456 return self._type_id_field_type
457
458 @property
459 def time_field_type(self) -> _OptUIntFt:
460 return self._time_field_type
461
462
463 class StreamTypeFeatures:
464 def __init__(self, packet_features: Optional[StreamTypePacketFeatures] = None,
465 event_features: Optional[StreamTypeEventFeatures] = None):
466 self._packet_features = StreamTypePacketFeatures()
467
468 if packet_features is not None:
469 self._packet_features = packet_features
470
471 self._event_features = StreamTypeEventFeatures()
472
473 if event_features is not None:
474 self._event_features = event_features
475
476 @property
477 def packet_features(self) -> StreamTypePacketFeatures:
478 return self._packet_features
479
480 @property
481 def event_features(self) -> StreamTypeEventFeatures:
482 return self._event_features
483
484
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
492 self._name = name
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)
496
497 # assign unique IDs
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)
501
502 self._set_features(features)
503 self._packet_context_field_type_extra_members = StructureFieldTypeMembers({})
504
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)
507
508 self._set_pkt_ctx_ft()
509 self._set_ev_header_ft()
510
511 def _set_features(self, features: Optional[StreamTypeFeatures]):
512 if features is not None:
513 self._features = features
514 return None
515
516 ev_time_ft = None
517 pkt_beginning_time_ft = None
518 pkt_end_time_ft = None
519
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
526
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))
530
531 def _set_ft_mapped_clk_type_name(self, ft: Optional[UnsignedIntegerFieldType]):
532 if ft is None:
533 return
534
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
538
539 def _set_pkt_ctx_ft(self):
540 members = None
541
542 def add_member_if_exists(name: str, ft: _FieldType, set_mapped_clk_type_name: bool = False):
543 nonlocal members
544
545 if ft is not None:
546 if set_mapped_clk_type_name:
547 self._set_ft_mapped_clk_type_name(typing.cast(UnsignedIntegerFieldType, ft))
548
549 members[name] = StructureFieldTypeMember(ft)
550
551 members = collections.OrderedDict([
552 (
553 'packet_size',
554 StructureFieldTypeMember(self._features.packet_features.total_size_field_type)
555 ),
556 (
557 'content_size',
558 StructureFieldTypeMember(self._features.packet_features.content_size_field_type)
559 )
560 ])
561
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,
565 True)
566 add_member_if_exists('events_discarded',
567 self._features.packet_features.discarded_events_counter_field_type)
568
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
573
574 self._pkt_ctx_ft = StructureFieldType(8, members)
575
576 def _set_ev_header_ft(self):
577 members = collections.OrderedDict()
578
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)
581
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)
586
587 self._ev_header_ft = StructureFieldType(8, members)
588
589 @property
590 def id(self) -> Optional[Id]:
591 return self._id
592
593 @property
594 def name(self) -> str:
595 return self._name
596
597 @property
598 def default_clock_type(self) -> Optional[ClockType]:
599 return self._default_clock_type
600
601 @property
602 def features(self) -> StreamTypeFeatures:
603 return self._features
604
605 @property
606 def packet_context_field_type_extra_members(self) -> StructureFieldTypeMembers:
607 return self._packet_context_field_type_extra_members
608
609 @property
610 def event_common_context_field_type(self) -> _OptStructFt:
611 return self._event_common_context_field_type
612
613 @property
614 def event_types(self) -> FrozenSet[EventType]:
615 return self._event_types
616
617
618 _OptUuidFt = Optional[Union[str, StaticArrayFieldType]]
619
620
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:
627 return default_ft
628
629 return typing.cast(_OptFt, user_ft)
630
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)))
638
639 @property
640 def magic_field_type(self) -> _OptUIntFt:
641 return self._magic_field_type
642
643 @property
644 def uuid_field_type(self) -> Optional[StaticArrayFieldType]:
645 return self._uuid_field_type
646
647 @property
648 def stream_type_id_field_type(self) -> _OptUIntFt:
649 return self._stream_type_id_field_type
650
651
652 class TraceType:
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)
657
658 # assign unique IDs
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)
662
663 self._uuid = uuid
664 self._set_features(features)
665 self._set_pkt_header_ft()
666 self._set_fts_effective_byte_order()
667
668 def _set_features(self, features: Optional[TraceTypeFeatures]):
669 if features is not None:
670 self._features = features
671 return
672
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)
676
677 def _set_pkt_header_ft(self):
678 members = collections.OrderedDict()
679
680 def add_member_if_exists(name: str, ft: _OptFt):
681 nonlocal members
682
683 if ft is not None:
684 members[name] = StructureFieldTypeMember(ft)
685
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)
690
691 def _set_fts_effective_byte_order(self):
692 def set_ft_effective_byte_order(ft: _OptFt):
693 if ft is None:
694 return
695
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)
705
706 # packet header field type
707 set_ft_effective_byte_order(self._pkt_header_ft)
708
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)
714
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)
719
720 @property
721 def default_byte_order(self) -> ByteOrder:
722 return self._default_byte_order
723
724 @property
725 def uuid(self) -> _OptUuid:
726 return self._uuid
727
728 @property
729 def stream_types(self) -> FrozenSet[StreamType]:
730 return self._stream_types
731
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
736
737 return None
738
739 @property
740 def features(self) -> TraceTypeFeatures:
741 return self._features
742
743
744 _EnvEntry = Union[str, int]
745 _EnvEntries = Mapping[str, _EnvEntry]
746
747
748 class TraceEnvironment(collections.abc.Mapping):
749 def __init__(self, environment: _EnvEntries):
750 self._env = {name: value for name, value in environment.items()}
751
752 def __getitem__(self, key: str) -> _EnvEntry:
753 return self._env[key]
754
755 def __iter__(self) -> Iterator[str]:
756 return iter(self._env)
757
758 def __len__(self) -> int:
759 return len(self._env)
760
761
762 class Trace:
763 def __init__(self, type: TraceType, environment: Optional[_EnvEntries] = None):
764 self._type = type
765 self._set_env(environment)
766
767 def _set_env(self, environment: Optional[_EnvEntries]):
768 init_env = collections.OrderedDict([
769 ('domain', 'bare'),
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())),
775 ])
776
777 if environment is None:
778 environment = {}
779
780 init_env.update(environment)
781 self._env = TraceEnvironment(typing.cast(_EnvEntries, init_env))
782
783 @property
784 def type(self) -> TraceType:
785 return self._type
786
787 @property
788 def environment(self) -> TraceEnvironment:
789 return self._env
790
791
792 _ClkTypeCTypes = Mapping[ClockType, str]
793
794
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()}
798
799 def __getitem__(self, key: ClockType) -> str:
800 return self._c_types[key]
801
802 def __iter__(self) -> Iterator[ClockType]:
803 return iter(self._c_types)
804
805 def __len__(self) -> int:
806 return len(self._c_types)
807
808
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
814
815 @property
816 def identifier_prefix_definition(self) -> bool:
817 return self._identifier_prefix_definition
818
819 @property
820 def default_stream_type_name_definition(self) -> bool:
821 return self._default_stream_type_name_definition
822
823
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
832
833 self._header_options = ConfigurationCodeGenerationHeaderOptions()
834
835 if header_options is not None:
836 self._header_options = header_options
837
838 self._clock_type_c_types = ClockTypeCTypes({})
839
840 if clock_type_c_types is not None:
841 self._clock_type_c_types = ClockTypeCTypes(clock_type_c_types)
842
843 @property
844 def identifier_prefix(self) -> str:
845 return self._identifier_prefix
846
847 @property
848 def file_name_prefix(self) -> str:
849 return self._file_name_prefix
850
851 @property
852 def default_stream_type(self) -> Optional[StreamType]:
853 return self._default_stream_type
854
855 @property
856 def header_options(self) -> ConfigurationCodeGenerationHeaderOptions:
857 return self._header_options
858
859 @property
860 def clock_type_c_types(self) -> ClockTypeCTypes:
861 return self._clock_type_c_types
862
863
864 class ConfigurationOptions:
865 def __init__(self,
866 code_generation_options: Optional[ConfigurationCodeGenerationOptions] = None):
867 self._code_generation_options = ConfigurationCodeGenerationOptions()
868
869 if code_generation_options is not None:
870 self._code_generation_options = code_generation_options
871
872 @property
873 def code_generation_options(self) -> ConfigurationCodeGenerationOptions:
874 return self._code_generation_options
875
876
877 class Configuration:
878 def __init__(self, trace: Trace, options: Optional[ConfigurationOptions] = None):
879 self._trace = trace
880 self._options = ConfigurationOptions()
881
882 if options is not None:
883 self._options = options
884
885 clk_type_c_types = self._options.code_generation_options.clock_type_c_types
886
887 for stream_type in trace.type.stream_types:
888 def_clk_type = stream_type.default_clock_type
889
890 if def_clk_type is None:
891 continue
892
893 if def_clk_type not in clk_type_c_types:
894 clk_type_c_types._c_types[def_clk_type] = 'uint32_t'
895
896 @property
897 def trace(self) -> Trace:
898 return self._trace
899
900 @property
901 def options(self) -> ConfigurationOptions:
902 return self._options
This page took 0.049166 seconds and 4 git commands to generate.