Coverage for backend/django/Economics/scheduling/series.py: 85%

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from decimal import Decimal, InvalidOperation 

5from typing import TYPE_CHECKING 

6from django.db.models import Q 

7 

8from core.auxiliary.models.DataCell import DataCell 

9from core.auxiliary.models.DataColumn import DataColumn 

10from core.auxiliary.models.DataRow import DataRow 

11from core.auxiliary.models.PropertyInfo import PropertyInfo 

12from core.auxiliary.models.PropertyValue import PropertyValue 

13from core.auxiliary.models.Scenario import OptimizationDegreesOfFreedom, Scenario 

14from core.auxiliary.models.Solution import Solution 

15from Economics.scheduling.durations import ( 

16 AnnualizedCycleStep, 

17 annualize_cycle, 

18 annualized_active_duration_by_key, 

19 annualized_active_step_count, 

20 annualized_remainder_cycle_steps, 

21) 

22from Economics.scheduling.composite.compiler import ( 

23 active_composite_steps, 

24 annual_composite_traversal, 

25 composite_source_scenario_ids, 

26 saved_compiled_composite_schedule, 

27) 

28from Economics.scheduling.services import schedule_scenario_option 

29from Economics.settings_profiles.services.settings_profiles import get_settings_profile 

30from Economics.shared.choices import EconomicsScheduleMode 

31 

32if TYPE_CHECKING: 

33 from Economics.studies.models import EconomicsStudy 

34 

35 

36@dataclass(frozen=True) 

37class ScheduleValuePoint: 

38 """One row-aligned production-schedule value for a property.""" 

39 

40 row_index: int 

41 elapsed_hours: Decimal 

42 interval_hours: Decimal 

43 value: Decimal | None 

44 source: str 

45 source_scenario_id: int | None = None 

46 source_scenario_name: str = "" 

47 source_row_index: int | None = None 

48 source_rule_id: int | None = None 

49 source_rule_label: str = "" 

50 

51 

52@dataclass(frozen=True) 

53class ScheduleSeriesResolution: 

54 """Resolved schedule values plus diagnostics used by economics calculations.""" 

55 

56 scenario_id: int 

57 property_info_id: int 

58 property_name: str 

59 unit: str 

60 source: str 

61 schedule_varying: bool 

62 row_count: int 

63 points: tuple[ScheduleValuePoint, ...] 

64 message: str = "" 

65 

66 

67@dataclass(frozen=True) 

68class ScheduleRowDuration: 

69 """Annual operating duration assigned to one schedule row.""" 

70 

71 row_index: int 

72 annual_hours: Decimal 

73 source_scenario_id: int | None = None 

74 source_row_index: int | None = None 

75 source_rule_id: int | None = None 

76 source_rule_label: str = "" 

77 

78 

79@dataclass(frozen=True) 

80class ScheduleTimelineStep: 

81 """One timestep in the annualized production-schedule timeline.""" 

82 

83 index: int 

84 row_index: int 

85 elapsed_hours: Decimal 

86 duration_hours: Decimal 

87 

88 

89@dataclass(frozen=True) 

90class ScheduleTimelineBucketSegment: 

91 """Duration contributed by one source schedule row inside a grouped timeline bucket.""" 

92 

93 row_index: int 

94 duration_hours: Decimal 

95 

96 

97@dataclass(frozen=True) 

98class ScheduleTimelineBucket: 

99 """A bounded timeline bucket used when charting dense annual schedules.""" 

100 

101 index: int 

102 row_index: int 

103 elapsed_hours: Decimal 

104 duration_hours: Decimal 

105 segments: tuple[ScheduleTimelineBucketSegment, ...] 

106 

107 

108def property_is_schedule_varying(scenario: Scenario, property_info: PropertyInfo) -> bool: 

109 """Return whether a property should use production-schedule behavior.""" 

110 

111 if _input_column_for_property(scenario=scenario, property_info=property_info) is not None: 

112 return True 

113 if _is_optimization_dof(scenario=scenario, property_info=property_info): 113 ↛ 114line 113 didn't jump to line 114 because the condition on line 113 was never true

