Coverage for piemap/dsl.py: 88.04%

1import collections

2import json

3from typing import no_type_check

5import piemap.projections as pr

6from piemap import PIPE, SEMI, log

8NEEDED_NUMBER_OF_AXIS_MAX = 16

9NULL_STR_REP = 'NULL'

10REC_SEP = ';'

11ROW_SEP = '\n'

14@no_type_check

15def dumps(config):

16 """Dump configuration into string."""

17 return json.dumps(config)

20@no_type_check

21def dump(config, handle):

22 """Dump configuration into JSON file via handle."""

23 return json.dump(config, handle)

26@no_type_check

27def loads(text):

28 """Load configuration from string."""

29 return NotImplemented

32@no_type_check

33def load(handle):

34 """Load configuration from JSON file handle."""

35 return json.load(handle)

38@no_type_check

39def default_linear():

40 """DRY."""

41 return {

42 'AXIS_INDEX': '',

43 'AXIS_NAME': 'Dimension',

44 'AXIS_TYPE': 'LINEAR',

45 'AXIS_MIN': 0.00,

46 'AXIS_LIMIT': 0.80,

47 'AXIS_MAX': 1.00,

48 'AXIS_LIMIT_FOLDED': False,

49 'AXIS_MIN_FOLDED': False,

50 'AXIS_VALUE': NULL_STR_REP,

51 'AXIS_UNIT': '1',

52 'AXIS_META': '',

53 }

56@no_type_check

57def default_keys():

58 """DRY."""

59 return list(default_linear().keys())

62@no_type_check

63def default_linear_values():

64 """DRY."""

65 return list(default_linear().values())

68@no_type_check

69def default_folded():

70 """DRY."""

71 return {

72 'AXIS_INDEX': '',

73 'AXIS_NAME': 'DimensionFolded',

74 'AXIS_TYPE': 'FOLDED',

75 'AXIS_MIN': 0.00,

76 'AXIS_LIMIT': 0.80,

77 'AXIS_MAX': 1.00,

78 'AXIS_LIMIT_FOLDED': False,

79 'AXIS_MIN_FOLDED': False,

80 'AXIS_VALUE': NULL_STR_REP,

81 'AXIS_UNIT': 'dB',

82 'AXIS_META': '',

83 }

86@no_type_check

87def default_folded_values():

88 """DRY."""

89 return list(default_folded().values())

92@no_type_check

93def unsafe(text):

94 """TODO(sthagen) keep/drop list applies here."""

95 return text

98@no_type_check

99def is_numeric(text):

100 """Migration artifact."""

101 try:

102 _ = float(text)

103 return True

104 except (TypeError, ValueError):

105 return False

106

107

108@no_type_check

109def maybe_int(value):

110 """Rococo."""

111 if float(value) == float(int(float(value))):

112 return int(value)

113

114 return value

115

116

117@no_type_check

118def compact_value(text):

119 """HACK A DID ACK ..."""

120 if not is_numeric(text):

121 return text

122

123 try:

124 as_int = int(text)

125 return as_int

126 except (TypeError, ValueError):

127 return float(text)

128

129

130@no_type_check

131def example_default():

132 """Maybe not wanted but matches reference behavior."""

133 some_axis_maps = [default_linear()]

134 some_axis_maps[0]['AXIS_INDEX'] = 0

135

136 some_axis_maps.append(default_linear())

137 some_axis_maps[1]['AXIS_INDEX'] = 1

138 some_axis_maps[1]['AXIS_NAME'] = 'Dimension2'

139

140 some_axis_maps.append(default_folded())

141 some_axis_maps[2]['AXIS_INDEX'] = 2

142

143 return some_axis_maps, ['Default used, since no input given.']

144

145

146@no_type_check

147def parse(text: str):

148 """Parse the DSL contained in text."""

149 if not text.strip():

150 return example_default()

151

152 has_index_collision = False

153 has_index_order_mismatch = False

154

155 info_queue, some_axis_maps = [], []

