File: test_schedule.py

package info (click to toggle)
opm-common 2025.10%2Bds-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 96,920 kB
  • sloc: cpp: 291,772; python: 3,609; sh: 198; xml: 174; pascal: 136; makefile: 12
file content (122 lines) | stat: -rw-r--r-- 4,273 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import unittest
import datetime as dt

from opm.io.parser import Parser
from opm.io.ecl_state import EclipseState
from opm.io.schedule import Schedule

try:
    from tests.utils import test_path
except ImportError:
    from utils import test_path

class TestSchedule(unittest.TestCase):


    @classmethod
    def setUpClass(cls):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        cls.sch = Schedule( deck, state )

    def testWells(self):
        self.assertTrue( isinstance( self.sch.get_wells(0), list) )
        self.assertEqual(2, len(self.sch.get_wells(0)))

        with self.assertRaises(Exception):
            self.sch.get_well('foo', 0)

    def testContains(self):
        self.assertTrue('PROD' in self.sch)
        self.assertTrue('INJ'  in self.sch)
        self.assertTrue('NOT'  not in self.sch)
        self.assertFalse('NOT' in self.sch)

    def testStartEnd(self):
        self.assertEqual(dt.datetime(2015, 1, 1),   self.sch.start)
        self.assertEqual(dt.datetime(2029, 12, 28), self.sch.end)

    def testReportsteps(self):
        reportsteps = self.sch.reportsteps
        self.assertEqual(176, len(reportsteps))
        self.assertEqual(dt.datetime(2016, 1, 1), reportsteps[7])

    def testGroups(self):

        G1 = self.sch[0].group( 'G1')
        self.assertTrue(G1.name == 'G1')
        self.assertTrue(G1.num_wells == 2)

        names = G1.well_names

        self.assertEqual(2, len(names))

        self.assertTrue(self.sch.get_well('INJ', 0).isinjector())
        self.assertTrue(self.sch.get_well('PROD', 0).isproducer())

        with self.assertRaises(Exception):
            self.sch[0].group('foo')

    def test_injection_properties(self):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        sch = Schedule( deck, state )
        report_step = 4
        well_name = 'INJ'
        prop = sch.get_injection_properties(well_name, report_step)
        self.assertEqual(prop['surf_inj_rate'], 4700.0) # Mscf/day
        self.assertEqual(prop['resv_inj_rate'], 0.0) # rb/day
        self.assertEqual(prop['bhp_target'], 4000.0) # psi
        self.assertEqual(prop['thp_target'], 0.0)
        with self.assertRaises(IndexError):
            prop = sch.get_injection_properties("UNDEF", report_step)
        with self.assertRaises(KeyError):
            prop = sch.get_injection_properties("PROD", report_step)

    def test_production_properties(self):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        sch = Schedule( deck, state )
        report_step = 4
        well_name = 'PROD'
        prop = sch.get_production_properties(well_name, report_step)
        self.assertEqual(prop['alq_value'], 0.0)
        self.assertEqual(prop['bhp_target'], 500.0)
        self.assertEqual(prop['gas_rate'], 6200.0)
        self.assertEqual(prop['oil_rate'], 0.0)
        self.assertEqual(prop['water_rate'], 0.0)
        self.assertEqual(prop['liquid_rate'], 0.0)
        self.assertEqual(prop['resv_rate'], 0.0)
        self.assertEqual(prop['thp_target'], 0.0)
        with self.assertRaises(IndexError):
            prop = sch.get_production_properties("UNDEF", report_step)
        with self.assertRaises(KeyError):
            prop = sch.get_production_properties("INJ", report_step)

    def test_well_names(self):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        sch = Schedule( deck, state )
        wnames = sch.well_names("*")
        self.assertTrue("PROD" in wnames)
        self.assertTrue("INJ" in wnames)
        self.assertEqual(len(wnames), 2)


    def test_getitem(self):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        sch = Schedule( deck, state )
        self.assertEqual(len(sch), 176)
        with self.assertRaises(IndexError):
            a = sch[200]

        st100 = sch[100]
        nupcol = st100.nupcol

    def test_open_shut(self):
        deck  = Parser().parse(test_path('spe3/SPE3CASE1.DATA'))
        state = EclipseState(deck)
        sch = Schedule( deck, state )
if __name__ == "__main__":
    unittest.main()