114 return True 

115 return not _is_process_dof(property_info) 

116 

117 

118def study_property_is_schedule_varying(*, study: EconomicsStudy, property_info: PropertyInfo) -> bool: 

119 """Return whether a property varies over the study's active schedule.""" 

120 

121 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 121 ↛ 123line 121 didn't jump to line 123 because the condition on line 121 was always true

122 return property_is_schedule_varying(study.schedule_scenario, property_info) 

123 if study.schedule_mode != EconomicsScheduleMode.COMPOSITE: 

124 return False 

125 source_ids = composite_source_scenario_ids(saved_compiled_composite_schedule(study)) 

126 if not source_ids: 

127 return False 

128 scenarios = Scenario.objects.filter(pk__in=source_ids, flowsheet_state=study.flowsheet_state) 

129 return any(property_is_schedule_varying(scenario, property_info) for scenario in scenarios) 

130 

131 

132def study_schedule_varying_property_ids( 

133 *, 

134 study: EconomicsStudy, 

135 property_ids: set[int], 

136) -> set[int]: 

137 """Classify many properties against a study schedule with bounded queries.""" 

138 if not property_ids: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 return set() 

140 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

141 scenario_ids = {study.schedule_scenario_id} 

142 elif study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true

143 scenario_ids = set(composite_source_scenario_ids(saved_compiled_composite_schedule(study))) 

144 else: 

145 return set() 

146 if not scenario_ids: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true

147 return set() 

148 

149 return _schedule_varying_property_ids( 

150 flowsheet_state=study.flowsheet_state, 

151 scenario_ids=scenario_ids, 

152 property_ids=property_ids, 

153 ) 

154 

155 

156def _schedule_varying_property_ids( 

157 *, 

158 flowsheet_state, 

159 scenario_ids: set[int], 

160 property_ids: set[int], 

161) -> set[int]: 

162 """Classify properties for an already-resolved set of schedule scenarios.""" 

163 

164 input_property_ids = set( 

165 DataColumn.objects.filter( 

166 flowsheet_state=flowsheet_state, 

167 scenario_id__in=scenario_ids, 

168 property_value__property_id__in=property_ids, 

169 ).values_list("property_value__property_id", flat=True) 

170 ) 

171 optimization_property_ids = set( 

172 OptimizationDegreesOfFreedom.objects.filter( 

173 flowsheet_state=flowsheet_state, 

174 scenario_id__in=scenario_ids, 

175 propertyValue__property_id__in=property_ids, 

176 ).values_list("propertyValue__property_id", flat=True) 

177 ) 

178 process_dof_property_ids = set( 

179 PropertyValue.objects.filter( 

180 flowsheet_state=flowsheet_state, 

181 property_id__in=property_ids, 

182 ) 

183 .filter( 

184 Q(enabled=True, controlManipulated__isnull=True) 

185 | Q(controlSetPoint__isnull=False) 

186 ) 

187 .values_list("property_id", flat=True) 

188 ) 

189 return input_property_ids | optimization_property_ids | (property_ids - process_dof_property_ids) 

190 

191 

192def study_schedule_property_resolution( 

193 *, 

194 study: EconomicsStudy, 

195 property_info: PropertyInfo, 

196) -> ScheduleSeriesResolution: 

197 """Resolve a property against either a single or composite study schedule.""" 

198 

199 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

200 return property_schedule_resolution(scenario=study.schedule_scenario, property_info=property_info) 

201 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 

202 return _composite_property_schedule_resolution(study=study, property_info=property_info) 

203 return _empty_study_resolution( 

204 study=study, 

205 property_info=property_info, 

206 schedule_varying=False, 

207 message="This study does not use a production schedule.", 

208 ) 

209 

210 

211def study_schedule_property_resolutions( 

212 *, 

213 study: EconomicsStudy, 

214 property_infos: list[PropertyInfo] | tuple[PropertyInfo, ...], 

215) -> dict[int, ScheduleSeriesResolution]: 

