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 DynamicArrayFieldType(_ArrayFieldType):
 226     def __init__(self, length_field_type: UnsignedIntegerFieldType, element_field_type: _FieldType):
 227         super().__init__(element_field_type)
 228         self._length_field_type = length_field_type
 229         self._length_ft_member_name: Optional[str] = None
 230
 231     @property
 232     def length_field_type(self):
 233         return self._length_field_type
 234
 235     @property
 236     def size_is_dynamic(self):
 237         return True
 238
 239
 240 class StructureFieldTypeMember:
 241     def __init__(self, field_type: _FieldType):
 242         self._field_type = field_type
 243
 244     @property
 245     def field_type(self) -> _FieldType:
 246         return self._field_type
 247
 248
 249 _StructFtMembers = Mapping[str, StructureFieldTypeMember]
 250
 251
 252 class StructureFieldTypeMembers(collections.abc.Mapping):
 253     def __init__(self, members: _StructFtMembers):
 254         self._members = collections.OrderedDict()
 255
 256         for name, member in members.items():
 257             assert type(member) is StructureFieldTypeMember
 258             self._members[name] = member
 259
 260     def __getitem__(self, key: str) -> StructureFieldTypeMember:
 261         return self._members[key]
 262
 263     def __iter__(self) -> Iterator[str]:
 264         return iter(self._members)
 265
 266     def __len__(self) -> int:
 267         return len(self._members)
 268
 269
 270 class StructureFieldType(_FieldType):
 271     def __init__(self, minimum_alignment: Alignment = Alignment(1),
 272                  members: Optional[_StructFtMembers] = None):
 273         self._minimum_alignment = minimum_alignment
 274         self._members = StructureFieldTypeMembers({})
 275
 276         if members is not None:
 277             self._members = StructureFieldTypeMembers(members)
 278
 279         self._set_alignment()
 280         self._set_dyn_array_ft_length_ft_member_names()
 281
 282     def _set_alignment(self):
 283         self._alignment: Alignment = self._minimum_alignment
 284
 285         for member in self._members.values():
 286             if member.field_type.alignment > self._alignment:
 287                 self._alignment = member.field_type.alignment
 288
 289     def _set_dyn_array_ft_length_ft_member_names(self):
 290         for member in self._members.values():
 291             if type(member.field_type) is DynamicArrayFieldType:
 292                 # Find length field type member name within the same
 293                 # structure field type members.
 294                 for len_name, len_member in self._members.items():
 295                     if member.field_type.length_field_type is len_member.field_type:
 296                         member.field_type._length_ft_member_name = len_name
 297                         len_member.field_type._is_len = True
 298                         break
 299
 300             if member.field_type.alignment > self._alignment:
 301                 self._alignment = member.field_type.alignment
 302
 303     @property
 304     def minimum_alignment(self) -> Alignment:
 305         return self._minimum_alignment
 306
 307     @property
 308     def alignment(self) -> Alignment:
 309         return self._alignment
 310
 311     @property
 312     def size_is_dynamic(self):
 313         return any([member.field_type.size_is_dynamic for member in self.members.values()])
 314
 315     @property
 316     def members(self) -> StructureFieldTypeMembers:
 317         return self._members
 318
 319
 320 class _UniqueByName:
 321     _name: str
 322
 323     def __eq__(self, other: Any) -> bool:
 324         if type(other) is not type(self):
 325             return False
 326
 327         return self._name == other._name
 328
 329     def __lt__(self, other: '_UniqueByName'):
 330         assert type(self) is type(other)
 331         return self._name < other._name
 332
 333     def __hash__(self) -> int:
 334         return hash(self._name)
 335
 336
 337 _OptFt = Optional[_FieldType]
 338 _OptStructFt = Optional[StructureFieldType]
 339 LogLevel = typing.NewType('LogLevel', int)
 340
 341
 342 class EventType(_UniqueByName):
 343     def __init__(self, name: str, log_level: Optional[LogLevel] = None,
 344                  specific_context_field_type: _OptStructFt = None, payload_field_type: _OptStructFt = None):
 345         self._id: Optional[Id] = None
 346         self._name = name
 347         self._log_level = log_level
 348         self._specific_context_field_type = specific_context_field_type
 349         self._payload_field_type = payload_field_type
 350
 351     @property
 352     def id(self) -> Optional[Id]:
 353         return self._id
 354
 355     @property
 356     def name(self) -> str:
 357         return self._name
 358
 359     @property
 360     def log_level(self) -> Optional[LogLevel]:
 361         return self._log_level
 362
 363     @property
 364     def specific_context_field_type(self) -> _OptStructFt:
 365         return self._specific_context_field_type
 366
 367     @property
 368     def payload_field_type(self) -> _OptStructFt:
 369         return self._payload_field_type
 370
 371
 372 class ClockTypeOffset:
 373     def __init__(self, seconds: int = 0, cycles: Count = Count(0)):
 374         self._seconds = seconds
 375         self._cycles = cycles
 376
 377     @property
 378     def seconds(self) -> int:
 379         return self._seconds
 380
 381     @property
 382     def cycles(self) -> Count:
 383         return self._cycles
 384
 385
 386 _OptUuid = Optional[uuidp.UUID]
 387
 388
 389 class ClockType(_UniqueByName):
 390     def __init__(self, name: str, frequency: Count = Count(int(1e9)), uuid: _OptUuid = None,
 391                  description: _OptStr = None, precision: Count = Count(0),
 392                  offset: Optional[ClockTypeOffset] = None, origin_is_unix_epoch: bool = False):
 393         self._name = name
 394         self._frequency = frequency
 395         self._uuid = uuid
 396         self._description = description
 397         self._precision = precision
 398         self._offset = ClockTypeOffset()
 399
 400         if offset is not None:
 401             self._offset = offset
 402
 403         self._origin_is_unix_epoch = origin_is_unix_epoch
 404
 405     @property
 406     def name(self) -> str:
 407         return self._name
 408
 409     @property
 410     def frequency(self) -> Count:
 411         return self._frequency
 412
 413     @property
 414     def uuid(self) -> _OptUuid:
 415         return self._uuid
 416
 417     @property
 418     def description(self) -> _OptStr:
 419         return self._description
 420
 421     @property
 422     def precision(self) -> Count:
 423         return self._precision
 424
 425     @property
 426     def offset(self) -> ClockTypeOffset:
 427         return self._offset
 428
 429     @property
 430     def origin_is_unix_epoch(self) -> bool:
 431         return self._origin_is_unix_epoch
 432
 433
 434 DEFAULT_FIELD_TYPE = 'default'
 435 _DefaultableUIntFt = Union[str, UnsignedIntegerFieldType]
 436 _OptDefaultableUIntFt = Optional[_DefaultableUIntFt]
 437 _OptUIntFt = Optional[UnsignedIntegerFieldType]
 438
 439
 440 class StreamTypePacketFeatures:
 441     def __init__(self, total_size_field_type: _DefaultableUIntFt = DEFAULT_FIELD_TYPE,
 442                  content_size_field_type: _DefaultableUIntFt = DEFAULT_FIELD_TYPE,
 443                  beginning_time_field_type: _OptDefaultableUIntFt = None,
 444                  end_time_field_type: _OptDefaultableUIntFt = None,
 445                  discarded_events_counter_field_type: _OptDefaultableUIntFt = None):
 446         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 447             if user_ft == DEFAULT_FIELD_TYPE:
 448                 return UnsignedIntegerFieldType(64)
 449
 450             return typing.cast(_OptUIntFt, user_ft)
 451
 452         self._total_size_field_type = get_ft(total_size_field_type)
 453         self._content_size_field_type = get_ft(content_size_field_type)
 454         self._beginning_time_field_type = get_ft(beginning_time_field_type)
 455         self._end_time_field_type = get_ft(end_time_field_type)
 456         self._discarded_events_counter_field_type = get_ft(discarded_events_counter_field_type)
 457
 458     @property
 459     def total_size_field_type(self) -> _OptUIntFt:
 460         return self._total_size_field_type
 461
 462     @property
 463     def content_size_field_type(self) -> _OptUIntFt:
 464         return self._content_size_field_type
 465
 466     @property
 467     def beginning_time_field_type(self) -> _OptUIntFt:
 468         return self._beginning_time_field_type
 469
 470     @property
 471     def end_time_field_type(self) -> _OptUIntFt:
 472         return self._end_time_field_type
 473
 474     @property
 475     def discarded_events_counter_field_type(self) -> _OptUIntFt:
 476         return self._discarded_events_counter_field_type
 477
 478
 479 class StreamTypeEventFeatures:
 480     def __init__(self, type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 481                  time_field_type: _OptDefaultableUIntFt = None):
 482         def get_ft(user_ft: _OptDefaultableUIntFt) -> _OptUIntFt:
 483             if user_ft == DEFAULT_FIELD_TYPE:
 484                 return UnsignedIntegerFieldType(64)
 485
 486             return typing.cast(_OptUIntFt, user_ft)
 487
 488         self._type_id_field_type = get_ft(type_id_field_type)
 489         self._time_field_type = get_ft(time_field_type)
 490
 491     @property
 492     def type_id_field_type(self) -> _OptUIntFt:
 493         return self._type_id_field_type
 494
 495     @property
 496     def time_field_type(self) -> _OptUIntFt:
 497         return self._time_field_type
 498
 499
 500 class StreamTypeFeatures:
 501     def __init__(self, packet_features: Optional[StreamTypePacketFeatures] = None,
 502                  event_features: Optional[StreamTypeEventFeatures] = None):
 503         if packet_features is None:
 504             self._packet_features = StreamTypePacketFeatures()
 505         else:
 506             self._packet_features = packet_features
 507
 508         if event_features is None:
 509             self._event_features = StreamTypeEventFeatures()
 510         else:
 511             self._event_features = event_features
 512
 513     @property
 514     def packet_features(self) -> StreamTypePacketFeatures:
 515         return self._packet_features
 516
 517     @property
 518     def event_features(self) -> StreamTypeEventFeatures:
 519         return self._event_features
 520
 521
 522 class StreamType(_UniqueByName):
 523     def __init__(self, name: str, event_types: Set[EventType],
 524                  default_clock_type: Optional[ClockType] = None,
 525                  features: Optional[StreamTypeFeatures] = None,
 526                  packet_context_field_type_extra_members: Optional[_StructFtMembers] = None,
 527                  event_common_context_field_type: _OptStructFt = None):
 528         self._id: Optional[Id] = None
 529         self._name = name
 530         self._default_clock_type = default_clock_type
 531         self._event_common_context_field_type = event_common_context_field_type
 532         self._event_types = frozenset(event_types)
 533
 534         # assign unique IDs
 535         for index, ev_type in enumerate(sorted(self._event_types, key=lambda evt: evt.name)):
 536             assert ev_type._id is None
 537             ev_type._id = Id(index)
 538
 539         self._set_features(features)
 540         self._packet_context_field_type_extra_members = StructureFieldTypeMembers({})
 541
 542         if packet_context_field_type_extra_members is not None:
 543             self._packet_context_field_type_extra_members = StructureFieldTypeMembers(packet_context_field_type_extra_members)
 544
 545         self._set_pkt_ctx_ft()
 546         self._set_ev_header_ft()
 547
 548     def _set_features(self, features: Optional[StreamTypeFeatures]):
 549         if features is not None:
 550             self._features = features
 551             return None
 552
 553         ev_time_ft = None
 554         pkt_beginning_time_ft = None
 555         pkt_end_time_ft = None
 556
 557         if self._default_clock_type is not None:
 558             # Automatic time field types because the stream type has a
 559             # default clock type.
 560             ev_time_ft = DEFAULT_FIELD_TYPE
 561             pkt_beginning_time_ft = DEFAULT_FIELD_TYPE
 562             pkt_end_time_ft = DEFAULT_FIELD_TYPE
 563
 564         self._features = StreamTypeFeatures(StreamTypePacketFeatures(beginning_time_field_type=pkt_beginning_time_ft,
 565                                                                      end_time_field_type=pkt_end_time_ft),
 566                                             StreamTypeEventFeatures(time_field_type=ev_time_ft))
 567
 568     def _set_ft_mapped_clk_type_name(self, ft: Optional[UnsignedIntegerFieldType]):
 569         if ft is None:
 570             return
 571
 572         if self._default_clock_type is not None:
 573             assert isinstance(ft, UnsignedIntegerFieldType)
 574             ft._mapped_clk_type_name = self._default_clock_type.name
 575
 576     def _set_pkt_ctx_ft(self):
 577         members = None
 578
 579         def add_member_if_exists(name: str, ft: _FieldType, set_mapped_clk_type_name: bool = False):
 580             nonlocal members
 581
 582             if ft is not None:
 583                 if set_mapped_clk_type_name:
 584                     self._set_ft_mapped_clk_type_name(typing.cast(UnsignedIntegerFieldType, ft))
 585
 586                 members[name] = StructureFieldTypeMember(ft)
 587
 588         members = collections.OrderedDict([
 589             (
 590                 'packet_size',
 591                 StructureFieldTypeMember(self._features.packet_features.total_size_field_type)
 592             ),
 593             (
 594                 'content_size',
 595                 StructureFieldTypeMember(self._features.packet_features.content_size_field_type)
 596             )
 597         ])
 598
 599         add_member_if_exists('timestamp_begin',
 600                              self._features.packet_features.beginning_time_field_type, True)
 601         add_member_if_exists('timestamp_end', self._features.packet_features.end_time_field_type,
 602                              True)
 603         add_member_if_exists('events_discarded',
 604                              self._features.packet_features.discarded_events_counter_field_type)
 605
 606         if self._packet_context_field_type_extra_members is not None:
 607             for name, field_type in self._packet_context_field_type_extra_members.items():
 608                 assert name not in members
 609                 members[name] = field_type
 610
 611         self._pkt_ctx_ft = StructureFieldType(8, members)
 612
 613     def _set_ev_header_ft(self):
 614         members = collections.OrderedDict()
 615
 616         if self._features.event_features.type_id_field_type is not None:
 617             members['id'] = StructureFieldTypeMember(self._features.event_features.type_id_field_type)
 618
 619         if self._features.event_features.time_field_type is not None:
 620             ft = self._features.event_features.time_field_type
 621             self._set_ft_mapped_clk_type_name(ft)
 622             members['timestamp'] = StructureFieldTypeMember(ft)
 623
 624         self._ev_header_ft = StructureFieldType(8, members)
 625
 626     @property
 627     def id(self) -> Optional[Id]:
 628         return self._id
 629
 630     @property
 631     def name(self) -> str:
 632         return self._name
 633
 634     @property
 635     def default_clock_type(self) -> Optional[ClockType]:
 636         return self._default_clock_type
 637
 638     @property
 639     def features(self) -> StreamTypeFeatures:
 640         return self._features
 641
 642     @property
 643     def packet_context_field_type_extra_members(self) -> StructureFieldTypeMembers:
 644         return self._packet_context_field_type_extra_members
 645
 646     @property
 647     def event_common_context_field_type(self) -> _OptStructFt:
 648         return self._event_common_context_field_type
 649
 650     @property
 651     def event_types(self) -> FrozenSet[EventType]:
 652         return self._event_types
 653
 654
 655 _OptUuidFt = Optional[Union[str, StaticArrayFieldType]]
 656
 657
 658 class TraceTypeFeatures:
 659     def __init__(self, magic_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE,
 660                  uuid_field_type: _OptUuidFt = None,
 661                  stream_type_id_field_type: _OptDefaultableUIntFt = DEFAULT_FIELD_TYPE):
 662         def get_field_type(user_ft: Optional[Union[str, _FieldType]],
 663                            create_default_ft: Callable[[], _FieldType]) -> _OptFt:
 664             if user_ft == DEFAULT_FIELD_TYPE:
 665                 return create_default_ft()
 666
 667             return typing.cast(_OptFt, user_ft)
 668
 669         def create_default_magic_ft():
 670             return UnsignedIntegerFieldType(32)
 671
 672         def create_default_uuid_ft():
 673             return StaticArrayFieldType(Count(16), UnsignedIntegerFieldType(8))
 674
 675         def create_default_stream_type_id_ft():
 676             return UnsignedIntegerFieldType(64)
 677
 678         self._magic_field_type = typing.cast(_OptUIntFt, get_field_type(magic_field_type, create_default_magic_ft))
 679         self._uuid_field_type = typing.cast(Optional[StaticArrayFieldType],
 680                                             get_field_type(uuid_field_type, create_default_uuid_ft))
 681         self._stream_type_id_field_type = typing.cast(_OptUIntFt,
 682                                                       get_field_type(stream_type_id_field_type,
 683                                                                      create_default_stream_type_id_ft))
 684
 685     @property
 686     def magic_field_type(self) -> _OptUIntFt:
 687         return self._magic_field_type
 688
 689     @property
 690     def uuid_field_type(self) -> Optional[StaticArrayFieldType]:
 691         return self._uuid_field_type
 692
 693     @property
 694     def stream_type_id_field_type(self) -> _OptUIntFt:
 695         return self._stream_type_id_field_type
 696
 697
 698 class TraceType:
 699     def __init__(self, stream_types: Set[StreamType], uuid: _OptUuid = None,
 700                  features: Optional[TraceTypeFeatures] = None):
 701         self._stream_types = frozenset(stream_types)
 702
 703         # assign unique IDs
 704         for index, stream_type in enumerate(sorted(self._stream_types, key=lambda st: st.name)):
 705             assert stream_type._id is None
 706             stream_type._id = Id(index)
 707
 708         self._uuid = uuid
 709         self._set_features(features)
 710         self._set_pkt_header_ft()
 711
 712     def _set_features(self, features: Optional[TraceTypeFeatures]):
 713         if features is not None:
 714             self._features = features
 715             return
 716
 717         # automatic UUID field type because the trace type has a UUID
 718         uuid_ft = None if self._uuid is None else DEFAULT_FIELD_TYPE
 719         self._features = TraceTypeFeatures(uuid_field_type=uuid_ft)
 720
 721     def _set_pkt_header_ft(self):
 722         members = collections.OrderedDict()
 723
 724         def add_member_if_exists(name: str, ft: _OptFt):
 725             nonlocal members
 726
 727             if ft is not None:
 728                 members[name] = StructureFieldTypeMember(ft)
 729
 730         add_member_if_exists('magic', self._features.magic_field_type)
 731         add_member_if_exists('uuid', self._features.uuid_field_type)
 732         add_member_if_exists('stream_id', self._features.stream_type_id_field_type)
 733         self._pkt_header_ft = StructureFieldType(8, members)
 734
 735     @property
 736     def uuid(self) -> _OptUuid:
 737         return self._uuid
 738
 739     @property
 740     def stream_types(self) -> FrozenSet[StreamType]:
 741         return self._stream_types
 742
 743     def stream_type(self, name: str) -> Optional[StreamType]:
 744         for cand_stream_type in self._stream_types:
 745             if cand_stream_type.name == name:
 746                 return cand_stream_type
 747
 748         return None
 749
 750     @property
 751     def features(self) -> TraceTypeFeatures:
 752         return self._features
 753
 754     @property
 755     def clock_types(self) -> Set[ClockType]:
 756         clk_types = set()
 757
 758         for stream_type in self._stream_types:
 759             if stream_type.default_clock_type is not None:
 760                 clk_types.add(stream_type.default_clock_type)
 761
 762         return clk_types
 763
 764
 765 _EnvEntry = Union[str, int]
 766 _EnvEntries = Mapping[str, _EnvEntry]
 767
 768
 769 class TraceEnvironment(collections.abc.Mapping):
 770     def __init__(self, environment: _EnvEntries):
 771         self._env = {name: value for name, value in environment.items()}
 772
 773     def __getitem__(self, key: str) -> _EnvEntry:
 774         return self._env[key]
 775
 776     def __iter__(self) -> Iterator[str]:
 777         return iter(self._env)
 778
 779     def __len__(self) -> int:
 780         return len(self._env)
 781
 782
 783 class Trace:
 784     def __init__(self, type: TraceType, environment: Optional[_EnvEntries] = None):
 785         self._type = type
 786         self._set_env(environment)
 787
 788     def _set_env(self, environment: Optional[_EnvEntries]):
 789         init_env = collections.OrderedDict([
 790             ('domain', 'bare'),
 791             ('tracer_name', 'barectf'),
 792             ('tracer_major', barectf_version.__major_version__),
 793             ('tracer_minor', barectf_version.__minor_version__),
 794             ('tracer_patch', barectf_version.__patch_version__),
 795             ('barectf_gen_date', str(datetime.datetime.now().isoformat())),
 796         ])
 797
 798         if environment is None:
 799             environment = {}
 800
 801         init_env.update(environment)
 802         self._env = TraceEnvironment(typing.cast(_EnvEntries, init_env))
 803
 804     @property
 805     def type(self) -> TraceType:
 806         return self._type
 807
 808     @property
 809     def environment(self) -> TraceEnvironment:
 810         return self._env
 811
 812
 813 _ClkTypeCTypes = Mapping[ClockType, str]
 814
 815
 816 class ClockTypeCTypes(collections.abc.Mapping):
 817     def __init__(self, c_types: _ClkTypeCTypes):
 818         self._c_types = {clk_type: c_type for clk_type, c_type in c_types.items()}
 819
 820     def __getitem__(self, key: ClockType) -> str:
 821         return self._c_types[key]
 822
 823     def __iter__(self) -> Iterator[ClockType]:
 824         return iter(self._c_types)
 825
 826     def __len__(self) -> int:
 827         return len(self._c_types)
 828
 829
 830 class ConfigurationCodeGenerationHeaderOptions:
 831     def __init__(self, identifier_prefix_definition: bool = False,
 832                  default_stream_type_name_definition: bool = False):
 833         self._identifier_prefix_definition = identifier_prefix_definition
 834         self._default_stream_type_name_definition = default_stream_type_name_definition
 835
 836     @property
 837     def identifier_prefix_definition(self) -> bool:
 838         return self._identifier_prefix_definition
 839
 840     @property
 841     def default_stream_type_name_definition(self) -> bool:
 842         return self._default_stream_type_name_definition
 843
 844
 845 class ConfigurationCodeGenerationOptions:
 846     def __init__(self, identifier_prefix: str = 'barectf_', file_name_prefix: str = 'barectf',
 847                  default_stream_type: Optional[StreamType] = None,
 848                  header_options: Optional[ConfigurationCodeGenerationHeaderOptions] = None,
 849                  clock_type_c_types: Optional[_ClkTypeCTypes] = None):
 850         self._identifier_prefix = identifier_prefix
 851         self._file_name_prefix = file_name_prefix
 852         self._default_stream_type = default_stream_type
 853
 854         self._header_options = ConfigurationCodeGenerationHeaderOptions()
 855
 856         if header_options is not None:
 857             self._header_options = header_options
 858
 859         self._clock_type_c_types = ClockTypeCTypes({})
 860
 861         if clock_type_c_types is not None:
 862             self._clock_type_c_types = ClockTypeCTypes(clock_type_c_types)
 863
 864     @property
 865     def identifier_prefix(self) -> str:
 866         return self._identifier_prefix
 867
 868     @property
 869     def file_name_prefix(self) -> str:
 870         return self._file_name_prefix
 871
 872     @property
 873     def default_stream_type(self) -> Optional[StreamType]:
 874         return self._default_stream_type
 875
 876     @property
 877     def header_options(self) -> ConfigurationCodeGenerationHeaderOptions:
 878         return self._header_options
 879
 880     @property
 881     def clock_type_c_types(self) -> ClockTypeCTypes:
 882         return self._clock_type_c_types
 883
 884
 885 class ConfigurationOptions:
 886     def __init__(self,
 887                  code_generation_options: Optional[ConfigurationCodeGenerationOptions] = None):
 888         self._code_generation_options = ConfigurationCodeGenerationOptions()
 889
 890         if code_generation_options is not None:
 891             self._code_generation_options = code_generation_options
 892
 893     @property
 894     def code_generation_options(self) -> ConfigurationCodeGenerationOptions:
 895         return self._code_generation_options
 896
 897
 898 class Configuration:
 899     def __init__(self, trace: Trace, target_byte_order: ByteOrder,
 900                  options: Optional[ConfigurationOptions] = None):
 901         self._trace = trace
 902         self._options = ConfigurationOptions()
 903         self._target_byte_order = target_byte_order
 904
 905         if options is not None:
 906             self._options = options
 907
 908         clk_type_c_types = self._options.code_generation_options.clock_type_c_types
 909
 910         for stream_type in trace.type.stream_types:
 911             def_clk_type = stream_type.default_clock_type
 912
 913             if def_clk_type is None:
 914                 continue
 915
 916             if def_clk_type not in clk_type_c_types:
 917                 clk_type_c_types._c_types[def_clk_type] = 'uint32_t'
 918
 919     @property
 920     def trace(self) -> Trace:
 921         return self._trace
 922
 923     @property
 924     def target_byte_order(self):
 925         return self._target_byte_order
 926
 927     @property
 928     def options(self) -> ConfigurationOptions:
 929         return self._options