OilfieldToolsNet/opm-common
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test well.py all pass with new Well2.

Browse Source
This commit is contained in:
Steinar Foss 2019-06-19 13:51:43 +02:00
parent a1d4fe13af
commit 4161a1339a
3 changed files with 60 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
python/python/sunbeam/schedule.py
View File

@ -11,9 +11,8 @@ class Schedule(object):
lw = len(self.wells)
return 'Schedule(timesteps: %d, wells: %d)' % (lt, lw)
@property
def wells(self):
return list(map(Well, self._wells))
def get_wells(self, timestep = 0):
return list(map(Well, self._get_wells(timestep)))
def group(self, timestep=0):
return {grp.name: grp for grp in self.groups(timestep)}
@ -28,16 +27,14 @@ class Schedule(object):
@delegate(lib.Well)
class Well(object):
def pos(self, timestep = None):
if timestep is None:
return self.I(), self.J(), self.ref()
return self.I(timestep), self.J(timestep), self.ref(timestep)
def pos(self):
return self.I(), self.J(), self.ref()
def __repr__(self):
return 'Well(name = "%s")' % self.name
def connections(self, timestep):
return list(map(Connection, self._connections(timestep)))
def connections(self):
return list(map(Connection, self._connections()))
def __eq__(self,other):
return self._sun.__equal__(other._sun)
@ -48,13 +45,13 @@ class Well(object):
return fn
@staticmethod
def injector(timestep):
def fn(well): return well.isinjector(timestep)
def injector():
def fn(well): return well.isinjector()
return fn
@staticmethod
def producer(timestep):
def fn(well): return well.isproducer(timestep)
def producer():
def fn(well): return well.isproducer()
return fn
# using the names flowing and closed for functions that test if a well is
@ -62,13 +59,13 @@ class Well(object):
# imperative words 'open' and 'close' (or 'shut') for *changing* the status
# later
@staticmethod
def flowing(timestep):
def fn(well): return well.status(timestep) == 'OPEN'
def flowing():
def fn(well): return well.status() == 'OPEN'
return fn
@staticmethod
def closed(timestep):
def fn(well): return well.status(timestep) == 'SHUT'
def closed():
def fn(well): return well.status() == 'SHUT'
return fn
@staticmethod

2
python/sunbeam/schedule.cpp
View File