216 """Resolve many properties while sharing the active schedule read. 

217 

218 Scenario schedules are the common bulk-costing path, so their rows, input 

219 cells, and output solutions are loaded once. Composite resolution retains 

220 its existing traversal semantics and delegates property-by-property. 

221 """ 

222 

223 unique_properties = {property_info.pk: property_info for property_info in property_infos} 

224 if not unique_properties: 

225 return {} 

226 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

227 return property_schedule_resolutions( 

228 scenario=study.schedule_scenario, 

229 property_infos=tuple(unique_properties.values()), 

230 ) 

231 return { 

232 property_info_id: study_schedule_property_resolution( 

233 study=study, 

234 property_info=property_info, 

235 ) 

236 for property_info_id, property_info in unique_properties.items() 

237 } 

238 

239 

240def property_schedule_resolution( 

241 *, 

242 scenario: Scenario, 

243 property_info: PropertyInfo, 

244) -> ScheduleSeriesResolution: 

245 """Resolve a property to row-aligned production-schedule values.""" 

246 

247 option = schedule_scenario_option(scenario) 

248 schedule_varying = property_is_schedule_varying(scenario, property_info) 

249 if option is None: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 return _empty_resolution( 

251 scenario=scenario, 

252 property_info=property_info, 

253 schedule_varying=schedule_varying, 

254 message="Run this production schedule before using it for economics.", 

255 ) 

256 if not schedule_varying: 

257 return _empty_resolution( 

258 scenario=scenario, 

259 property_info=property_info, 

260 schedule_varying=False, 

261 message="This property uses a steady value for this production schedule.", 

262 ) 

263 

264 rows = _schedule_rows(scenario) 

265 input_column = _input_column_for_property(scenario=scenario, property_info=property_info) 

266 if input_column is not None: 

267 values_by_row = _input_values_by_row(input_column=input_column) 

268 source = "input" 

269 else: 

270 values_by_row = _output_values_by_row(scenario=scenario, property_info=property_info) 

271 source = "output" 

272 

273 interval_hours = option.interval_hours 

274 points = tuple( 

275 ScheduleValuePoint( 

276 row_index=row.index, 

277 elapsed_hours=Decimal(position) * interval_hours, 

278 interval_hours=interval_hours, 

279 value=_decimal_or_none(values_by_row.get(row.index)), 

280 source=source, 

281 source_scenario_id=scenario.pk, 

282 source_scenario_name=scenario.displayName or "Production schedule", 

283 source_row_index=row.index, 

284 ) 

285 for position, row in enumerate(rows) 

286 ) 

287 missing_count = sum(1 for point in points if point.value is None) 

288 message = "" 

289 if missing_count: 

290 message = "This property is not available for every operating state in the selected production schedule." 

291 return ScheduleSeriesResolution( 

292 scenario_id=scenario.pk, 

293 property_info_id=property_info.pk, 

294 property_name=property_info.displayName, 

295 unit=property_info.unit, 

296 source=source, 

297 schedule_varying=True, 

298 row_count=len(rows), 

299 points=points, 

300 message=message, 

301 ) 

302 

303 

304def property_schedule_resolutions( 

305 *, 

306 scenario: Scenario, 

307 property_infos: tuple[PropertyInfo, ...], 

308) -> dict[int, ScheduleSeriesResolution]: 

309 """Resolve multiple properties from one scenario query graph.""" 

310 

311 properties_by_id = {property_info.pk: property_info for property_info in property_infos} 

312 if not properties_by_id: 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true

313 return {} 

314 

315 option = schedule_scenario_option(scenario) 

316 varying_ids = _schedule_varying_property_ids( 

317 flowsheet_state=scenario.flowsheet_state, 

318 scenario_ids={scenario.pk}, 

319 property_ids=set(properties_by_id), 

320 ) 

321 if option is None: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true

322 return { 

323 property_info_id: _empty_resolution( 

324 scenario=scenario, 

325 property_info=property_info, 

326 schedule_varying=property_info_id in varying_ids, 

327 message="Run this production schedule before using it for economics.", 

328 ) 

329 for property_info_id, property_info in properties_by_id.items() 

330 } 