156 axis_values_rows_req_string: str = unsafe(text)

157 axis_values_rows_req = [row.strip() for row in axis_values_rows_req_string.split(ROW_SEP) if row.strip()]

158 n_axis_rows_req = len(axis_values_rows_req)

159 for n, row_string in enumerate(

160 axis_values_rows_req[:NEEDED_NUMBER_OF_AXIS_MAX]

161 ): # Was: array_slice(axis_values_rows_req, 0, NEEDED_NUMBER_OF_AXIS_MAX)

162 axis_values = default_folded_values() if ';FOLDED;' in row_string else default_linear_values()

163 axis_values_cand = row_string.split(REC_SEP)

164 log.info(f'__{"FOLDED" if ";FOLDED;" in row_string else "NO_FOLD"}__')

165 log.info(f'has axis_values[{len(axis_values)}]({axis_values})')

166 if len(axis_values) >= len(axis_values_cand): 166 ↛ 175line 166 didn't jump to line 175, because the condition on line 166 was never false

167 kks = default_keys()

168 for i, v in enumerate(axis_values_cand):

169 if v != '':

170 be_v = compact_value(v)

171 log.debug(f'[{i}]( == {kks[i]}): ({axis_values[i]}) --> ({be_v})')

172 axis_values[i] = be_v

173 else:

174 log.debug(f'[{i}]( == {kks[i]}): ({axis_values[i]}) kept ({axis_values[i]})')

175 if len(axis_values) > len(axis_values_cand):

176 kks = default_keys()

177 for i, v in enumerate(axis_values[len(axis_values_cand) :], start=len(axis_values_cand)):

178 log.debug(f'[{i}]( == {kks[i]}): ({axis_values[i]}) untouched ({axis_values[i]})')

179

180 log.info(f'has axis candidate[{len(axis_values_cand)}]({axis_values_cand})')

181 axis_map = dict(zip(default_keys(), axis_values))

182 log.info(f'... axis map is ({axis_map})')

183 numeric_axis_types = ('LINEAR', 'FOLDED')

184 if axis_map['AXIS_INDEX'] == '':

185 log.info(f"has index idea({axis_map['AXIS_INDEX']}) is False?")

186 log.info(f'... axis map is ({axis_map})')

187 axis_map['AXIS_INDEX'] = n

188 i_cfc = axis_map['AXIS_INDEX']

189 else:

190 print()

191 log.info(f"has index idea({axis_map['AXIS_INDEX']}) is True")

192 index_cand = str(axis_map['AXIS_INDEX'])

193 if is_numeric(index_cand) and float(index_cand) == float(int(float(index_cand))):

194 log.info(f'is_numeric({index_cand}) is True')

195 i_cfc = str(maybe_int(index_cand))

196 axis_map['AXIS_INDEX'] = int(i_cfc)

197 if index_cand != i_cfc: # Was !== in PHP 197 ↛ 198line 197 didn't jump to line 198, because the condition on line 197 was never true

198 info_queue.append(f'NOK index ({index_cand}) requested, accepted as ({i_cfc})')

199 else:

200 info_queue.append(f' OK index ({index_cand}) requested, accepted as ({i_cfc})')

201 else:

202 log.info(f'is_numeric({index_cand}) is False')

203 i_cfc = str(n)

204 info_queue.append(

205 f'NOK invalid index ({index_cand}) requested, accepted per parsing order as ({i_cfc})'

206 )

207 if axis_map['AXIS_TYPE'] in numeric_axis_types:

208 if axis_map['AXIS_TYPE'] == 'LINEAR':

209 if is_numeric(axis_map['AXIS_LIMIT']) and is_numeric(axis_map['AXIS_MAX']):

210 axis_map['AXIS_MIN'] = maybe_int(

211 pr.min_from_limit_max(axis_map['AXIS_LIMIT'], axis_map['AXIS_MAX'])

212 )

213 else:

214 info_queue.append(

215 f"NOK limit({axis_map['AXIS_LIMIT']}) and max({axis_map['AXIS_MAX']}) not both numeric,"

216 f" ignored {axis_map['AXIS_TYPE'].lower()} axis at index ({i_cfc})"

217 )

218 if axis_map['AXIS_VALUE'] != NULL_STR_REP and not is_numeric(axis_map['AXIS_VALUE']):

219 axis_map['AXIS_VALUE'] = NULL_STR_REP

220 else: # axis_map['AXIS_TYPE'] == 'FOLDED':

221 log.info(f'... claim of FOLDED is True? Axis map is ({axis_map})')

222 if is_numeric(axis_map['AXIS_LIMIT']) and is_numeric(axis_map['AXIS_MAX']):

223 axis_map['AXIS_MIN'] = maybe_int(

224 pr.min_from_limit_max(axis_map['AXIS_LIMIT'], axis_map['AXIS_MAX'])

225 )

226 axis_map['AXIS_LIMIT_FOLDED'] = maybe_int(

227 pr.limit_folded_from_limit_max(axis_map['AXIS_LIMIT'], axis_map['AXIS_MAX'])

228 )

229 axis_map['AXIS_MIN_FOLDED'] = maybe_int(

230 pr.min_folded_from_limit_max(axis_map['AXIS_LIMIT'], axis_map['AXIS_MAX'])

231 )

232 else:

233 info_queue.append(

234 f"NOK limit({axis_map['AXIS_LIMIT']}) and max({axis_map['AXIS_MAX']}) not both numeric,"

235 f' ignored folded axis at index ({i_cfc})'

236 )

237

238 some_axis_maps.append(axis_map)

239

240 n_axis_rows = len(some_axis_maps)

241 if n_axis_rows_req > n_axis_rows:

242 info_queue.append(

243 f'{n_axis_rows_req} dimensions requested, but only {n_axis_rows} accepted.'

244 f' Maximum is {NEEDED_NUMBER_OF_AXIS_MAX}'

245 )

246

247 best_effort_re_order_map = {}

248 collect_index_cand_list = []

249 for x, data in enumerate(some_axis_maps):

250 index_cand = data['AXIS_INDEX']

251 if is_numeric(index_cand):

252 i_cfc = int(index_cand)

253 else:

254 i_cfc = x

255 collect_index_cand_list.append(i_cfc)

256 if not is_numeric(index_cand) or i_cfc != index_cand or index_cand < 0 or index_cand >= n_axis_rows:

257 has_index_collision = True

258 conflict_reason = 'NO_INTEGER'

259 if i_cfc != index_cand:

260 conflict_reason = 'DC_INTEGER'

261

262 if is_numeric(index_cand) and index_cand < 0:

263 conflict_reason = 'LT_ZERO'

264

265 elif is_numeric(index_cand) and index_cand >= n_axis_rows:

266 conflict_reason = 'GT_NROW'

267

268 info_queue.append(

269 f'Conflicting index rules. Failing candidate is ({index_cand}), reason is {conflict_reason}'

270 )

271 some_axis_maps[x]['AXIS_INDEX'] = x

272

273 if index_cand != x:

274 has_index_order_mismatch = True

275 info_queue.append(f'Index positions not ordered. Misplaced candidate is {index_cand}, found at {x}')

276

277 best_effort_re_order_map[index_cand] = data

278

279 collect_index_cand_set = set(collect_index_cand_list)

280 if len(collect_index_cand_set) != len(collect_index_cand_list):

281 has_index_collision = True

282 hist = collections.Counter(collect_index_cand_list)

283 blame_list = []

284 for xx, nn in hist.items():

285 if nn != 1: 285 ↛ 284line 285 didn't jump to line 284, because the condition on line 285 was never false

286 blame_list.append(xx)

287

288 blame_list.sort()

289 info_queue.append(

290 f'Conflicting index positions. Failing candidate/s is/are [{", ".join(str(x) for x in blame_list)}],'

291 ' reason is NONUNIQUE_INDEX'

292 )

293

294 if not has_index_collision and has_index_order_mismatch:

