test_offset_converter.test_custom_properties()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 14
Code Lines 11

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 11
dl 0
loc 14
rs 9.85
c 0
b 0
f 0
cc 1
nop 0
1
import warnings
2
3
import numpy as np
4
import pytest
5
from oemof.tools.debugging import ExperimentalFeatureWarning
6
7
from oemof import solph
8
from oemof.solph._plumbing import sequence
9
from oemof.solph.components._offset_converter import (
10
    slope_offset_from_nonconvex_input,
11
)
12
from oemof.solph.components._offset_converter import (
13
    slope_offset_from_nonconvex_output,
14
)
15
16
17
def create_energysystem_stub(num_in, num_out):
18
    es = solph.EnergySystem(
19
        timeindex=solph.create_time_index(year=2023, number=9),
20
        infer_last_interval=True,
21
    )
22
23
    for i in range(num_in):
24
        bus_label = f"bus input {i}"
25
        b = solph.Bus(bus_label)
26
        es.add(b)
27
        es.add(
28
            solph.components.Source(f"source {i}", outputs={b: solph.Flow()})
29
        )
30
31
    for o in range(num_out):
32
        bus_label = f"bus output {o}"
33
        b = solph.Bus(bus_label)
34
        es.add(b)
35
        es.add(solph.components.Sink(f"sink {o}", inputs={b: solph.Flow()}))
36
37
    return es
38
39
40
def solve_and_extract_results(es):
41
    model = solph.Model(es)
42
    model.solve("cbc")
43
    results = solph.views.convert_keys_to_strings(model.results())
44
45
    assert (
46
        model.solver_results["Solver"][0]["Termination condition"]
47
        != "infeasible"
48
    )
49
    return results
50
51
52
def check_results(
53
    results,
54
    reference_bus,
55
    nominal_capacity,
56
    minimal_value,
57
    eta_at_nom,
58
    eta_at_min,
59
):
60
    for bus in eta_at_nom:
61
        if "input" in reference_bus.label:
62
            slope, offset = slope_offset_from_nonconvex_input(
63
                1,
64
                minimal_value / nominal_capacity,
65
                eta_at_nom[bus],
66
                eta_at_min[bus],
67
            )
68
            reference_flow = results[reference_bus.label, "offset converter"][
69
                "sequences"
70
            ]["flow"]
71
            reference_flow_status = results[
72
                reference_bus.label, "offset converter"
73
            ]["sequences"]["status"]
74
        else:
75
            slope, offset = slope_offset_from_nonconvex_output(
76
                1,
77
                minimal_value / nominal_capacity,
78
                eta_at_nom[bus],
79
                eta_at_min[bus],
80
            )
81
            reference_flow = results["offset converter", reference_bus.label][
82
                "sequences"
83
            ]["flow"]
84
            reference_flow_status = results[
85
                "offset converter", reference_bus.label
86
            ]["sequences"]["status"]
87
88
        flow_expected = (
89
            offset * nominal_capacity * reference_flow_status
90
            + slope * reference_flow
91
        )
92
        if "input" in bus.label:
93
            flow_actual = results[bus.label, "offset converter"]["sequences"][
94
                "flow"
95
            ]
96
        else:
97
            flow_actual = results["offset converter", bus.label]["sequences"][
98
                "flow"
99
            ]
100
101
        np.testing.assert_array_almost_equal(flow_actual, flow_expected)
102
103
104
def add_OffsetConverter(
105
    es, reference_bus, nominal_capacity, minimal_value, eta_at_nom, eta_at_min
106
):
107
    # Use of experimental API to access nodes by label.
108
    # Can be removed with future release of network.
109
    with warnings.catch_warnings():
110
        warnings.simplefilter("ignore", ExperimentalFeatureWarning)
111
        oc_inputs = {
112
            b: solph.Flow()
113
            for label, b in es.node.items()
114
            if "bus input" in label
115
        }
116
        oc_outputs = {
117
            b: solph.Flow()
118
            for label, b in es.node.items()
119
            if "bus output" in label
120
        }
121
122
        if reference_bus in oc_outputs:
123
            f = oc_outputs[reference_bus]