331 

332 rows = _schedule_rows(scenario) 

333 input_values = _scenario_input_values_by_property( 

334 scenario=scenario, 

335 property_ids=set(properties_by_id), 

336 ) 

337 output_values = _scenario_output_values_by_property( 

338 scenario=scenario, 

339 property_ids=set(properties_by_id) - set(input_values), 

340 ) 

341 return { 

342 property_info_id: _batched_property_schedule_resolution( 

343 scenario=scenario, 

344 property_info=property_info, 

345 rows=rows, 

346 interval_hours=option.interval_hours, 

347 schedule_varying=property_info_id in varying_ids, 

348 input_values=input_values, 

349 output_values=output_values, 

350 ) 

351 for property_info_id, property_info in properties_by_id.items() 

352 } 

353 

354 

355def _scenario_input_values_by_property( 

356 *, 

357 scenario: Scenario, 

358 property_ids: set[int], 

359) -> dict[int, dict[int, float | None]]: 

360 """Load the first input column and row values for every requested property.""" 

361 

362 input_columns: dict[int, DataColumn] = {} 

363 columns = ( 

364 DataColumn.objects.filter( 

365 flowsheet_state=scenario.flowsheet_state, 

366 scenario=scenario, 

367 property_value__property_id__in=property_ids, 

368 ) 

369 .select_related("property_value") 

370 .order_by("property_value__property_id", "created_at", "pk") 

371 ) 

372 for column in columns: 

373 input_columns.setdefault(column.property_value.property_id, column) 

374 

375 input_values: dict[int, dict[int, float | None]] = { 

376 property_info_id: {} for property_info_id in input_columns 

377 } 

378 selected_column_ids = {column.pk for column in input_columns.values()} 

379 cells = ( 

380 DataCell.objects.filter( 

381 data_column_id__in=selected_column_ids, 

382 data_row__scenario=scenario, 

383 ) 

384 .select_related("data_row", "data_column__property_value") 

385 .order_by("data_column__property_value__property_id", "data_row__index", "pk") 

386 ) 

387 for cell in cells: 

388 if cell.data_row_id is not None: 388 ↛ 387line 388 didn't jump to line 387 because the condition on line 388 was always true

389 property_info_id = cell.data_column.property_value.property_id 

390 input_values[property_info_id][cell.data_row.index] = cell.value 

391 return input_values 

392 

393 

394def _scenario_output_values_by_property( 

395 *, 

396 scenario: Scenario, 

397 property_ids: set[int], 

398) -> dict[int, dict[int, float | None]]: 

399 """Load the first solved value for each requested property and schedule row.""" 

400 

401 output_values: dict[int, dict[int, float | None]] = { 

402 property_info_id: {} for property_info_id in property_ids 

403 } 

404 solutions = ( 

405 Solution.objects.filter( 

406 flowsheet_state=scenario.flowsheet_state, 

407 scenario=scenario, 

408 property__property_id__in=output_values, 

409 solve_index__isnull=False, 

410 ) 

411 .select_related("property") 

412 .order_by("property__property_id", "solve_index", "pk") 

413 ) 

414 for solution in solutions: 

415 values_by_row = output_values[solution.property.property_id] 

416 values_by_row.setdefault( 

417 solution.solve_index, 

418 solution.values[0] if solution.values else None, 

419 ) 

420 return output_values 

421 

422 

423def _batched_property_schedule_resolution( 

424 *, 

425 scenario: Scenario, 

426 property_info: PropertyInfo, 

427 rows: list[DataRow], 

428 interval_hours: Decimal, 

429 schedule_varying: bool, 

430 input_values: dict[int, dict[int, float | None]], 

431 output_values: dict[int, dict[int, float | None]], 

432) -> ScheduleSeriesResolution: 

433 """Build one resolution from the shared scenario query results.""" 

434 

435 if not schedule_varying: 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true