295 some_axis_maps = []

296 for k, data in sorted(best_effort_re_order_map.items()):

297 some_axis_maps.append(data)

298

299 """

300 $normalizedInputDataRows = array();

301 foreach($someAxisMaps as $x => $data) {

302 $axisValues = array_values($data);

303 $normalizedInputDataRows[] = implode(";",$axisValues);

304 }

305 //DEBUG echo '<pre>ReAssembledRows:'."\n".print_r($normalizedInputDataRows,True).'</pre>';

306 $normalizedInputDataString = implode("\n",$normalizedInputDataRows);

307 //DEBUG echo '<pre>ReAssembledNormalizedInput:'."\n".print_r($normalizedInputDataString,True).'</pre>';

308 echo 'AxisSpecTest: '."\n";

309 echo '<form style="display:inline;" method="post" action="'.$_SERVER['PHP_SELF'].'">'."\n";

310 echo '<textarea style="font-sice:small;" cols="80" rows="16" name="AXIS_SPEC_ROWS">'.$normalizedInputDataString. \

311 </textarea>'."\n";

312 echo '<input type="submit" name="Subme" value="parse" />'."\n";

313 echo '</form>'.'<br />'."\n";

314 echo '[<a href="'.$_SERVER['PHP_SELF'].'">RESET</a>] to some default to get started.<br />'."\n";

315 //echo 'Implicit Keys: '.implode(';',$axisDefaultKeys).'<br />'."\n";

316 echo '<pre>';

317 echo 'Testing Module: '.$_SERVER['PHP_SELF']."\n";

318 $infoQueue[] = 'TestInput: AXIS_SPEC_ROWS=<pre>'."\n".$axisValuesRowsReqString.'</pre>'."\n";

319 echo ' TestOutput[0]:'."\n";

320 echo '$someAxisMaps='."\n";

321 //DEBUG echo print_r($someAxisMaps,True);

322 echo '</pre>';

323 echo '<table style="width:75%;"><tr><th>Laufd.Nr.</th><th>Name</th><th>Type</th><th>Min</th><th>Limit</th>\

324 <th>Max</th><th>LimitFolded</th><th>MinFolded</th><th>Value</th><th>Unit</th></tr>'."\n";

325 foreach($someAxisMaps as $i => $data) {

326 $displayRow = array_values($data);

327 echo '<tr><td>';

328 echo implode('</td><td>',$displayRow);

329 echo '</td></tr>';

330 }

331 echo '</table>'."\n";

332 if($infoQueue) {

333 echo '<h2>Info:</h2>';

334 echo '<ul><li>';

335 echo implode('</li><li>',$infoQueue);

336 echo '</li></ul>';

337 }

338 echo '</body>'."\n";

339 echo '</html>'."\n";

340 """

341 return some_axis_maps, info_queue

342

343

344@no_type_check

345def table(axis_maps, info_queue) -> str:

346 """Display axis maps as markdown table."""

347 normalized_input_data_rows = [SEMI.join(str(v) for v in axis.values()) for axis in axis_maps]

348 log.debug('ReAssembledRows:')

349 for row in normalized_input_data_rows:

350 log.debug(row)

351 normalized_input_data_string = '\n'.join(normalized_input_data_rows)

352 log.debug('ReAssembledNormalizedInput:')

353 log.debug(normalized_input_data_string)

354 table_headers = ('ListNo.', 'Name', 'Type', 'Min', 'Limit', 'Max', 'LimitFolded', 'MinFolded', 'Value', 'Unit')

355 table_header_str_row = PIPE.join(table_headers)

356 table_body_str_rows = []

357 for i, data in enumerate(axis_maps, start=1):

358 display_row = PIPE.join(str(d) for d in data.values())

359 table_body_str_rows.append(display_row)

360 if info_queue:

361 log.warning('InfoQueue:')

362 for diagnostic in info_queue:

363 log.warning(f'- {diagnostic}')

364 return '\n'.join([table_header_str_row] + table_body_str_rows)