22This is the parser service.
33"""
44from io import StringIO
5- from pathlib import Path
6- from typing import List , Dict , Union
5+ from typing import List , Dict , Optional , Any
76import struct
87from operator import itemgetter
98
3534
3635class KOFParser :
3736 tema_codes_mapping = {
38- "2401" : model .MethodTypeEnum .RWS . name ,
39- "2402" : model .MethodTypeEnum .SA . name ,
40- "2403" : model .MethodTypeEnum .TP . name ,
41- "2405" : model .MethodTypeEnum .SS . name ,
42- "2406" : model .MethodTypeEnum .RP . name ,
43- "2407" : model .MethodTypeEnum .CPT . name ,
44- "2409" : model .MethodTypeEnum .RS . name ,
45- "2410" : model .MethodTypeEnum .SR . name ,
46- "2411" : model .MethodTypeEnum .SPT . name ,
47- "2412" : model .MethodTypeEnum .RCD . name ,
48- "2413" : model .MethodTypeEnum .PZ . name ,
49- "2414" : model .MethodTypeEnum .PT . name ,
50- "2415" : model .MethodTypeEnum .SVT . name ,
51- "2417" : model .MethodTypeEnum .INC . name ,
52- "2418" : model .MethodTypeEnum .TOT . name ,
37+ "2401" : model .MethodTypeEnum .RWS ,
38+ "2402" : model .MethodTypeEnum .SA ,
39+ "2403" : model .MethodTypeEnum .TP ,
40+ "2405" : model .MethodTypeEnum .SS ,
41+ "2406" : model .MethodTypeEnum .RP ,
42+ "2407" : model .MethodTypeEnum .CPT ,
43+ "2409" : model .MethodTypeEnum .RS ,
44+ "2410" : model .MethodTypeEnum .SR ,
45+ "2411" : model .MethodTypeEnum .SPT ,
46+ "2412" : model .MethodTypeEnum .RCD ,
47+ "2413" : model .MethodTypeEnum .PZ ,
48+ "2414" : model .MethodTypeEnum .PT ,
49+ "2415" : model .MethodTypeEnum .SVT ,
50+ "2417" : model .MethodTypeEnum .INC ,
51+ "2418" : model .MethodTypeEnum .TOT ,
5352 }
5453
55- def temakode_to_method (self , temakode : str ) -> str :
54+ def __init__ (self ):
55+ self .fieldspecs : List [List [Any ]] = [
56+ # Name, Start, Width, Type
57+ ["ID" , 5 , 10 , str ],
58+ ["TEMAKODE" , 16 , 8 , str ],
59+ ["x" , 25 , 12 , float ],
60+ ["y" , 38 , 11 , float ],
61+ ["z" , 50 , 8 , float ],
62+ ]
63+ self .iname , self .istart , self .iwidth , self .itype = 0 , 1 , 2 , 3 # field indexes
64+ self .fieldspecs .sort (key = itemgetter (self .istart ))
65+ self .field_indices = range (len (self .fieldspecs ))
66+ self .struct_unpacker = self .get_struct_unpacker (self .fieldspecs , self .istart , self .iwidth )
67+
68+ def temakode_to_method (self , temakode : str ) -> Optional [str ]:
5669
5770 if not temakode or temakode not in self .tema_codes_mapping .keys ():
5871 return None
5972
60- return self .tema_codes_mapping [temakode ]
73+ return self .tema_codes_mapping [temakode ]. name
6174
75+ @staticmethod
6276 def get_struct_unpacker (fieldspecs , istart , iwidth ):
6377 """
6478 Build the format string for struct.unpack to use, based on the fieldspecs.
@@ -77,27 +91,11 @@ def get_struct_unpacker(fieldspecs, istart, iwidth):
7791 struct_unpacker = struct .Struct (unpack_fmt ).unpack_from
7892 return struct_unpacker
7993
80- fieldspecs = [
81- # Name, Start, Width, Type
82- ["ID" , 5 , 10 , str ],
83- ["TEMAKODE" , 16 , 8 , str ],
84- ["x" , 25 , 12 , float ],
85- ["y" , 38 , 11 , float ],
86- ["z" , 50 , 8 , float ],
87- ]
88- iname , istart , iwidth , itype = 0 , 1 , 2 , 3 # field indexes
89-
90- fieldspecs .sort (key = itemgetter (istart ))
91- struct_unpacker = get_struct_unpacker (fieldspecs , istart , iwidth )
92- field_indices = range (len (fieldspecs ))
93-
94- def parse (self , filepath_or_buffer : Union [str , Path , StringIO ], srid :int )-> List [model .Location ]:
94+ def parse (self , filepath_or_buffer : Any , srid : int ) -> List [model .Location ]:
9595 if self ._is_file_like (filepath_or_buffer ):
9696 f = filepath_or_buffer
9797 close_file = False
9898 else :
99- # Read file with errors="replace" to catch UnicodeDecodeErrors
100- # f = open(filepath_or_buffer, "r", encoding='ASCII', errors="replace")
10199 f = open (filepath_or_buffer , "rb" )
102100 close_file = True
103101 try :
@@ -107,12 +105,11 @@ def parse(self, filepath_or_buffer: Union[str, Path, StringIO], srid:int)-> List
107105 f .close ()
108106
109107 def read_kof (self , file : StringIO , srid : int ) -> List [model .Location ]:
110- locations : Dict [model .Location ] = {}
111- # data = codecs.getreader("iso-8859-1")(file)
108+ locations : Dict [str , model .Location ] = {}
112109 for line in file .readlines ():
113110 if line [0 :3 ] == b" 05" :
114111 raw_fields = self .struct_unpacker (line ) # split line into field values
115- line_data = {}
112+ line_data : Dict [ str , Any ] = {}
116113 for i in self .field_indices :
117114 fieldspec = self .fieldspecs [i ]
118115 fieldname = fieldspec [self .iname ]
@@ -137,7 +134,7 @@ def read_kof(self, file: StringIO, srid: int) -> List[model.Location]:
137134 return [location for name , location in locations .items ()]
138135
139136 @staticmethod
140- def _is_file_like (obj ):
137+ def _is_file_like (obj ) -> bool :
141138 """Check if object is file like
142139
143140 Returns
@@ -146,11 +143,10 @@ def _is_file_like(obj):
146143 Return True if obj is file like, otherwise return False
147144 """
148145
149- if not (hasattr (obj , ' read' ) or hasattr (obj , ' write' )):
146+ if not (hasattr (obj , " read" ) or hasattr (obj , " write" )):
150147 return False
151148
152149 if not hasattr (obj , "__iter__" ):
153150 return False
154151
155152 return True
156-
0 commit comments