436 return _empty_resolution( 

437 scenario=scenario, 

438 property_info=property_info, 

439 schedule_varying=False, 

440 message="This property uses a steady value for this production schedule.", 

441 ) 

442 source = "input" if property_info.pk in input_values else "output" 

443 values_by_row = ( 

444 input_values[property_info.pk] 

445 if source == "input" 

446 else output_values[property_info.pk] 

447 ) 

448 points = tuple( 

449 ScheduleValuePoint( 

450 row_index=row.index, 

451 elapsed_hours=Decimal(position) * interval_hours, 

452 interval_hours=interval_hours, 

453 value=_decimal_or_none(values_by_row.get(row.index)), 

454 source=source, 

455 source_scenario_id=scenario.pk, 

456 source_scenario_name=scenario.displayName or "Production schedule", 

457 source_row_index=row.index, 

458 ) 

459 for position, row in enumerate(rows) 

460 ) 

461 missing_count = sum(1 for point in points if point.value is None) 

462 return ScheduleSeriesResolution( 

463 scenario_id=scenario.pk, 

464 property_info_id=property_info.pk, 

465 property_name=property_info.displayName, 

466 unit=property_info.unit, 

467 source=source, 

468 schedule_varying=True, 

469 row_count=len(rows), 

470 points=points, 

471 message=( 

472 "This property is not available for every operating state in the selected production schedule." 

473 if missing_count 

474 else "" 

475 ), 

476 ) 

477 

478 

479def study_schedule_row_durations(*, study: EconomicsStudy) -> tuple[ScheduleRowDuration, ...]: 

480 """Return annual durations for the rows in the study's active schedule.""" 

481 

482 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

483 return annual_schedule_row_durations(study=study, scenario=study.schedule_scenario) 

484 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 484 ↛ 486line 484 didn't jump to line 486 because the condition on line 484 was always true

485 return _composite_annual_schedule_row_durations(study=study) 

486 return () 

487 

488 

489def annual_schedule_row_durations( 

490 *, 

491 study: EconomicsStudy, 

492 scenario: Scenario, 

493) -> tuple[ScheduleRowDuration, ...]: 

494 """Return per-row annual durations, prorating the final partial cycle.""" 

495 

496 option = schedule_scenario_option(scenario) 

497 profile = get_settings_profile(study) 

498 annual_operating_hours = profile.annual_operating_hours if profile is not None else None 

499 if option is None or annual_operating_hours is None or annual_operating_hours <= 0: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true

500 return () 

501 

502 rows = _schedule_rows(scenario) 

503 if not rows: 503 ↛ 504line 503 didn't jump to line 504 because the condition on line 503 was never true

504 return () 

505 

506 durations_by_row = annualized_active_duration_by_key( 

507 cycle_steps=rows, 

508 duration_accessor=lambda _row: option.interval_hours, 

509 active_step_predicate=lambda _row: True, 

510 key_accessor=lambda row: row.index, 

511 annual_operating_hours=annual_operating_hours, 

512 ) 

513 

514 durations: list[ScheduleRowDuration] = [] 

515 for row in rows: 

516 durations.append( 

517 ScheduleRowDuration( 

518 row_index=row.index, 

519 annual_hours=durations_by_row.get(row.index, Decimal("0")), 

520 source_scenario_id=scenario.pk, 

521 source_row_index=row.index, 

522 ) 

523 ) 

524 return tuple(durations) 

525 

526 

527def study_schedule_timeline_steps(*, study: EconomicsStudy) -> tuple[ScheduleTimelineStep, ...]: 

528 """Expand the study's active schedule over annual operating hours.""" 

529 

530 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

531 return annual_schedule_timeline_steps(study=study, scenario=study.schedule_scenario) 

532 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 

533 return _composite_annual_schedule_timeline_steps(study=study) 

534 return () 

535 

536 

537def annual_schedule_timeline_steps( 

538 *, 

539 study: EconomicsStudy, 

540 scenario: Scenario, 

541) -> tuple[ScheduleTimelineStep, ...]: 

542 """Expand a compatible production schedule over annual operating hours.""" 

543 

