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, Callable
  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     @property
  47     def size_is_dynamic(self):
  48         return False
  49
  50
  51 class _BitArrayFieldType(_FieldType):
  52     def __init__(self, size: Count, alignment: Alignment = Alignment(1)):
  53         self._size = size
  54         self._alignment = alignment
  55
  56     @property
  57     def size(self) -> Count:
  58         return self._size
  59
  60     @property
  61     def alignment(self) -> Alignment:
  62         return self._alignment
  63
  64
  65 class DisplayBase(enum.Enum):
  66     BINARY = 2
  67     OCTAL = 8
  68     DECIMAL = 10
  69     HEXADECIMAL = 16
  70
  71
  72 class _IntegerFieldType(_BitArrayFieldType):
  73     def __init__(self, size: Count, alignment: Optional[Alignment] = None,
  74                  preferred_display_base: DisplayBase = DisplayBase.DECIMAL):
  75         if alignment is None:
  76             alignment = Alignment(8 if size % 8 == 0 else 1)
  77
  78         super().__init__(size, alignment)
  79         self._preferred_display_base = preferred_display_base
  80
  81     @property
  82     def preferred_display_base(self) -> DisplayBase:
  83         return self._preferred_display_base
  84
  85
  86 class UnsignedIntegerFieldType(_IntegerFieldType):
  87     def __init__(self, *args, **kwargs):
  88         super().__init__(*args, **kwargs)
  89         self._mapped_clk_type_name = None
  90         self._is_len = False
  91
  92
  93 class SignedIntegerFieldType(_IntegerFieldType):
  94     pass
  95
  96
  97 class EnumerationFieldTypeMappingRange:
  98     def __init__(self, lower: int, upper: int):
  99         self._lower = lower
 100         self._upper = upper
 101
 102     @property
 103     def lower(self) -> int:
 104         return self._lower
 105
 106     @property
 107     def upper(self) -> int:
 108         return self._upper
 109
 110     def __eq__(self, other: Any) -> bool:
 111         if type(other) is not type(self):
 112             return False
 113
 114         return (self._lower, self._upper) == (other._lower, other._upper)
 115
 116     def __hash__(self) -> int:
 117         return hash((self._lower, self._upper))
 118
 119     def contains(self, value: int) -> bool:
 120         return self._lower <= value <= self._upper
 121
 122
 123 class EnumerationFieldTypeMapping:
 124     def __init__(self, ranges: Set[EnumerationFieldTypeMappingRange]):
 125         self._ranges = frozenset(ranges)
 126
 127     @property
 128     def ranges(self) -> FrozenSet[EnumerationFieldTypeMappingRange]:
 129         return self._ranges
 130
 131     def ranges_contain_value(self, value: int) -> bool:
 132         return any([rg.contains(value) for rg in self._ranges])
 133
 134
 135 _EnumFtMappings = Mapping[str, EnumerationFieldTypeMapping]
 136
 137
 138 class EnumerationFieldTypeMappings(collections.abc.Mapping):
 139     def __init__(self, mappings: _EnumFtMappings):
 140         self._mappings = {label: mapping for label, mapping in mappings.items()}
 141
 142     def __getitem__(self, key: str) -> EnumerationFieldTypeMapping:
 143         return self._mappings[key]
 144
 145     def __iter__(self) -> Iterator[str]:
 146         return iter(self._mappings)
 147
 148     def __len__(self) -> int:
 149         return len(self._mappings)
 150
 151
 152 class _EnumerationFieldType(_IntegerFieldType):
 153     def __init__(self, size: Count, alignment: Optional[Alignment] = None,
 154                  preferred_display_base: DisplayBase = DisplayBase.DECIMAL,
 155                  mappings: Optional[_EnumFtMappings] = None):
 156         super().__init__(size, alignment, preferred_display_base)
 157         self._mappings = EnumerationFieldTypeMappings({})
 158
 159         if mappings is not None:
 160             self._mappings = EnumerationFieldTypeMappings(mappings)
 161
 162     @property
 163     def mappings(self) -> EnumerationFieldTypeMappings:
 164         return self._mappings
 165
 166     def labels_for_value(self, value: int) -> Set[str]:
 167         labels = set()
 168
 169         for label, mapping in self._mappings.items():
 170             if mapping.ranges_contain_value(value):
 171                 labels.add(label)
 172
 173         return labels
 174
 175
 176 class UnsignedEnumerationFieldType(_EnumerationFieldType, UnsignedIntegerFieldType):
 177     pass
 178
 179
 180 class SignedEnumerationFieldType(_EnumerationFieldType, SignedIntegerFieldType):
 181     pass
 182
 183
 184 class RealFieldType(_BitArrayFieldType):
 185     pass
 186
 187
 188 class StringFieldType(_FieldType):
 189     @property
 190     def alignment(self) -> Alignment:
 191         return Alignment(8)
 192
 193     @property
 194     def size_is_dynamic(self):
 195         return True
 196
 197
 198 class _ArrayFieldType(_FieldType):
 199     def __init__(self, element_field_type: _FieldType):
 200         self._element_field_type = element_field_type
 201
 202     @property
 203     def element_field_type(self) -> _FieldType:
 204         return self._element_field_type
 205
 206     @property
 207     def alignment(self) -> Alignment:
 208         return self._element_field_type.alignment
 209
 210     @property
 211     def size_is_dynamic(self):
 212         return self._element_field_type.size_is_dynamic
 213
 214
 215 class StaticArrayFieldType(_ArrayFieldType):
 216     def __init__(self, length: Count, element_field_type: _FieldType):
 217         super().__init__(element_field_type)
 218         self._length = length
 219
 220     @property
 221     def length(self) -> Count:
 222         return self._length
 223
 224
 225 class StructureFieldTypeMember:
 226     def __init__(self, field_type: _FieldType):
 227         self._field_type = field_type
 228
 229     @property
 230     def field_type(self) -> _FieldType:
 231         return self._field_type
 232
 233
 234 _StructFtMembers = Mapping[str, StructureFieldTypeMember]
 235
 236
 237 class StructureFieldTypeMembers(collections.abc.Mapping):
 238     def __init__(self, members: _StructFtMembers):
 239         self._members = collections.OrderedDict()
 240
 241         for name, member in members.items():
 242             assert type(member) is StructureFieldTypeMember
 243             self._members[name] = member
 244
 245     def __getitem__(self, key: str) -> StructureFieldTypeMember:
 246         return self._members[key]
 247
 248     def __iter__(self) -> Iterator[str]:
 249         return iter(self._members)
 250
 251     def __len__(self) -> int:
 252         return len(self._members)
 253
 254
 255 class StructureFieldType(_FieldType):
 256     def __init__(self, minimum_alignment: Alignment = Alignment(1),
 257                  members: Optional[_StructFtMembers] = None):
 258         self._minimum_alignment = minimum_alignment
 259         self._members = StructureFieldTypeMembers({})
 260
 261         if members is not None:
 262             self._members = StructureFieldTypeMembers(members)
 263
 264         self._set_alignment()
 265
 266     def _set_alignment(self):
 267         self._alignment: Alignment = self._minimum_alignment
 268
 269         for member in self._members.values():
 270             if member.field_type.alignment > self._alignment:
 271                 self._alignment = member.field_type.alignment
 272
 273     @property
 274     def minimum_alignment(self) -> Alignment:
 275         return self._minimum_alignment
 276
 277     @property
 278     def alignment(self) -> Alignment:
 279         return self._alignment
 280
 281     @property
 282     def size_is_dynamic(self):
 283         return any([member.field_type.size_is_dynamic for member in self.members.values()])
 284
 285     @property
 286     def members(self) -> StructureFieldTypeMembers:
 287         return self._members
 288
 289
 290 class _UniqueByName:
 291     _name: str
 292
 293     def __eq__(self, other: Any) -> bool:
 294         if type(other) is not type(self):
 295             return False
 296
 297         return self._name == other._name
 298
 299     def __lt__(self, other: '_UniqueByName'):
 300         assert type(self) is type(other)
 301         return self._name < other._name
 302
 303     def __hash__(self) -> int:
 304         return hash(self._name)
 305
 306
 307 _OptFt = Optional[_FieldType]
 308 _OptStructFt = Optional[StructureFieldType]
 309 LogLevel = typing.NewType('LogLevel', int)
 310
 311
 312 class EventType(_UniqueByName):
 313     def __init__(self, name: str, log_level: Optional[LogLevel] = None,
 314                  specific_context_field_type: _OptStructFt = None, payload_field_type: _OptStructFt = None):
 315         self._id: Optional[Id] = None
 316         self._name = name
 317         self._log_level = log_level
 318         self._specific_context_field_type = specific_context_field_type
 319         self._payload_field_type = payload_field_type
 320
 321     @property
 322     def id(self) -> Optional[Id]:
 323         return self._id
 324
 325     @property
 326     def name(self) -> str:
 327         return self._name
 328
 329     @property
 330     def log_level(self) -> Optional[LogLevel]:
 331         return self._log_level
 332
 333     @property
 334     def specific_context_field_type(self) -> _OptStructFt:
 335         return self._specific_context_field_type
 336
 337     @property
 338     def payload_field_type(self) -> _OptStructFt:
 339         return self._payload_field_type
 340
 341
 342 class ClockTypeOffset:
 343     def __init__(self, seconds: int = 0, cycles: Count = Count(0)):
 344         self._seconds = seconds
 345         self._cycles = cycles
 346
 347     @property
 348     def seconds(self) -> int:
 349         return self._seconds
 350
 351     @property
 352     def cycles(self) -> Count:
 353         return self._cycles
 354
 355
 356 _OptUuid = Optional[uuidp.UUID]
 357
 358
 359 class ClockType(_UniqueByName):
 360     def __init__(self, name: str, frequency: Count = Count(int(1e9)), uuid: _OptUuid = None,
 361                  description: _OptStr = None, precision: Count = Count(0),
 362                  offset: Optional[ClockTypeOffset] = None, origin_is_unix_epoch: bool = False):
 363         self._name = name
 364         self._frequency = frequency
 365         self._uuid = uuid
 366         self._description = description
 367         self._precision = precision
 368         self._offset = ClockTypeOffset()
 369
 370         if offset is not None:
 371             self._offset = offset
 372
 373         self._origin_is_unix_epoch = origin_is_unix_epoch
 374
 375     @property
 376     def name(self) -> str:
 377         return self._name
 378
 379     @property
 380     def frequency(self) -> Count:
 381         return self._frequency
 382
 383     @property
 384     def uuid(self) -> _OptUuid:
 385         return self._uuid
 386
 387     @property
 388     def description(self) -> _OptStr:
 389         return self._description
 390
 391     @property
 392     def precision(self) -> Count:
 393         return self._precision
 394
 395     @property
 396     def offset(self) -> ClockTypeOffset:
 397         return self._offset
 398
 399     @property
 400     def origin_is_unix_epoch(self) -> bool:
 401         return self._origin_is_unix_epoch
 402
 403
 404 DEFAULT_FIELD_TYPE = 'default'
 405 _DefaultableUIntFt = Union[str, UnsignedIntegerFieldType]
 406 _OptDefaultableUIntFt = Optional[_DefaultableUIntFt]
 407 _OptUIntFt = Optional[UnsignedIntegerFieldType]
 408
 409
 410 class StreamTypePacketFeatures:
 411     def __init__(self, total_size_field_type: _DefaultableUIntFt = DEFAULT_FIELD_TYPE,
 412                  content_size_field_type: _DefaultableUIntFt = DEFAULT_FIELD_TYPE,
 413                  beginning_time_field_type: _OptDefaultableUIntFt = None,
 414                  end_time_field_type: _OptDefaultableUIntFt = None,
 415                  discarded_events_counter_field_type: _OptDefaultableUIntFt = None):
 416         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 417             if user_ft == DEFAULT_FIELD_TYPE:
 418                 return UnsignedIntegerFieldType(64)
 419
 420             return typing.cast(_OptUIntFt, user_ft)
 421
 422         self._total_size_field_type = get_ft(total_size_field_type)
 423         self._content_size_field_type = get_ft(content_size_field_type)
 424         self._beginning_time_field_type = get_ft(beginning_time_field_type)
 425         self._end_time_field_type = get_ft(end_time_field_type)
 426         self._discarded_events_counter_field_type = get_ft(discarded_events_counter_field_type)
 427
 428     @property
 429     def total_size_field_type(self) -> _OptUIntFt:
 430         return self._total_size_field_type
 431
 432     @property
 433     def content_size_field_type(self) -> _OptUIntFt:
 434         return self._content_size_field_type
 435
 436     @property
 437     def beginning_time_field_type(self) -> _OptUIntFt:
 438         return self._beginning_time_field_type
 439
 440     @property
 441     def end_time_field_type(self) -> _OptUIntFt:
 442         return self._end_time_field_type
 443
 444     @property
 445     def discarded_events_counter_field_type(self) -> _OptUIntFt:
 446         return self._discarded_events_counter_field_type
 447
 448
 449 class StreamTypeEventFeatures:
 450     def __init__(self, type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 451                  time_field_type: _OptDefaultableUIntFt = None):
 452         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 453             if user_ft == DEFAULT_FIELD_TYPE:
 454                 return UnsignedIntegerFieldType(64)
 455
 456             return typing.cast(_OptUIntFt, user_ft)
 457
 458         self._type_id_field_type = get_ft(type_id_field_type)
 459         self._time_field_type = get_ft(time_field_type)
 460
 461     @property
 462     def type_id_field_type(self) -> _OptUIntFt:
 463         return self._type_id_field_type
 464
 465     @property
 466     def time_field_type(self) -> _OptUIntFt:
 467         return self._time_field_type
 468
 469
 470 class StreamTypeFeatures:
 471     def __init__(self, packet_features: Optional[StreamTypePacketFeatures] = None,
 472                  event_features: Optional[StreamTypeEventFeatures] = None):
 473         if packet_features is None:
 474             self._packet_features = StreamTypePacketFeatures()
 475         else:
 476             self._packet_features = packet_features
 477
 478         if event_features is None:
 479             self._event_features = StreamTypeEventFeatures()
 480         else:
 481             self._event_features = event_features
 482
 483     @property
 484     def packet_features(self) -> StreamTypePacketFeatures:
 485         return self._packet_features
 486
 487     @property
 488     def event_features(self) -> StreamTypeEventFeatures:
 489         return self._event_features
 490
 491
 492 class StreamType(_UniqueByName):
 493     def __init__(self, name: str, event_types: Set[EventType],
 494                  default_clock_type: Optional[ClockType] = None,
 495                  features: Optional[StreamTypeFeatures] = None,
 496                  packet_context_field_type_extra_members: Optional[_StructFtMembers] = None,
 497                  event_common_context_field_type: _OptStructFt = None):
 498         self._id: Optional[Id] = None
 499         self._name = name
 500         self._default_clock_type = default_clock_type
 501         self._event_common_context_field_type = event_common_context_field_type
 502         self._event_types = frozenset(event_types)
 503
 504         # assign unique IDs
 505         for index, ev_type in enumerate(sorted(self._event_types, key=lambda evt: evt.name)):
 506             assert ev_type._id is None
 507             ev_type._id = Id(index)
 508
 509         self._set_features(features)
 510         self._packet_context_field_type_extra_members = StructureFieldTypeMembers({})
 511
 512         if packet_context_field_type_extra_members is not None:
 513             self._packet_context_field_type_extra_members = StructureFieldTypeMembers(packet_context_field_type_extra_members)
 514
 515         self._set_pkt_ctx_ft()
 516         self._set_ev_header_ft()
 517
 518     def _set_features(self, features: Optional[StreamTypeFeatures]):
 519         if features is not None:
 520             self._features = features
 521             return None
 522
 523         ev_time_ft = None
 524         pkt_beginning_time_ft = None
 525         pkt_end_time_ft = None
 526
 527         if self._default_clock_type is not None:
 528             # Automatic time field types because the stream type has a
 529             # default clock type.
 530             ev_time_ft = DEFAULT_FIELD_TYPE
 531             pkt_beginning_time_ft = DEFAULT_FIELD_TYPE
 532             pkt_end_time_ft = DEFAULT_FIELD_TYPE
 533
 534         self._features = StreamTypeFeatures(StreamTypePacketFeatures(beginning_time_field_type=pkt_beginning_time_ft,
 535                                                                      end_time_field_type=pkt_end_time_ft),
 536                                             StreamTypeEventFeatures(time_field_type=ev_time_ft))
 537
 538     def _set_ft_mapped_clk_type_name(self, ft: Optional[UnsignedIntegerFieldType]):
 539         if ft is None:
 540             return
 541
 542         if self._default_clock_type is not None:
 543             assert isinstance(ft, UnsignedIntegerFieldType)
 544             ft._mapped_clk_type_name = self._default_clock_type.name
 545
 546     def _set_pkt_ctx_ft(self):
 547         members = None
 548
 549         def add_member_if_exists(name: str, ft: _FieldType, set_mapped_clk_type_name: bool = False):
 550             nonlocal members
 551
 552             if ft is not None:
 553                 if set_mapped_clk_type_name:
 554                     self._set_ft_mapped_clk_type_name(typing.cast(UnsignedIntegerFieldType, ft))
 555
 556                 members[name] = StructureFieldTypeMember(ft)
 557
 558         members = collections.OrderedDict([
 559             (
 560                 'packet_size',
 561                 StructureFieldTypeMember(self._features.packet_features.total_size_field_type)
 562             ),
 563             (
 564                 'content_size',
 565                 StructureFieldTypeMember(self._features.packet_features.content_size_field_type)
 566             )
 567         ])
 568
 569         add_member_if_exists('timestamp_begin',
 570                              self._features.packet_features.beginning_time_field_type, True)
 571         add_member_if_exists('timestamp_end', self._features.packet_features.end_time_field_type,
 572                              True)
 573         add_member_if_exists('events_discarded',
 574                              self._features.packet_features.discarded_events_counter_field_type)
 575
 576         if self._packet_context_field_type_extra_members is not None:
 577             for name, field_type in self._packet_context_field_type_extra_members.items():
 578                 assert name not in members
 579                 members[name] = field_type
 580
 581         self._pkt_ctx_ft = StructureFieldType(8, members)
 582
 583     def _set_ev_header_ft(self):
 584         members = collections.OrderedDict()
 585
 586         if self._features.event_features.type_id_field_type is not None:
 587             members['id'] = StructureFieldTypeMember(self._features.event_features.type_id_field_type)
 588
 589         if self._features.event_features.time_field_type is not None:
 590             ft = self._features.event_features.time_field_type
 591             self._set_ft_mapped_clk_type_name(ft)
 592             members['timestamp'] = StructureFieldTypeMember(ft)
 593
 594         self._ev_header_ft = StructureFieldType(8, members)
 595
 596     @property
 597     def id(self) -> Optional[Id]:
 598         return self._id
 599
 600     @property
 601     def name(self) -> str:
 602         return self._name
 603
 604     @property
 605     def default_clock_type(self) -> Optional[ClockType]:
 606         return self._default_clock_type
 607
 608     @property
 609     def features(self) -> StreamTypeFeatures:
 610         return self._features
 611
 612     @property
 613     def packet_context_field_type_extra_members(self) -> StructureFieldTypeMembers:
 614         return self._packet_context_field_type_extra_members
 615
 616     @property
 617     def event_common_context_field_type(self) -> _OptStructFt:
 618         return self._event_common_context_field_type
 619
 620     @property
 621     def event_types(self) -> FrozenSet[EventType]:
 622         return self._event_types
 623
 624
 625 _OptUuidFt = Optional[Union[str, StaticArrayFieldType]]
 626
 627
 628 class TraceTypeFeatures:
 629     def __init__(self, magic_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 630                  uuid_field_type: _OptUuidFt = None,
 631                  stream_type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE):
 632         def get_field_type(user_ft: Optional[Union[str, _FieldType]],
 633                            create_default_ft: Callable[[], _FieldType]) -> _OptFt:
 634             if user_ft == DEFAULT_FIELD_TYPE:
 635                 return create_default_ft()
 636
 637             return typing.cast(_OptFt, user_ft)
 638
 639         def create_default_magic_ft():
 640             return UnsignedIntegerFieldType(32)
 641
 642         def create_default_uuid_ft():
 643             return StaticArrayFieldType(Count(16), UnsignedIntegerFieldType(8))
 644
 645         def create_default_stream_type_id_ft():
 646             return UnsignedIntegerFieldType(64)
 647
 648         self._magic_field_type = typing.cast(_OptUIntFt, get_field_type(magic_field_type, create_default_magic_ft))
 649         self._uuid_field_type = typing.cast(Optional[StaticArrayFieldType],
 650                                             get_field_type(uuid_field_type, create_default_uuid_ft))
 651         self._stream_type_id_field_type = typing.cast(_OptUIntFt,
 652                                                       get_field_type(stream_type_id_field_type,
 653                                                                      create_default_stream_type_id_ft))
 654
 655     @property
 656     def magic_field_type(self) -> _OptUIntFt:
 657         return self._magic_field_type
 658
 659     @property
 660     def uuid_field_type(self) -> Optional[StaticArrayFieldType]:
 661         return self._uuid_field_type
 662
 663     @property
 664     def stream_type_id_field_type(self) -> _OptUIntFt:
 665         return self._stream_type_id_field_type
 666
 667
 668 class TraceType:
 669     def __init__(self, stream_types: Set[StreamType], uuid: _OptUuid = None,
 670                  features: Optional[TraceTypeFeatures] = None):
 671         self._stream_types = frozenset(stream_types)
 672
 673         # assign unique IDs
 674         for index, stream_type in enumerate(sorted(self._stream_types, key=lambda st: st.name)):
 675             assert stream_type._id is None
 676             stream_type._id = Id(index)
 677
 678         self._uuid = uuid
 679         self._set_features(features)
 680         self._set_pkt_header_ft()
 681
 682     def _set_features(self, features: Optional[TraceTypeFeatures]):
 683         if features is not None:
 684             self._features = features
 685             return
 686
 687         # automatic UUID field type because the trace type has a UUID
 688         uuid_ft = None if self._uuid is None else DEFAULT_FIELD_TYPE
 689         self._features = TraceTypeFeatures(uuid_field_type=uuid_ft)
 690
 691     def _set_pkt_header_ft(self):
 692         members = collections.OrderedDict()
 693
 694         def add_member_if_exists(name: str, ft: _OptFt):
 695             nonlocal members
 696
 697             if ft is not None:
 698                 members[name] = StructureFieldTypeMember(ft)
 699
 700         add_member_if_exists('magic', self._features.magic_field_type)
 701         add_member_if_exists('uuid', self._features.uuid_field_type)
 702         add_member_if_exists('stream_id', self._features.stream_type_id_field_type)
 703         self._pkt_header_ft = StructureFieldType(8, members)
 704
 705     @property
 706     def uuid(self) -> _OptUuid:
 707         return self._uuid
 708
 709     @property
 710     def stream_types(self) -> FrozenSet[StreamType]:
 711         return self._stream_types
 712
 713     def stream_type(self, name: str) -> Optional[StreamType]:
 714         for cand_stream_type in self._stream_types:
 715             if cand_stream_type.name == name:
 716                 return cand_stream_type
 717
 718         return None
 719
 720     @property
 721     def features(self) -> TraceTypeFeatures:
 722         return self._features
 723
 724     @property
 725     def clock_types(self) -> Set[ClockType]:
 726         clk_types = set()
 727
 728         for stream_type in self._stream_types:
 729             if stream_type.default_clock_type is not None:
 730                 clk_types.add(stream_type.default_clock_type)
 731
 732         return clk_types
 733
 734
 735 _EnvEntry = Union[str, int]
 736 _EnvEntries = Mapping[str, _EnvEntry]
 737
 738
 739 class TraceEnvironment(collections.abc.Mapping):
 740     def __init__(self, environment: _EnvEntries):
 741         self._env = {name: value for name, value in environment.items()}
 742
 743     def __getitem__(self, key: str) -> _EnvEntry:
 744         return self._env[key]
 745
 746     def __iter__(self) -> Iterator[str]:
 747         return iter(self._env)
 748
 749     def __len__(self) -> int:
 750         return len(self._env)
 751
 752
 753 class Trace:
 754     def __init__(self, type: TraceType, environment: Optional[_EnvEntries] = None):
 755         self._type = type
 756         self._set_env(environment)
 757
 758     def _set_env(self, environment: Optional[_EnvEntries]):
 759         init_env = collections.OrderedDict([
 760             ('domain', 'bare'),
 761             ('tracer_name', 'barectf'),
 762             ('tracer_major', barectf_version.__major_version__),
 763             ('tracer_minor', barectf_version.__minor_version__),
 764             ('tracer_patch', barectf_version.__patch_version__),
 765             ('barectf_gen_date', str(datetime.datetime.now().isoformat())),
 766         ])
 767
 768         if environment is None:
 769             environment = {}
 770
 771         init_env.update(environment)
 772         self._env = TraceEnvironment(typing.cast(_EnvEntries, init_env))
 773
 774     @property
 775     def type(self) -> TraceType:
 776         return self._type
 777
 778     @property
 779     def environment(self) -> TraceEnvironment:
 780         return self._env
 781
 782
 783 _ClkTypeCTypes = Mapping[ClockType, str]
 784
 785
 786 class ClockTypeCTypes(collections.abc.Mapping):
 787     def __init__(self, c_types: _ClkTypeCTypes):
 788         self._c_types = {clk_type: c_type for clk_type, c_type in c_types.items()}
 789
 790     def __getitem__(self, key: ClockType) -> str:
 791         return self._c_types[key]
 792
 793     def __iter__(self) -> Iterator[ClockType]:
 794         return iter(self._c_types)
 795
 796     def __len__(self) -> int:
 797         return len(self._c_types)
 798
 799
 800 class ConfigurationCodeGenerationHeaderOptions:
 801     def __init__(self, identifier_prefix_definition: bool = False,
 802                  default_stream_type_name_definition: bool = False):
 803         self._identifier_prefix_definition = identifier_prefix_definition
 804         self._default_stream_type_name_definition = default_stream_type_name_definition
 805
 806     @property
 807     def identifier_prefix_definition(self) -> bool:
 808         return self._identifier_prefix_definition
 809
 810     @property
 811     def default_stream_type_name_definition(self) -> bool:
 812         return self._default_stream_type_name_definition
 813
 814
 815 class ConfigurationCodeGenerationOptions:
 816     def __init__(self, identifier_prefix: str = 'barectf_', file_name_prefix: str = 'barectf',
 817                  default_stream_type: Optional[StreamType] = None,
 818                  header_options: Optional[ConfigurationCodeGenerationHeaderOptions] = None,
 819                  clock_type_c_types: Optional[_ClkTypeCTypes] = None):
 820         self._identifier_prefix = identifier_prefix
 821         self._file_name_prefix = file_name_prefix
 822         self._default_stream_type = default_stream_type
 823
 824         self._header_options = ConfigurationCodeGenerationHeaderOptions()
 825
 826         if header_options is not None:
 827             self._header_options = header_options
 828
 829         self._clock_type_c_types = ClockTypeCTypes({})
 830
 831         if clock_type_c_types is not None:
 832             self._clock_type_c_types = ClockTypeCTypes(clock_type_c_types)
 833
 834     @property
 835     def identifier_prefix(self) -> str:
 836         return self._identifier_prefix
 837
 838     @property
 839     def file_name_prefix(self) -> str:
 840         return self._file_name_prefix
 841
 842     @property
 843     def default_stream_type(self) -> Optional[StreamType]:
 844         return self._default_stream_type
 845
 846     @property
 847     def header_options(self) -> ConfigurationCodeGenerationHeaderOptions:
 848         return self._header_options
 849
 850     @property
 851     def clock_type_c_types(self) -> ClockTypeCTypes:
 852         return self._clock_type_c_types
 853
 854
 855 class ConfigurationOptions:
 856     def __init__(self,
 857                  code_generation_options: Optional[ConfigurationCodeGenerationOptions] = None):
 858         self._code_generation_options = ConfigurationCodeGenerationOptions()
 859
 860         if code_generation_options is not None:
 861             self._code_generation_options = code_generation_options
 862
 863     @property
 864     def code_generation_options(self) -> ConfigurationCodeGenerationOptions:
 865         return self._code_generation_options
 866
 867
 868 class Configuration:
 869     def __init__(self, trace: Trace, target_byte_order: ByteOrder,
 870                  options: Optional[ConfigurationOptions] = None):
 871         self._trace = trace
 872         self._options = ConfigurationOptions()
 873         self._target_byte_order = target_byte_order
 874
 875         if options is not None:
 876             self._options = options
 877
 878         clk_type_c_types = self._options.code_generation_options.clock_type_c_types
 879
 880         for stream_type in trace.type.stream_types:
 881             def_clk_type = stream_type.default_clock_type
 882
 883             if def_clk_type is None:
 884                 continue
 885
 886             if def_clk_type not in clk_type_c_types:
 887                 clk_type_c_types._c_types[def_clk_type] = 'uint32_t'
 888
 889     @property
 890     def trace(self) -> Trace:
 891         return self._trace
 892
 893     @property
 894     def target_byte_order(self):
 895         return self._target_byte_order
 896
 897     @property
 898     def options(self) -> ConfigurationOptions:
 899         return self._options