1919from typing import Final , Iterable , List , Optional , Tuple , TYPE_CHECKING , TypedDict , Union
2020
2121PACKAGE_NAME_MESSAGE_TYPE_SEPARATOR : Final = '/'
22+ ANNOTATION_DELIMITER : Final = '@'
23+ OPTIONAL_ANNOTATION : Final = ANNOTATION_DELIMITER + 'optional'
2224COMMENT_DELIMITER : Final = '#'
2325CONSTANT_SEPARATOR : Final = '='
2426ARRAY_UPPER_BOUND_TOKEN : Final = '<='
8385 class Annotations (TypedDict , total = False ):
8486 comment : List [str ]
8587 unit : str
88+ optional : bool
8689
8790
8891class InvalidSpecification (Exception ):
@@ -109,6 +112,10 @@ class UnknownMessageType(InvalidSpecification):
109112 pass
110113
111114
115+ class MultipleOptionalAnnotations (InvalidSpecification ):
116+ pass
117+
118+
112119class InvalidValue (Exception ):
113120
114121 def __init__ (self , type_ : Union ['Type' , str ], value_string : str ,
@@ -408,7 +415,7 @@ def __init__(self, pkg_name: str, msg_name: str, fields: Iterable['Field'],
408415 self .msg_name = msg_name
409416 self .annotations : 'Annotations' = {}
410417
411- self .fields = []
418+ self .fields : list [ Field ] = []
412419 for index , field in enumerate (fields ):
413420 if not isinstance (field , Field ):
414421 raise TypeError ("field %u must be a 'Field' instance" % index )
@@ -422,7 +429,7 @@ def __init__(self, pkg_name: str, msg_name: str, fields: Iterable['Field'],
422429 'the fields iterable contains duplicate names: %s' %
423430 ', ' .join (sorted (duplicate_field_names )))
424431
425- self .constants = []
432+ self .constants : list [ Constant ] = []
426433 for index , constant in enumerate (constants ):
427434 if not isinstance (constant , Constant ):
428435 raise TypeError ("constant %u must be a 'Constant' instance" %
@@ -485,10 +492,11 @@ def parse_message_string(pkg_name: str, msg_name: str,
485492 fields : List [Field ] = []
486493 constants : List [Constant ] = []
487494 last_element : Union [Field , Constant , None ] = None # either a field or a constant
495+ is_optional = False
488496 # replace tabs with spaces
489497 message_string = message_string .replace ('\t ' , ' ' )
490498
491- current_comments = []
499+ current_comments : list [ str ] = []
492500 message_comments , lines = extract_file_level_comments (message_string )
493501 for line in lines :
494502 line = line .rstrip ()
@@ -522,6 +530,18 @@ def parse_message_string(pkg_name: str, msg_name: str,
522530 if not line :
523531 continue
524532
533+ annotation_index = line .rfind (OPTIONAL_ANNOTATION )
534+ if annotation_index >= 0 :
535+ if is_optional :
536+ raise MultipleOptionalAnnotations (
537+ f'Already declared @optional. Error detected with { line } .' )
538+
539+ line = line [len (OPTIONAL_ANNOTATION ):].lstrip ()
540+ is_optional = True
541+
542+ if not line :
543+ continue
544+
525545 type_string , _ , rest = line .partition (' ' )
526546 rest = rest .lstrip ()
527547 if not rest :
@@ -555,6 +575,9 @@ def parse_message_string(pkg_name: str, msg_name: str,
555575 last_element = constants [- 1 ]
556576
557577 # add "unused" comments to the field / constant
578+ if is_optional :
579+ last_element .annotations ['optional' ] = is_optional
580+ is_optional = False
558581 comment_lines = last_element .annotations .setdefault (
559582 'comment' , [])
560583 comment_lines += current_comments
0 commit comments