544 option = schedule_scenario_option(scenario) 

545 profile = get_settings_profile(study) 

546 annual_operating_hours = profile.annual_operating_hours if profile is not None else None 

547 if option is None or annual_operating_hours is None or annual_operating_hours <= 0: 

548 return () 

549 

550 rows = _schedule_rows(scenario) 

551 if not rows: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true

552 return () 

553 

554 annualized_steps = annualize_cycle( 

555 cycle_steps=rows, 

556 duration_accessor=lambda _row: option.interval_hours, 

557 active_step_predicate=lambda _row: True, 

558 annual_operating_hours=annual_operating_hours, 

559 ) 

560 return tuple( 

561 ScheduleTimelineStep( 

562 index=step_index, 

563 row_index=step.source.index, 

564 elapsed_hours=step.elapsed_hours, 

565 duration_hours=step.duration_hours, 

566 ) 

567 for step_index, step in enumerate(annualized_steps) 

568 ) 

569 

570 

571def _scenario_annualized_active_step_count( 

572 *, 

573 rows: list[DataRow], 

574 interval_hours: Decimal, 

575 annual_operating_hours: Decimal, 

576) -> int: 

577 return annualized_active_step_count( 

578 cycle_steps=rows, 

579 duration_accessor=lambda _row: interval_hours, 

580 active_step_predicate=lambda _row: True, 

581 annual_operating_hours=annual_operating_hours, 

582 ) 

583 

584 

585def _scenario_annualized_remainder_steps( 

586 *, 

587 rows: list[DataRow], 

588 interval_hours: Decimal, 

589 annual_operating_hours: Decimal, 

590) -> tuple[AnnualizedCycleStep[DataRow], ...]: 

591 return annualized_remainder_cycle_steps( 

592 cycle_steps=rows, 

593 duration_accessor=lambda _row: interval_hours, 

594 active_step_predicate=lambda _row: True, 

595 annual_operating_hours=annual_operating_hours, 

596 ) 

597 

598 

599def study_schedule_timeline_buckets( 

600 *, 

601 study: EconomicsStudy, 

602 bucket_count: int, 

603) -> tuple[ScheduleTimelineBucket, ...]: 

604 """Return bounded active-schedule timeline buckets without expanding every step.""" 

605 

606 if bucket_count <= 0: 606 ↛ 607line 606 didn't jump to line 607 because the condition on line 606 was never true

607 return () 

608 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

609 return annual_schedule_timeline_buckets( 

610 study=study, 

611 scenario=study.schedule_scenario, 

612 bucket_count=bucket_count, 

613 ) 

614 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 614 ↛ 616line 614 didn't jump to line 616 because the condition on line 614 was always true

615 return _composite_annual_schedule_timeline_buckets(study=study, bucket_count=bucket_count) 

616 return () 

617 

618 

619def annual_schedule_timeline_buckets( 

620 *, 

621 study: EconomicsStudy, 

622 scenario: Scenario, 

623 bucket_count: int, 

624) -> tuple[ScheduleTimelineBucket, ...]: 

625 """Group a scenario schedule timeline directly from step ranges.""" 

626 

627 option = schedule_scenario_option(scenario) 

628 profile = get_settings_profile(study) 

629 annual_operating_hours = profile.annual_operating_hours if profile is not None else None 

630 if ( 630 ↛ 637line 630 didn't jump to line 637 because the condition on line 630 was never true

631 option is None 

632 or annual_operating_hours is None 

633 or annual_operating_hours <= 0 

634 or option.interval_hours <= 0 

635 or bucket_count <= 0 

636 ): 

637 return () 

638 

639 rows = _schedule_rows(scenario) 

640 if not rows: 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true

641 return () 

642 

643 step_count = _scenario_annualized_active_step_count( 

644 rows=rows, 

645 interval_hours=option.interval_hours, 

646 annual_operating_hours=annual_operating_hours, 

647 ) 

648 if step_count <= 0: 648 ↛ 649line 648 didn't jump to line 649 because the condition on line 648 was never true

649 return () 

650 