124
            get_slope_and_offset = slope_offset_from_nonconvex_output
125
            fix = [0] + np.linspace(
126
                minimal_value, nominal_capacity, 9
127
            ).tolist()
128
        else:
129
            f = oc_inputs[reference_bus]
130
            get_slope_and_offset = slope_offset_from_nonconvex_input
131
            fix = [0] + np.linspace(
132
                minimal_value * eta_at_min[es.node["bus output 0"]],
133
                nominal_capacity * eta_at_nom[es.node["bus output 0"]],
134
                9,
135
            ).tolist()
136
137
        fix_flow = es.flows()[es.node["bus output 0"], es.node["sink 0"]]
138
        fix_flow.fix = fix
139
        fix_flow.nominal_capacity = 1
140
141
        slopes = {}
142
        offsets = {}
143
144
        for bus in list(oc_inputs) + list(oc_outputs):
145
            if bus == reference_bus:
146
                continue
147
            slope, offset = get_slope_and_offset(
148
                1,
149
                minimal_value / nominal_capacity,
150
                eta_at_nom[bus],
151
                eta_at_min[bus],
152
            )
153
            slopes[bus] = slope
154
            offsets[bus] = offset
155
156
        f.nonconvex = solph.NonConvex()
157
        f.nominal_capacity = nominal_capacity
158
        f.min = sequence(minimal_value / nominal_capacity)
159
160
        oc = solph.components.OffsetConverter(
161
            label="offset converter",
162
            inputs=oc_inputs,
163
            outputs=oc_outputs,
164
            conversion_factors=slopes,
165
            normed_offsets=offsets,
166
        )
167
168
        es.add(oc)
169
170
171
def test_custom_properties():
172
    bus1 = solph.Bus()
173
    bus2 = solph.Bus()
174
    oc = solph.components.OffsetConverter(
175
        inputs={
176
            bus1: solph.Flow(nominal_capacity=2, nonconvex=solph.NonConvex())
177
        },
178
        outputs={bus2: solph.Flow()},
179
        conversion_factors={bus2: 2},
180
        normed_offsets={bus2: -0.5},
181
        custom_properties={"foo": "bar"},
182
    )
183
184
    assert oc.custom_properties["foo"] == "bar"
185
186
187 View Code Duplication
def test_invalid_conversion_factor():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
188
    bus1 = solph.Bus()
189
    bus2 = solph.Bus()
190
    with pytest.raises(ValueError, match="Conversion factors cannot be "):
191
        solph.components.OffsetConverter(
192
            inputs={
193
                bus1: solph.Flow(
194
                    nominal_capacity=2, nonconvex=solph.NonConvex()
195
                )
196
            },
197
            outputs={bus2: solph.Flow()},
198
            conversion_factors={
199
                bus1: 1,
200
                bus2: 2,
201
            },
202
            normed_offsets={bus2: -0.5},
203
        )
204
205
206 View Code Duplication
def test_invalid_normed_offset():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
207
    bus1 = solph.Bus()
208
    bus2 = solph.Bus()
209
    with pytest.raises(ValueError, match="Normed offsets cannot be "):
210
        solph.components.OffsetConverter(
211
            inputs={
212
                bus1: solph.Flow(
213
                    nominal_capacity=2, nonconvex=solph.NonConvex()
214
                )
215
            },
216
            outputs={bus2: solph.Flow()},
217
            conversion_factors={
218
                bus2: 2,
219
            },
220
            normed_offsets={
221
                bus1: -0.2,
222
                bus2: -0.5,
223
            },
224
        )
225
226
227
def test_wrong_number_of_coefficients():
228
    bus1 = solph.Bus()
229
    bus2 = solph.Bus()
230
    with pytest.raises(ValueError, match="Two coefficients"):
231
        solph.components.OffsetConverter(
232
            inputs={
233
                bus1: solph.Flow(
234
                    nominal_capacity=2, nonconvex=solph.NonConvex()
235
                )
236
            },
237
            outputs={bus2: solph.Flow()},
238
            coefficients=(1, 2, 3),
239
        )