@ -91,7 +91,7 @@ void sunbeam::export_Schedule(py::module& module) {
.def_property_readonly( "start", &get_start_time )
.def_property_readonly( "end", &get_end_time )
.def_property_readonly( "timesteps", &get_timesteps )
.def( "get_wells", &Schedule::getWells2)
.def( "_get_wells", &Schedule::getWells2)
.def("_getwell", &get_well)
.def( "__contains__", &has_well )
.def( "_group", &Schedule::getGroup, ref_internal)

85
python/tests/wells.py
View File

@ -7,24 +7,17 @@ class TestWells(unittest.TestCase):
def setUpClass(cls):
cls.sch = sunbeam.parse('spe3/SPE3CASE1.DATA').schedule
cls.timesteps = cls.sch.timesteps
cls.wells = cls.sch.wells
# cls.wells = cls.sch.wells
def inje(self):
return next(iter(filter(sunbeam.Well.injector(0), self.wells)))
def inje(self, wells):
return next(iter(filter(sunbeam.Well.injector(), wells)))
def prod(self):
return next(iter(filter(sunbeam.Well.producer(0), self.wells)))
def prod(self, wells):
return next(iter(filter(sunbeam.Well.producer(), wells)))
def testWellPos0(self):
well = self.wells[0]
i, j, refdepth = well.pos(0)
self.assertEqual(6, i)
self.assertEqual(6, j)
self.assertEqual(2247.9, refdepth)
def testWellProperty(self):
well = self.wells[0]
wells = self.sch.get_wells(0)
well = wells[0]
i, j, refdepth = well.pos()
self.assertEqual(6, i)
@ -32,43 +25,52 @@ class TestWells(unittest.TestCase):
self.assertEqual(2247.9, refdepth)
def testWellStatus(self):
for well in self.wells:
self.assertEqual("OPEN", well.status(0))
wells = self.sch.get_wells(0)
for well in wells:
self.assertEqual("OPEN", well.status())
def testGroupName(self):
for well in self.wells:
self.assertEqual("G1", well.group(0))
wells = self.sch.get_wells(0)
for well in wells:
self.assertEqual("G1", well.group())
def testPreferredPhase(self):
inje, prod = self.inje(), self.prod()
wells = self.sch.get_wells(0)
inje, prod = self.inje(wells), self.prod(wells)
self.assertEqual("GAS", inje.preferred_phase)
self.assertEqual("GAS", prod.preferred_phase)
def testGuideRate(self):
inje, prod = self.inje(), self.prod()
self.assertEqual(-1.0, inje.guide_rate(1))
self.assertEqual(-1.0, prod.guide_rate(1))
wells = self.sch.get_wells(1)
inje, prod = self.inje(wells), self.prod(wells)
self.assertEqual(-1.0, inje.guide_rate())
self.assertEqual(-1.0, prod.guide_rate())
self.assertEqual(-1.0, inje.guide_rate(14))
self.assertEqual(-1.0, prod.guide_rate(14))
wells = self.sch.get_wells(14)
inje, prod = self.inje(wells), self.prod(wells)
self.assertEqual(-1.0, inje.guide_rate())
self.assertEqual(-1.0, prod.guide_rate())
def testGroupControl(self):
inje, prod = self.inje(), self.prod()
self.assertTrue(inje.available_gctrl(1))
self.assertTrue(prod.available_gctrl(1))
wells = self.sch.get_wells(1)
inje, prod = self.inje(wells), self.prod(wells)
self.assertTrue(inje.available_gctrl())
self.assertTrue(prod.available_gctrl())
self.assertTrue(inje.available_gctrl(14))
self.assertTrue(prod.available_gctrl(14))
wells = self.sch.get_wells(14)
inje, prod = self.inje(wells), self.prod(wells)
self.assertTrue(inje.available_gctrl())
self.assertTrue(prod.available_gctrl())
def testWellDefinedFilter(self):
defined0 = filter(sunbeam.Well.defined(0), self.wells)
defined1 = filter(sunbeam.Well.defined(1), self.wells)
defined0 = list(filter(sunbeam.Well.defined(0), self.sch.get_wells(0) ))
defined1 = list(filter(sunbeam.Well.defined(1), self.sch.get_wells(1) ))
self.assertEqual(len(list(defined0)), 2)
self.assertEqual(len(list(defined1)), 2)
def testWellProdInjeFilter(self):
inje = list(filter(sunbeam.Well.injector(0), self.wells))
prod = list(filter(sunbeam.Well.producer(0), self.wells))
inje = list(filter(sunbeam.Well.injector(), self.sch.get_wells(0) ))
prod = list(filter(sunbeam.Well.producer(), self.sch.get_wells(0) ))
self.assertEqual(len(inje), 1)
self.assertEqual(len(prod), 1)
@ -77,20 +79,23 @@ class TestWells(unittest.TestCase):
self.assertEqual(prod[0].name, "PROD")
def testOpenFilter(self):
open_at_1 = sunbeam.Well.flowing(1)
flowing = list(filter(open_at_1, self.wells))
closed = list(filter(lambda well: not open_at_1(well), self.wells))
wells = self.sch.get_wells(1)
open_at_1 = sunbeam.Well.flowing()
flowing = list(filter(open_at_1, wells))
closed = list(filter(lambda well: not open_at_1(well), wells))
self.assertEqual(2, len(flowing))
self.assertEqual(0, len(closed))
flowing1 = filter(lambda x: not sunbeam.Well.closed(1)(x), self.wells)
closed1 = filter(sunbeam.Well.closed(1), self.wells)
flowing1 = filter(lambda x: not sunbeam.Well.closed(1)(x), wells)
closed1 = filter(sunbeam.Well.closed(), wells)
self.assertListEqual(list(closed), list(closed1))
def testCompletions(self):
w0 = self.wells[0]
c0,c1 = w0.connections(len(self.timesteps) - 1)
num_steps = len( self.sch.timesteps )
w0 = self.sch.get_wells(num_steps - 1)[0]
c0,c1 = w0.connections()
self.assertEqual((6,6,2), c0.pos)
self.assertEqual((6,6,3), c1.pos)