651 bucket_total = min(bucket_count, step_count) 

652 final_step_index = step_count - 1 

653 final_remainder_steps = _scenario_annualized_remainder_steps( 

654 rows=rows, 

655 interval_hours=option.interval_hours, 

656 annual_operating_hours=annual_operating_hours, 

657 ) 

658 final_step_duration = ( 

659 final_remainder_steps[-1].duration_hours 

660 if final_remainder_steps 

661 else option.interval_hours 

662 ) 

663 buckets: list[ScheduleTimelineBucket] = [] 

664 for bucket_index in range(bucket_total): 

665 start_step = bucket_index * step_count // bucket_total 

666 end_step = (bucket_index + 1) * step_count // bucket_total 

667 if start_step >= end_step: 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true

668 continue 

669 elapsed_hours = min(Decimal(start_step) * option.interval_hours, annual_operating_hours) 

670 end_elapsed_hours = min(Decimal(end_step) * option.interval_hours, annual_operating_hours) 

671 duration_hours = end_elapsed_hours - elapsed_hours 

672 if duration_hours <= 0: 672 ↛ 673line 672 didn't jump to line 673 because the condition on line 672 was never true

673 continue 

674 segments = _scenario_bucket_segments( 

675 rows=rows, 

676 start_step=start_step, 

677 end_step=end_step, 

678 interval_hours=option.interval_hours, 

679 final_step_index=final_step_index, 

680 final_step_duration=final_step_duration, 

681 ) 

682 buckets.append( 

683 ScheduleTimelineBucket( 

684 index=len(buckets), 

685 row_index=rows[(end_step - 1) % len(rows)].index, 

686 elapsed_hours=elapsed_hours, 

687 duration_hours=duration_hours, 

688 segments=segments, 

689 ) 

690 ) 

691 return tuple(buckets) 

692 

693 

694def study_schedule_timeline_step_count(*, study: EconomicsStudy) -> int: 

695 """Return an annualized active schedule step count for the study.""" 

696 

697 if study.schedule_mode == EconomicsScheduleMode.SCENARIO and study.schedule_scenario_id: 

698 return annual_schedule_timeline_step_count(study=study, scenario=study.schedule_scenario) 

699 if study.schedule_mode == EconomicsScheduleMode.COMPOSITE: 

700 return _composite_annual_schedule_timeline_step_count(study=study) 

701 return 0 

702 

703 

704def annual_schedule_timeline_step_count( 

705 *, 

706 study: EconomicsStudy, 

707 scenario: Scenario, 

708) -> int: 

709 """Return the annualized timeline step count without expanding the timeline.""" 

710 

711 option = schedule_scenario_option(scenario) 

712 profile = get_settings_profile(study) 

713 annual_operating_hours = profile.annual_operating_hours if profile is not None else None 

714 if ( 

715 option is None 

716 or annual_operating_hours is None 

717 or annual_operating_hours <= 0 

718 or option.interval_hours <= 0 

719 ): 

720 return 0 

721 rows = _schedule_rows(scenario) 

722 if not rows: 722 ↛ 723line 722 didn't jump to line 723 because the condition on line 722 was never true

723 return 0 

724 return _scenario_annualized_active_step_count( 

725 rows=rows, 

726 interval_hours=option.interval_hours, 

727 annual_operating_hours=annual_operating_hours, 

728 ) 

729 

730 

731def _scenario_bucket_segments( 

732 *, 

733 rows: list[DataRow], 

734 start_step: int, 

735 end_step: int, 

736 interval_hours: Decimal, 

737 final_step_index: int, 

738 final_step_duration: Decimal, 

739) -> tuple[ScheduleTimelineBucketSegment, ...]: 

740 """Aggregate source-row durations for a bounded scenario timeline range.""" 

741 

742 row_count = len(rows) 

743 segments = [] 

744 for position, row in enumerate(rows): 

745 step_count = _cyclic_index_count( 

746 start=start_step, 

747 end=end_step, 

748 position=position, 

749 cycle_length=row_count, 

750 ) 