240
241
242 View Code Duplication
def test_OffsetConverter_single_input_output_ref_output():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
243
    num_in = 1
244
    num_out = 1
245
    es = create_energysystem_stub(num_in, num_out)
246
247
    nominal_capacity = 10
248
    minimal_value = 3
249
250
    eta_at_nom = {es.groups["bus input 0"]: 0.7}
251
    eta_at_min = {es.groups["bus input 0"]: 0.5}
252
253
    add_OffsetConverter(
254
        es,
255
        es.groups["bus output 0"],
256
        nominal_capacity,
257
        minimal_value,
258
        eta_at_nom,
259
        eta_at_min,
260
    )
261
262
    results = solve_and_extract_results(es)
263
264
    check_results(
265
        results,
266
        es.groups["bus output 0"],
267
        nominal_capacity,
268
        minimal_value,
269
        eta_at_nom,
270
        eta_at_min,
271
    )
272
273
274 View Code Duplication
def test_OffsetConverter_single_input_output_ref_output_eta_decreasing():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
275
    num_in = 1
276
    num_out = 1
277
    es = create_energysystem_stub(num_in, num_out)
278
279
    nominal_capacity = 10
280
    minimal_value = 3
281
282
    eta_at_nom = {es.groups["bus input 0"]: 0.5}
283
    eta_at_min = {es.groups["bus input 0"]: 0.7}
284
285
    add_OffsetConverter(
286
        es,
287
        es.groups["bus output 0"],
288
        nominal_capacity,
289
        minimal_value,
290
        eta_at_nom,
291
        eta_at_min,
292
    )
293
294
    results = solve_and_extract_results(es)
295
296
    check_results(
297
        results,
298
        es.groups["bus output 0"],
299
        nominal_capacity,
300
        minimal_value,
301
        eta_at_nom,
302
        eta_at_min,
303
    )
304
305
306 View Code Duplication
def test_OffsetConverter_single_input_output_ref_input():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
307
    num_in = 1
308
    num_out = 1
309
    es = create_energysystem_stub(num_in, num_out)
310
311
    nominal_capacity = 10
312
    minimal_value = 3
313
314
    eta_at_nom = {es.groups["bus output 0"]: 0.7}
315
    eta_at_min = {es.groups["bus output 0"]: 0.5}
316
317
    add_OffsetConverter(
318
        es,
319
        es.groups["bus input 0"],
320
        nominal_capacity,
321
        minimal_value,
322
        eta_at_nom,
323
        eta_at_min,
324
    )
325
326
    results = solve_and_extract_results(es)
327
328
    check_results(
329
        results,
330
        es.groups["bus input 0"],
331
        nominal_capacity,
332
        minimal_value,
333
        eta_at_nom,
334
        eta_at_min,
335
    )
336
337
338 View Code Duplication
def test_OffsetConverter_single_input_output_ref_input_eta_decreasing():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
339
    num_in = 1
340
    num_out = 1
341
    es = create_energysystem_stub(num_in, num_out)
342
343
    nominal_capacity = 10
344
    minimal_value = 3
345
346
    eta_at_nom = {es.groups["bus output 0"]: 0.5}
347
    eta_at_min = {es.groups["bus output 0"]: 0.7}
348
349
    add_OffsetConverter(
350
        es,
351
        es.groups["bus input 0"],
352
        nominal_capacity,
353
        minimal_value,
354
        eta_at_nom,
355
        eta_at_min,
356
    )
357
358
    results = solve_and_extract_results(es)
359
360
    check_results(
361
        results,
362
        es.groups["bus input 0"],
363
        nominal_capacity,
364
        minimal_value,
365
        eta_at_nom,
366
        eta_at_min,
367
    )
368
369
370 View Code Duplication
def test_OffsetConverter_double_input_output_ref_input():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
371
    num_in = 2
372
    num_out = 2
373
    es = create_energysystem_stub(num_in, num_out)
374
375
    nominal_capacity = 10
376
    minimal_value = 3
377
378
    eta_at_nom = {
379
        es.groups["bus output 0"]: 0.7,
380
        es.groups["bus output 1"]: 0.2,
381
        es.groups["bus input 1"]: 0.2,
382
    }
