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, byte_order: ByteOrder, alignment: Alignment = Alignment(1)):
  53         self._size = size
  54         self._byte_order = byte_order
  55         self._alignment = alignment
  56
  57     @property
  58     def size(self) -> Count:
  59         return self._size
  60
  61     @property
  62     def byte_order(self) -> ByteOrder:
  63         return self._byte_order
  64
  65     @property
  66     def alignment(self) -> Alignment:
  67         return self._alignment
  68
  69
  70 class DisplayBase(enum.Enum):
  71     BINARY = 2
  72     OCTAL = 8
  73     DECIMAL = 10
  74     HEXADECIMAL = 16
  75
  76
  77 class _IntegerFieldType(_BitArrayFieldType):
  78     def __init__(self, size: Count, byte_order: ByteOrder, alignment: Optional[Alignment] = None,
  79                  preferred_display_base: DisplayBase = DisplayBase.DECIMAL):
  80         if alignment is None:
  81             alignment = Alignment(8 if size % 8 == 0 else 1)
  82
  83         super().__init__(size, byte_order, 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: ByteOrder, alignment: Optional[Alignment] = None,
 158                  preferred_display_base: DisplayBase = DisplayBase.DECIMAL,
 159                  mappings: Optional[_EnumFtMappings] = None):
 160         super().__init__(size, byte_order, alignment, preferred_display_base)
 161         self._mappings = EnumerationFieldTypeMappings({})
 162
 163         if mappings is not None:
 164             self._mappings = EnumerationFieldTypeMappings(mappings)
 165
 166     @property
 167     def mappings(self) -> EnumerationFieldTypeMappings:
 168         return self._mappings
 169
 170     def labels_for_value(self, value: int) -> Set[str]:
 171         labels = set()
 172
 173         for label, mapping in self._mappings.items():
 174             if mapping.ranges_contain_value(value):
 175                 labels.add(label)
 176
 177         return labels
 178
 179
 180 class UnsignedEnumerationFieldType(_EnumerationFieldType, UnsignedIntegerFieldType):
 181     pass
 182
 183
 184 class SignedEnumerationFieldType(_EnumerationFieldType, SignedIntegerFieldType):
 185     pass
 186
 187
 188 class RealFieldType(_BitArrayFieldType):
 189     pass
 190
 191
 192 class StringFieldType(_FieldType):
 193     @property
 194     def alignment(self) -> Alignment:
 195         return Alignment(8)
 196
 197     @property
 198     def size_is_dynamic(self):
 199         return True
 200
 201
 202 class _ArrayFieldType(_FieldType):
 203     def __init__(self, element_field_type: _FieldType):
 204         self._element_field_type = element_field_type
 205
 206     @property
 207     def element_field_type(self) -> _FieldType:
 208         return self._element_field_type
 209
 210     @property
 211     def alignment(self) -> Alignment:
 212         return self._element_field_type.alignment
 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                  default_byte_order: Optional[ByteOrder] = None):
 417         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 418             if user_ft == DEFAULT_FIELD_TYPE:
 419                 assert default_byte_order is not None
 420                 return UnsignedIntegerFieldType(64, typing.cast(ByteOrder, default_byte_order))
 421
 422             return typing.cast(_OptUIntFt, user_ft)
 423
 424         self._total_size_field_type = get_ft(total_size_field_type)
 425         self._content_size_field_type = get_ft(content_size_field_type)
 426         self._beginning_time_field_type = get_ft(beginning_time_field_type)
 427         self._end_time_field_type = get_ft(end_time_field_type)
 428         self._discarded_events_counter_field_type = get_ft(discarded_events_counter_field_type)
 429
 430     @property
 431     def total_size_field_type(self) -> _OptUIntFt:
 432         return self._total_size_field_type
 433
 434     @property
 435     def content_size_field_type(self) -> _OptUIntFt:
 436         return self._content_size_field_type
 437
 438     @property
 439     def beginning_time_field_type(self) -> _OptUIntFt:
 440         return self._beginning_time_field_type
 441
 442     @property
 443     def end_time_field_type(self) -> _OptUIntFt:
 444         return self._end_time_field_type
 445
 446     @property
 447     def discarded_events_counter_field_type(self) -> _OptUIntFt:
 448         return self._discarded_events_counter_field_type
 449
 450
 451 class StreamTypeEventFeatures:
 452     def __init__(self, type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 453                  time_field_type: _OptDefaultableUIntFt = None,
 454                  default_byte_order: Optional[ByteOrder] = None):
 455         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 456             if user_ft == DEFAULT_FIELD_TYPE:
 457                 assert default_byte_order is not None
 458                 return UnsignedIntegerFieldType(64, typing.cast(ByteOrder, default_byte_order))
 459
 460             return typing.cast(_OptUIntFt, user_ft)
 461
 462         self._type_id_field_type = get_ft(type_id_field_type)
 463         self._time_field_type = get_ft(time_field_type)
 464
 465     @property
 466     def type_id_field_type(self) -> _OptUIntFt:
 467         return self._type_id_field_type
 468
 469     @property
 470     def time_field_type(self) -> _OptUIntFt:
 471         return self._time_field_type
 472
 473
 474 class StreamTypeFeatures:
 475     def __init__(self, packet_features: Optional[StreamTypePacketFeatures] = None,
 476                  event_features: Optional[StreamTypeEventFeatures] = None,
 477                  default_byte_order: Optional[ByteOrder] = None):
 478         if packet_features is None:
 479             self._packet_features = StreamTypePacketFeatures(default_byte_order=default_byte_order)
 480         else:
 481             self._packet_features = packet_features
 482
 483         if event_features is None:
 484             self._event_features = StreamTypeEventFeatures(default_byte_order=default_byte_order)
 485         else:
 486             self._event_features = event_features
 487
 488     @property
 489     def packet_features(self) -> StreamTypePacketFeatures:
 490         return self._packet_features
 491
 492     @property
 493     def event_features(self) -> StreamTypeEventFeatures:
 494         return self._event_features
 495
 496
 497 class StreamType(_UniqueByName):
 498     def __init__(self, name: str, event_types: Set[EventType],
 499                  default_clock_type: Optional[ClockType] = None,
 500                  features: Optional[StreamTypeFeatures] = None,
 501                  default_feature_field_type_byte_order: Optional[ByteOrder] = None,
 502                  packet_context_field_type_extra_members: Optional[_StructFtMembers] = None,
 503                  event_common_context_field_type: _OptStructFt = None):
 504         self._id: Optional[Id] = None
 505         self._name = name
 506         self._default_clock_type = default_clock_type
 507         self._event_common_context_field_type = event_common_context_field_type
 508         self._event_types = frozenset(event_types)
 509
 510         # assign unique IDs
 511         for index, ev_type in enumerate(sorted(self._event_types, key=lambda evt: evt.name)):
 512             assert ev_type._id is None
 513             ev_type._id = Id(index)
 514
 515         self._set_features(features, default_feature_field_type_byte_order)
 516         self._packet_context_field_type_extra_members = StructureFieldTypeMembers({})
 517
 518         if packet_context_field_type_extra_members is not None:
 519             self._packet_context_field_type_extra_members = StructureFieldTypeMembers(packet_context_field_type_extra_members)
 520
 521         self._set_pkt_ctx_ft()
 522         self._set_ev_header_ft()
 523
 524     def _set_features(self, features: Optional[StreamTypeFeatures],
 525                       default_byte_order: Optional[ByteOrder]):
 526         if features is not None:
 527             self._features = features
 528             return None
 529
 530         ev_time_ft = None
 531         pkt_beginning_time_ft = None
 532         pkt_end_time_ft = None
 533
 534         if self._default_clock_type is not None:
 535             # Automatic time field types because the stream type has a
 536             # default clock type.
 537             ev_time_ft = DEFAULT_FIELD_TYPE
 538             pkt_beginning_time_ft = DEFAULT_FIELD_TYPE
 539             pkt_end_time_ft = DEFAULT_FIELD_TYPE
 540
 541         self._features = StreamTypeFeatures(StreamTypePacketFeatures(beginning_time_field_type=pkt_beginning_time_ft,
 542                                                                      end_time_field_type=pkt_end_time_ft,
 543                                                                      default_byte_order=default_byte_order),
 544                                             StreamTypeEventFeatures(time_field_type=ev_time_ft,
 545                                                                     default_byte_order=default_byte_order))
 546
 547     def _set_ft_mapped_clk_type_name(self, ft: Optional[UnsignedIntegerFieldType]):
 548         if ft is None:
 549             return
 550
 551         if self._default_clock_type is not None:
 552             assert isinstance(ft, UnsignedIntegerFieldType)
 553             ft._mapped_clk_type_name = self._default_clock_type.name
 554
 555     def _set_pkt_ctx_ft(self):
 556         members = None
 557
 558         def add_member_if_exists(name: str, ft: _FieldType, set_mapped_clk_type_name: bool = False):
 559             nonlocal members
 560
 561             if ft is not None:
 562                 if set_mapped_clk_type_name:
 563                     self._set_ft_mapped_clk_type_name(typing.cast(UnsignedIntegerFieldType, ft))
 564
 565                 members[name] = StructureFieldTypeMember(ft)
 566
 567         members = collections.OrderedDict([
 568             (
 569                 'packet_size',
 570                 StructureFieldTypeMember(self._features.packet_features.total_size_field_type)
 571             ),
 572             (
 573                 'content_size',
 574                 StructureFieldTypeMember(self._features.packet_features.content_size_field_type)
 575             )
 576         ])
 577
 578         add_member_if_exists('timestamp_begin',
 579                              self._features.packet_features.beginning_time_field_type, True)
 580         add_member_if_exists('timestamp_end', self._features.packet_features.end_time_field_type,
 581                              True)
 582         add_member_if_exists('events_discarded',
 583                              self._features.packet_features.discarded_events_counter_field_type)
 584
 585         if self._packet_context_field_type_extra_members is not None:
 586             for name, field_type in self._packet_context_field_type_extra_members.items():
 587                 assert name not in members
 588                 members[name] = field_type
 589
 590         self._pkt_ctx_ft = StructureFieldType(8, members)
 591
 592     def _set_ev_header_ft(self):
 593         members = collections.OrderedDict()
 594
 595         if self._features.event_features.type_id_field_type is not None:
 596             members['id'] = StructureFieldTypeMember(self._features.event_features.type_id_field_type)
 597
 598         if self._features.event_features.time_field_type is not None:
 599             ft = self._features.event_features.time_field_type
 600             self._set_ft_mapped_clk_type_name(ft)
 601             members['timestamp'] = StructureFieldTypeMember(ft)
 602
 603         self._ev_header_ft = StructureFieldType(8, members)
 604
 605     @property
 606     def id(self) -> Optional[Id]:
 607         return self._id
 608
 609     @property
 610     def name(self) -> str:
 611         return self._name
 612
 613     @property
 614     def default_clock_type(self) -> Optional[ClockType]:
 615         return self._default_clock_type
 616
 617     @property
 618     def features(self) -> StreamTypeFeatures:
 619         return self._features
 620
 621     @property
 622     def packet_context_field_type_extra_members(self) -> StructureFieldTypeMembers:
 623         return self._packet_context_field_type_extra_members
 624
 625     @property
 626     def event_common_context_field_type(self) -> _OptStructFt:
 627         return self._event_common_context_field_type
 628
 629     @property
 630     def event_types(self) -> FrozenSet[EventType]:
 631         return self._event_types
 632
 633
 634 _OptUuidFt = Optional[Union[str, StaticArrayFieldType]]
 635
 636
 637 class TraceTypeFeatures:
 638     def __init__(self, magic_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 639                  uuid_field_type: _OptUuidFt = None,
 640                  stream_type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 641                  default_byte_order: Optional[ByteOrder] = None):
 642         def get_field_type(user_ft: Optional[Union[str, _FieldType]],
 643                            create_default_ft: Callable[[], _FieldType]) -> _OptFt:
 644             if user_ft == DEFAULT_FIELD_TYPE:
 645                 return create_default_ft()
 646
 647             return typing.cast(_OptFt, user_ft)
 648
 649         def create_default_magic_ft():
 650             assert default_byte_order is not None
 651             return UnsignedIntegerFieldType(32, default_byte_order)
 652
 653         def create_default_uuid_ft():
 654             assert default_byte_order is not None
 655             return StaticArrayFieldType(Count(16), UnsignedIntegerFieldType(8, default_byte_order))
 656
 657         def create_default_stream_type_id_ft():
 658             assert default_byte_order is not None
 659             return UnsignedIntegerFieldType(64, default_byte_order)
 660
 661         self._magic_field_type = typing.cast(_OptUIntFt, get_field_type(magic_field_type, create_default_magic_ft))
 662         self._uuid_field_type = typing.cast(Optional[StaticArrayFieldType],
 663                                             get_field_type(uuid_field_type, create_default_uuid_ft))
 664         self._stream_type_id_field_type = typing.cast(_OptUIntFt,
 665                                                       get_field_type(stream_type_id_field_type,
 666                                                                      create_default_stream_type_id_ft))
 667
 668     @property
 669     def magic_field_type(self) -> _OptUIntFt:
 670         return self._magic_field_type
 671
 672     @property
 673     def uuid_field_type(self) -> Optional[StaticArrayFieldType]:
 674         return self._uuid_field_type
 675
 676     @property
 677     def stream_type_id_field_type(self) -> _OptUIntFt:
 678         return self._stream_type_id_field_type
 679
 680
 681 class TraceType:
 682     def __init__(self, stream_types: Set[StreamType], uuid: _OptUuid = None,
 683                  features: Optional[TraceTypeFeatures] = None,
 684                  default_feature_field_type_byte_order: Optional[ByteOrder] = None):
 685         self._stream_types = frozenset(stream_types)
 686
 687         # assign unique IDs
 688         for index, stream_type in enumerate(sorted(self._stream_types, key=lambda st: st.name)):
 689             assert stream_type._id is None
 690             stream_type._id = Id(index)
 691
 692         self._uuid = uuid
 693         self._set_features(features, default_feature_field_type_byte_order)
 694         self._set_pkt_header_ft()
 695
 696     def _set_features(self, features: Optional[TraceTypeFeatures],
 697                       default_byte_order: Optional[ByteOrder]):
 698         if features is not None:
 699             self._features = features
 700             return
 701
 702         # automatic UUID field type because the trace type has a UUID
 703         uuid_ft = None if self._uuid is None else DEFAULT_FIELD_TYPE
 704         self._features = TraceTypeFeatures(uuid_field_type=uuid_ft,
 705                                            default_byte_order=default_byte_order)
 706
 707     def _set_pkt_header_ft(self):
 708         members = collections.OrderedDict()
 709
 710         def add_member_if_exists(name: str, ft: _OptFt):
 711             nonlocal members
 712
 713             if ft is not None:
 714                 members[name] = StructureFieldTypeMember(ft)
 715
 716         add_member_if_exists('magic', self._features.magic_field_type)
 717         add_member_if_exists('uuid', self._features.uuid_field_type)
 718         add_member_if_exists('stream_id', self._features.stream_type_id_field_type)
 719         self._pkt_header_ft = StructureFieldType(8, members)
 720
 721     @property
 722     def uuid(self) -> _OptUuid:
 723         return self._uuid
 724
 725     @property
 726     def stream_types(self) -> FrozenSet[StreamType]:
 727         return self._stream_types
 728
 729     def stream_type(self, name: str) -> Optional[StreamType]:
 730         for cand_stream_type in self._stream_types:
 731             if cand_stream_type.name == name:
 732                 return cand_stream_type
 733
 734         return None
 735
 736     @property
 737     def features(self) -> TraceTypeFeatures:
 738         return self._features
 739
 740     @property
 741     def clock_types(self) -> Set[ClockType]:
 742         clk_types = set()
 743
 744         for stream_type in self._stream_types:
 745             if stream_type.default_clock_type is not None:
 746                 clk_types.add(stream_type.default_clock_type)
 747
 748         return clk_types
 749
 750
 751 _EnvEntry = Union[str, int]
 752 _EnvEntries = Mapping[str, _EnvEntry]
 753
 754
 755 class TraceEnvironment(collections.abc.Mapping):
 756     def __init__(self, environment: _EnvEntries):
 757         self._env = {name: value for name, value in environment.items()}
 758
 759     def __getitem__(self, key: str) -> _EnvEntry:
 760         return self._env[key]
 761
 762     def __iter__(self) -> Iterator[str]:
 763         return iter(self._env)
 764
 765     def __len__(self) -> int:
 766         return len(self._env)
 767
 768
 769 class Trace:
 770     def __init__(self, type: TraceType, environment: Optional[_EnvEntries] = None):
 771         self._type = type
 772         self._set_env(environment)
 773
 774     def _set_env(self, environment: Optional[_EnvEntries]):
 775         init_env = collections.OrderedDict([
 776             ('domain', 'bare'),
 777             ('tracer_name', 'barectf'),
 778             ('tracer_major', barectf_version.__major_version__),
 779             ('tracer_minor', barectf_version.__minor_version__),
 780             ('tracer_patch', barectf_version.__patch_version__),
 781             ('barectf_gen_date', str(datetime.datetime.now().isoformat())),
 782         ])
 783
 784         if environment is None:
 785             environment = {}
 786
 787         init_env.update(environment)
 788         self._env = TraceEnvironment(typing.cast(_EnvEntries, init_env))
 789
 790     @property
 791     def type(self) -> TraceType:
 792         return self._type
 793
 794     @property
 795     def environment(self) -> TraceEnvironment:
 796         return self._env
 797
 798
 799 _ClkTypeCTypes = Mapping[ClockType, str]
 800
 801
 802 class ClockTypeCTypes(collections.abc.Mapping):
 803     def __init__(self, c_types: _ClkTypeCTypes):
 804         self._c_types = {clk_type: c_type for clk_type, c_type in c_types.items()}
 805
 806     def __getitem__(self, key: ClockType) -> str:
 807         return self._c_types[key]
 808
 809     def __iter__(self) -> Iterator[ClockType]:
 810         return iter(self._c_types)
 811
 812     def __len__(self) -> int:
 813         return len(self._c_types)
 814
 815
 816 class ConfigurationCodeGenerationHeaderOptions:
 817     def __init__(self, identifier_prefix_definition: bool = False,
 818                  default_stream_type_name_definition: bool = False):
 819         self._identifier_prefix_definition = identifier_prefix_definition
 820         self._default_stream_type_name_definition = default_stream_type_name_definition
 821
 822     @property
 823     def identifier_prefix_definition(self) -> bool:
 824         return self._identifier_prefix_definition
 825
 826     @property
 827     def default_stream_type_name_definition(self) -> bool:
 828         return self._default_stream_type_name_definition
 829
 830
 831 class ConfigurationCodeGenerationOptions:
 832     def __init__(self, identifier_prefix: str = 'barectf_', file_name_prefix: str = 'barectf',
 833                  default_stream_type: Optional[StreamType] = None,
 834                  header_options: Optional[ConfigurationCodeGenerationHeaderOptions] = None,
 835                  clock_type_c_types: Optional[_ClkTypeCTypes] = None):
 836         self._identifier_prefix = identifier_prefix
 837         self._file_name_prefix = file_name_prefix
 838         self._default_stream_type = default_stream_type
 839
 840         self._header_options = ConfigurationCodeGenerationHeaderOptions()
 841
 842         if header_options is not None:
 843             self._header_options = header_options
 844
 845         self._clock_type_c_types = ClockTypeCTypes({})
 846
 847         if clock_type_c_types is not None:
 848             self._clock_type_c_types = ClockTypeCTypes(clock_type_c_types)
 849
 850     @property
 851     def identifier_prefix(self) -> str:
 852         return self._identifier_prefix
 853
 854     @property
 855     def file_name_prefix(self) -> str:
 856         return self._file_name_prefix
 857
 858     @property
 859     def default_stream_type(self) -> Optional[StreamType]:
 860         return self._default_stream_type
 861
 862     @property
 863     def header_options(self) -> ConfigurationCodeGenerationHeaderOptions:
 864         return self._header_options
 865
 866     @property
 867     def clock_type_c_types(self) -> ClockTypeCTypes:
 868         return self._clock_type_c_types
 869
 870
 871 class ConfigurationOptions:
 872     def __init__(self,
 873                  code_generation_options: Optional[ConfigurationCodeGenerationOptions] = None):
 874         self._code_generation_options = ConfigurationCodeGenerationOptions()
 875
 876         if code_generation_options is not None:
 877             self._code_generation_options = code_generation_options
 878
 879     @property
 880     def code_generation_options(self) -> ConfigurationCodeGenerationOptions:
 881         return self._code_generation_options
 882
 883
 884 class Configuration:
 885     def __init__(self, trace: Trace, target_byte_order: ByteOrder,
 886                  options: Optional[ConfigurationOptions] = None):
 887         self._trace = trace
 888         self._options = ConfigurationOptions()
 889         self._target_byte_order = target_byte_order
 890
 891         if options is not None:
 892             self._options = options
 893
 894         clk_type_c_types = self._options.code_generation_options.clock_type_c_types
 895
 896         for stream_type in trace.type.stream_types:
 897             def_clk_type = stream_type.default_clock_type
 898
 899             if def_clk_type is None:
 900                 continue
 901
 902             if def_clk_type not in clk_type_c_types:
 903                 clk_type_c_types._c_types[def_clk_type] = 'uint32_t'
 904
 905     @property
 906     def trace(self) -> Trace:
 907         return self._trace
 908
 909     @property
 910     def target_byte_order(self):
 911         return self._target_byte_order
 912
 913     @property
 914     def options(self) -> ConfigurationOptions:
 915         return self._options