751 if step_count <= 0: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true

752 continue 

753 duration_hours = Decimal(step_count) * interval_hours 

754 if start_step <= final_step_index < end_step and final_step_index % row_count == position: 

755 duration_hours -= interval_hours - final_step_duration 

756 if duration_hours > 0: 756 ↛ 744line 756 didn't jump to line 744 because the condition on line 756 was always true

757 segments.append(ScheduleTimelineBucketSegment(row_index=row.index, duration_hours=duration_hours)) 

758 return tuple(segments) 

759 

760 

761def _cyclic_index_count(*, start: int, end: int, position: int, cycle_length: int) -> int: 

762 """Count indexes in ``[start, end)`` whose cycle position matches ``position``.""" 

763 

764 if end <= start or cycle_length <= 0: 764 ↛ 765line 764 didn't jump to line 765 because the condition on line 764 was never true

765 return 0 

766 first = start + ((position - start) % cycle_length) 

767 if first >= end: 767 ↛ 768line 767 didn't jump to line 768 because the condition on line 767 was never true

768 return 0 

769 return ((end - 1 - first) // cycle_length) + 1 

770 

771 

772def _composite_property_schedule_resolution( 

773 *, 

774 study: EconomicsStudy, 

775 property_info: PropertyInfo, 

776) -> ScheduleSeriesResolution: 

777 compiled = saved_compiled_composite_schedule(study) 

778 if compiled.status != "valid": 778 ↛ 779line 778 didn't jump to line 779 because the condition on line 778 was never true

779 return _empty_study_resolution( 

780 study=study, 

781 property_info=property_info, 

782 schedule_varying=True, 

783 message=compiled.diagnostics[0].message if compiled.diagnostics else "Complete the composite schedule.", 

784 ) 

785 

786 active_steps = active_composite_steps(compiled) 

787 if not active_steps: 787 ↛ 788line 787 didn't jump to line 788 because the condition on line 787 was never true

788 return _empty_study_resolution( 

789 study=study, 

790 property_info=property_info, 

791 schedule_varying=False, 

792 message="This composite schedule has no active source rows.", 

793 ) 

794 

795 source_ids = composite_source_scenario_ids(compiled) 

796 scenarios = { 

797 scenario.pk: scenario 

798 for scenario in Scenario.objects.filter(pk__in=source_ids, flowsheet_state=study.flowsheet_state) 

799 } 

800 source_resolutions = { 

801 scenario_id: property_schedule_resolution(scenario=scenario, property_info=property_info) 

802 for scenario_id, scenario in scenarios.items() 

803 } 

804 schedule_varying = any(resolution.schedule_varying for resolution in source_resolutions.values()) 

805 if not schedule_varying: 805 ↛ 806line 805 didn't jump to line 806 because the condition on line 805 was never true

806 return _empty_study_resolution( 

807 study=study, 

808 property_info=property_info, 

809 schedule_varying=False, 

810 message="This property uses a steady value for this composite schedule.", 

811 ) 

812 

813 steady_value = _steady_property_decimal_or_none(property_info) 

814 source_points_by_row = { 

815 scenario_id: { 

816 point.source_row_index if point.source_row_index is not None else point.row_index: point 

817 for point in resolution.points 

818 } 

819 for scenario_id, resolution in source_resolutions.items() 

820 } 

821 traversal = annual_composite_traversal(study=study, compiled=compiled) 

822 annual_durations_by_step = traversal.durations_by_step() if traversal is not None else {} 

823 points: list[ScheduleValuePoint] = [] 

824 for step in active_steps: 

825 source_id = step.source_scenario 

826 if source_id is None: 826 ↛ 827line 826 didn't jump to line 827 because the condition on line 826 was never true

827 continue 

828 resolution = source_resolutions.get(source_id) 

829 value = steady_value 

830 source = "steady" 

831 if resolution is not None and resolution.schedule_varying: 831 ↛ 835line 831 didn't jump to line 835 because the condition on line 831 was always true

832 source_point = source_points_by_row.get(source_id, {}).get(step.source_row_index)