383
    eta_at_min = {
384
        es.groups["bus output 0"]: 0.5,
385
        es.groups["bus output 1"]: 0.3,
386
        es.groups["bus input 1"]: 0.2,
387
    }
388
389
    add_OffsetConverter(
390
        es,
391
        es.groups["bus input 0"],
392
        nominal_capacity,
393
        minimal_value,
394
        eta_at_nom,
395
        eta_at_min,
396
    )
397
398
    results = solve_and_extract_results(es)
399
400
    check_results(
401
        results,
402
        es.groups["bus input 0"],
403
        nominal_capacity,
404
        minimal_value,
405
        eta_at_nom,
406
        eta_at_min,
407
    )
408
409
410 View Code Duplication
def test_OffsetConverter_double_input_output_ref_output():
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
411
    num_in = 2
412
    num_out = 2
413
414
    es = create_energysystem_stub(num_in, num_out)
415
416
    nominal_capacity = 10
417
    minimal_value = 3
418
419
    eta_at_nom = {
420
        es.groups["bus input 0"]: 0.7,
421
        es.groups["bus output 1"]: 0.2,
422
        es.groups["bus input 1"]: 0.2,
423
    }
424
    eta_at_min = {
425
        es.groups["bus input 0"]: 0.5,
426
        es.groups["bus output 1"]: 0.3,
427
        es.groups["bus input 1"]: 0.2,
428
    }
429
430
    add_OffsetConverter(
431
        es,
432
        es.groups["bus output 0"],
433
        nominal_capacity,
434
        minimal_value,
435
        eta_at_nom,
436
        eta_at_min,
437
    )
438
439
    results = solve_and_extract_results(es)
440
441
    check_results(
442
        results,
443
        es.groups["bus output 0"],
444
        nominal_capacity,
445
        minimal_value,
446
        eta_at_nom,
447
        eta_at_min,
448
    )
449
450
451
def test_two_OffsetConverters_with_and_without_investment():
452
    num_in = 1
453
    num_out = 1
454
455
    es = create_energysystem_stub(num_in, num_out)
456
457
    nominal_capacity = 10
458
    minimal_value = 3
459
460
    eta_at_nom = {
461
        es.groups["bus input 0"]: 0.7,
462
    }
463
    eta_at_min = {
464
        es.groups["bus input 0"]: 0.5,
465
    }
466
467
    add_OffsetConverter(
468
        es,
469
        es.groups["bus output 0"],
470
        nominal_capacity,
471
        minimal_value,
472
        eta_at_nom,
473
        eta_at_min,
474
    )
475
476
    input_bus = es.groups["bus input 0"]
477
478
    oc = solph.components.OffsetConverter(
479
        label="investment offset converter",
480
        inputs={input_bus: solph.Flow()},
481
        outputs={
482
            es.groups["bus output 0"]: solph.Flow(
483
                nonconvex=solph.NonConvex(),
484
                nominal_capacity=solph.Investment(
485
                    maximum=nominal_capacity, ep_costs=10
486
                ),
487
            )
488
        },
489
        conversion_factors={input_bus: 1},
490
        normed_offsets={input_bus: 0},
491
    )
492
493
    es.add(oc)
494
495
    # Use of experimental API to access nodes by label.
496
    # Can be removed with future release of network.
497
    with warnings.catch_warnings():
498
        warnings.simplefilter("ignore", ExperimentalFeatureWarning)
499
        fix_flow = es.flows()[es.node["bus output 0"], es.node["sink 0"]]
500
        fix_flow.fix = [v * 2 for v in fix_flow.fix]
501
    # if the model solves it is feasible
502
    _ = solve_and_extract_results(es)
503
504
505
def test_OffsetConverter_05x_compatibility():
506
    num_in = 1
507
    num_out = 1
508
    es = create_energysystem_stub(num_in, num_out)
509
510
    nominal_capacity = 10
511
    minimal_value = 3
512
513
    # Use of experimental API to access nodes by label.
514
    # Can be removed with future release of network.
515
    with warnings.catch_warnings():
516
        warnings.simplefilter("ignore", category=ExperimentalFeatureWarning)
517
        fix = [0] + np.linspace(minimal_value, nominal_capacity, 9).tolist()
518
        fix_flow = es.flows()[es.node["bus output 0"], es.node["sink 0"]]
519
        fix_flow.fix = fix
520
        fix_flow.nominal_capacity = 1
521
522
    eta_at_nom = 0.7
523
    eta_at_min = 0.5
524
525
    slope = (nominal_capacity - minimal_value) / (
526
        nominal_capacity / eta_at_nom - minimal_value / eta_at_min
527
    )
528
    offset = minimal_value / nominal_capacity * (1 - slope / eta_at_min)
529
530
    warnings.filterwarnings("ignore", "", DeprecationWarning)
531
    oc = solph.components.OffsetConverter(
532
        label="offset converter",
533
        inputs={es.groups["bus input 0"]: solph.Flow()},
534
        outputs={
535
            es.groups["bus output 0"]: solph.Flow(
536
                nonconvex=solph.NonConvex(),
537
                nominal_capacity=nominal_capacity,
538
                min=minimal_value / nominal_capacity,
539
            )
540
        },
541
        coefficients=(offset, slope),
542
    )
543
544
    es.add(oc)
545
    warnings.filterwarnings("always", "", DeprecationWarning)
546
547
    results = solve_and_extract_results(es)
548
549
    slope, offset = slope_offset_from_nonconvex_output(
550
        1, minimal_value / nominal_capacity, 0.7, 0.5
551
    )
552
    output_flow = results["offset converter", "bus output 0"]["sequences"][
553
        "flow"
554
    ]
555
    output_flow_status = results["offset converter", "bus output 0"][
556
        "sequences"
557
    ]["status"]
558
559
    input_flow_expected = (
560
        offset * nominal_capacity * output_flow_status + slope * output_flow
561
    )
562
    input_flow_actual = results["bus input 0", "offset converter"][
563
        "sequences"
564
    ]["flow"]
565
566
    np.testing.assert_array_almost_equal(
567
        input_flow_actual, input_flow_expected
568
    )
569
570
571
def test_error_handling():
572
    input_bus = solph.Bus("bus1")
573
    output_bus = solph.Bus("bus2")
574
575
    with pytest.raises(TypeError, match="cannot be used in combination"):
576
        warnings.filterwarnings("ignore", "", DeprecationWarning)
577
        _ = solph.components.OffsetConverter(
578
            label="offset converter",
579
            inputs={input_bus: solph.Flow()},
580
            outputs={
581
                output_bus: solph.Flow(
582
                    nonconvex=solph.NonConvex(),
583
                    nominal_capacity=10,
584
                    min=0.3,
585
                )
586
            },
587
            # values are arbitarty just to test the error
588
            coefficients=(-1, 0.4),
589
            conversion_factors={input_bus: 1},
590
            normed_offsets={input_bus: 0},
591
        )
592
        warnings.filterwarnings("always", "", DeprecationWarning)
593
594
        with pytest.raises(
595
            ValueError, match="Conversion factors cannot be specified for"
596
        ):
597
            _ = solph.components.OffsetConverter(
598
                label="offset converter",
599
                inputs={input_bus: solph.Flow()},
600
                outputs={
601
                    output_bus: solph.Flow(
602
                        nonconvex=solph.NonConvex(),
603
                        nominal_capacity=10,
604
                        min=0.3,
605
                    )
606
                },
607
                conversion_factors={input_bus: 1, output_bus: 1},
608
                normed_offsets={input_bus: 0},
609
            )
610
611
        with pytest.raises(
612
            ValueError, match="Normed offsets cannot be specified for"
613
        ):
614
            _ = solph.components.OffsetConverter(
615
                label="offset converter",
616
                inputs={input_bus: solph.Flow()},
617
                outputs={
618
                    output_bus: solph.Flow(
619
                        nonconvex=solph.NonConvex(),
620
                        nominal_capacity=10,
621
                        min=0.3,
622
                    )
623
                },
624
                conversion_factors={input_bus: 1},
625
                normed_offsets={input_bus: 0, output_bus: 0},
626
            )
627