Skip to content

Commit

Permalink
Merge pull request #9 from bThink-BGU/0.0.5
Browse files Browse the repository at this point in the history
priorities,manual bprogram running bug fix,docs and tests
tomyaacov authored May 31, 2023
2 parents a65bbba + 7bcc86f commit b02b1e9
Showing 19 changed files with 651 additions and 113 deletions.
110 changes: 5 additions & 105 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,118 +1,18 @@
# BPpy: Behavioral Programming In Python
A Python implementation for the Behavioral Programming paradigm

## Install
## Installation
You can install ``bppy`` with pip:

```shell
pip install bppy
```

## Running the Hot-Cold Example
python bppy/examples/hot_cold_all.py


## Writing a BPpy program
[bppy/examples/hot_cold_all.py](bppy/examples/hot_cold_all.py):
```python
from bppy import *

@b_thread
def add_hot():
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}

@b_thread
def add_cold():
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}

@b_thread
def control_temp():
e = BEvent("Dummy")
while True:
e = yield {waitFor: All(), block: e}

if __name__ == "__main__":
b_program = BProgram(bthreads=[add_hot(), add_cold(), control_temp()],
event_selection_strategy=SimpleEventSelectionStrategy(),
listener=PrintBProgramRunnerListener())
b_program.run()
```

## Using Z3-Solver SMT
[bppy/examples/hot_cold_smt.py](bppy/examples/hot_cold_smt.py):
```python
from bppy import *

hot = Bool('hot')
cold = Bool('cold')

@b_thread
def three_hot():
for i in range(3):
while (yield {request: hot})[hot] == false:
pass

@b_thread
def three_cold():
for j in range(3):
m = yield {request: cold}
while m[cold] == false:
m = yield {request: cold}

@b_thread
def exclusion():
while True:
yield {block: And(hot, cold)}

@b_thread
def schedule():
yield {block: cold}

if __name__ == "__main__":
b_program = BProgram(bthreads=[three_cold(), three_hot(), exclusion(), schedule()],
event_selection_strategy=SMTEventSelectionStrategy(),
listener=PrintBProgramRunnerListener())
b_program.run()
```

## Dynamically adding b-threads

[bppy/examples/hot_cold_dynamic.py](bppy/examples/hot_cold_dynamic.py):
```python
from bppy import *


@b_thread
def add_hot():
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}


@b_thread
def add_cold():
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}


@b_thread
def control_temp(block_event):
block_event = yield {waitFor: All(), block: block_event}
b_program.add_bthread(control_temp(block_event))


if __name__ == "__main__":
b_program = BProgram(bthreads=[add_hot(), add_cold(), control_temp(BEvent("HOT"))],
event_selection_strategy=SimpleEventSelectionStrategy(),
listener=PrintBProgramRunnerListener())
b_program.run()
```
## Documentation
* [BPpy documentation](https://bppy.readthedocs.io/en/latest/)
* General introduction to Behavioral Programming can be found in [this](https://bpjs.readthedocs.io/en/develop/) BPjs tutorial and others

## Citing BPpy
To cite this repository in publications:
```
@misc{bppy,
1 change: 1 addition & 0 deletions bppy/__init__.py
Original file line number Diff line number Diff line change
@@ -9,6 +9,7 @@
from bppy.model.event_selection.solver_based_event_selection_strategy import *
from bppy.model.event_selection.smt_event_selection_strategy import *
from bppy.model.event_selection.experimental_smt_event_selection_strategy import *
from bppy.model.event_selection.priority_based_event_selection_strategy import *
from bppy.model.b_event import *
from bppy.model.bprogram import *
from bppy.model.event_set import *
140 changes: 140 additions & 0 deletions bppy/examples/tic_tac_toe_priorities.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
from bppy import *

x = lambda row, col: BEvent('X' + str(row) + str(col))
o = lambda row, col: BEvent('O' + str(row) + str(col))

LINES = [[(i, j) for j in range(3)] for i in range(3)] + [[(i, j) for i in range(3)] for j in range(3)] + [
[(i, i) for i in range(3)]] + [[(i, 3 - i - 1) for i in range(3)]]
x_lines = [[x(i, j) for (i, j) in line] for line in LINES]
o_lines = [[o(i, j) for (i, j) in line] for line in LINES]

any_x = [x(i, j) for i in range(3) for j in range(3)]
any_o = [o(i, j) for i in range(3) for j in range(3)]
move_events = EventSet(lambda e: e.name.startswith('X') or e.name.startswith('O'))

static_event = {
'OWin': BEvent('OWin'),
'XWin': BEvent('XWin'),
'draw': BEvent('Draw')
}


@b_thread
def square_taken(row, col):
yield {waitFor: [x(row, col), o(row, col)]}
yield {block: [x(row, col), o(row, col)]}


@b_thread
def enforce_turns():
while True:
yield {waitFor: any_x, block: any_o}
yield {waitFor: any_o, block: any_x}


@b_thread
def end_of_game():
yield {waitFor: list(static_event.values())}
yield {block: All()}


@b_thread
def detect_draw():
for r in range(3):
for c in range(3):
yield {waitFor: move_events}
yield {request: static_event['draw'], priority: 90}


@b_thread
def detect_x_win(line):
for i in range(3):
yield {waitFor: line}
yield {request: static_event['XWin'], priority: 100}


@b_thread
def detect_o_win(line):
for i in range(3):
yield {waitFor: line}
yield {request: static_event['OWin'], priority: 100}


# Preference to put O on the center
@b_thread
def center_preference():
while True:
yield {request: o(1, 1), priority: 35}


# Preference to put O on the corners
@b_thread
def corner_preference():
while True:
yield {request: [o(0, 0), o(0, 2), o(2, 0), o(2, 2)], priority: 20}


# Preference to put O on the sides
@b_thread
def side_preference():
while True:
yield {request: [o(0, 1), o(1, 0), o(1, 2), o(2, 1)], priority: 10}


# player O strategy to add a third O to win
@b_thread
def add_third_o(line):
for i in range(2):
yield {waitFor: line}
yield {request: line, priority: 50}


# player O strategy to prevent a third X
@b_thread
def prevent_third_x(xline, oline):
for i in range(2):
yield {waitFor: xline}
yield {request: oline, priority: 40}


@b_thread
def block_fork(xfork, ofork):
for i in range(2):
yield {waitFor: xfork}
yield {request: ofork, priority: 30}


forks22 = [[x(1, 2), x(2, 0)], [x(2, 1), x(0, 2)], [x(1, 2), x(2, 1)]], [o(2, 2), o(0, 2), o(2, 0)]
forks02 = [[x(1, 2), x(0, 0)], [x(0, 1), x(2, 2)], [x(1, 2), x(0, 1)]], [o(0, 2), o(0, 0), o(2, 2)]
forks20 = [[x(1, 0), x(2, 2)], [x(2, 1), x(0, 0)], [x(2, 1), x(1, 0)]], [o(2, 0), o(0, 0), o(2, 2)]
forks00 = [[x(0, 1), x(2, 0)], [x(1, 0), x(0, 2)], [x(0, 1), x(1, 0)]], [o(0, 0), o(0, 2), o(2, 0)]
forks_diag = [[x(0, 2), x(2, 0)], [x(0, 0), x(2, 2)]], [o(0, 1), o(1, 0), o(1, 2), o(2, 1)]



# simulate player X
@b_thread
def player_x():
while True:
yield {request: any_x}


if __name__ == "__main__":
bprog = BProgram(
bthreads=[square_taken(i, j) for i in range(3) for j in range(3)] +
[enforce_turns(), end_of_game(), detect_draw()] +
[detect_x_win(line) for line in x_lines] +
[detect_o_win(line) for line in o_lines] +
[center_preference(), corner_preference(), side_preference()] +
[add_third_o(line) for line in o_lines] +
[prevent_third_x(xline, oline) for (xline, oline) in zip(x_lines, o_lines)] +
[block_fork(xfork, forks22[1]) for xfork in forks22[0]] +
[block_fork(xfork, forks02[1]) for xfork in forks02[0]] +
[block_fork(xfork, forks20[1]) for xfork in forks20[0]] +
[block_fork(xfork, forks00[1]) for xfork in forks00[0]] +
[block_fork(xfork, forks_diag[1]) for xfork in forks_diag[0]] +
[player_x()],
event_selection_strategy=PriorityBasedEventSelectionStrategy(default_priority=0),
listener=PrintBProgramRunnerListener()
)
bprog.run()
25 changes: 18 additions & 7 deletions bppy/model/bprogram.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from importlib import import_module
from inspect import getmembers, isfunction
from itertools import tee
import warnings

from z3 import *

@@ -26,9 +27,12 @@ def setup(self):
isinstance(o[1], ExprRef) or isinstance(o[1], list)])

self.new_bt = self.bthreads
self.load_new_bthreads()


def advance_bthreads(self,tickets, m):
if len(self.new_bt) > 0:
warnings.warn("Some new bthreads are not loaded and are not affecting the bprogram. Use the load_new_bthreads method to load them.", RuntimeWarning)
for l in tickets:
if m is None or self.event_selection_strategy.is_satisfied(m, l):
try:
@@ -42,10 +46,21 @@ def advance_bthreads(self,tickets, m):
except (KeyError, StopIteration):
pass

def add_bthread(self,bt):
def add_bthread(self, bt):
self.new_bt.append(bt)

def load_new_bthreads(self):
while len(self.new_bt) > 0:
new_tickets = [{'bt': bt} for bt in self.new_bt]
self.new_bt.clear()
self.advance_bthreads(new_tickets, None)
self.tickets.extend(new_tickets)

def next_event(self):
if len(self.new_bt) > 0:
warnings.warn(
"Some new bthreads are not loaded and are not affecting the bprogram. Use the load_new_bthreads method to load them.",
RuntimeWarning)
return self.event_selection_strategy.select(self.tickets, self.external_events_queue)

def run(self):
@@ -56,12 +71,8 @@ def run(self):
# Main loop
interrupted = False
while not interrupted:
#for dynamic adding new bthreads
while len(self.new_bt)>0:
new_tickets = [{'bt': bt} for bt in self.new_bt]
self.new_bt.clear()
self.advance_bthreads(new_tickets,None)
self.tickets.extend(new_tickets)
# for dynamically added bthreads
self.load_new_bthreads()

event = self.next_event()
# Finish the program if no event is selected
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from bppy.model.event_selection.simple_event_selection_strategy import SimpleEventSelectionStrategy
from collections.abc import Iterable
from bppy.model.b_event import BEvent


class PriorityBasedEventSelectionStrategy(SimpleEventSelectionStrategy):

def __init__(self, default_priority=0):
self.default_priority = default_priority

def selectable_events(self, statements):
possible_events = set()
for statement in statements:
if 'request' in statement: # should be eligible for sets
p = statement.get('priority', self.default_priority)
if isinstance(statement['request'], Iterable):
possible_events.update([(x, p) for x in statement['request']])
elif isinstance(statement['request'], BEvent):
possible_events.add((statement['request'], p))
else:
raise TypeError("request parameter should be BEvent or iterable")
for statement in statements:
if 'block' in statement:
if isinstance(statement.get('block'), BEvent):
possible_events = {(x, p) for x, p in possible_events if x != statement.get('block')}
else:
possible_events = {(x, p) for x, p in possible_events if x not in statement.get('block')}
if len(possible_events) == 0:
return possible_events
max_priority = max([p for _, p in possible_events])
return {x for x, p in possible_events if p == max_priority}
3 changes: 3 additions & 0 deletions bppy/model/sync_statement.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
request = "request"
waitFor = "waitFor"
block = "block"
mustFinish = "mustFinish"
priority = "priority"

4 changes: 4 additions & 0 deletions docs/Examples/external_events.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Using External Events
+++++++++++++++++++++

.. literalinclude :: ../../bppy/examples/external_events.py
4 changes: 4 additions & 0 deletions docs/Examples/hello_world.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Hello World
+++++++++++

.. literalinclude :: ../../bppy/examples/hello_world.py
4 changes: 4 additions & 0 deletions docs/Examples/hot_cold.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Hot Cold
+++++++++++

.. literalinclude :: ../../bppy/examples/hot_cold_all.py
4 changes: 4 additions & 0 deletions docs/Examples/hot_cold_dynamic.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Dynamically Adding b-threads
++++++++++++++++++++++++++++

.. literalinclude :: ../../bppy/examples/hot_cold_dynamic.py
4 changes: 4 additions & 0 deletions docs/Examples/hot_cold_smt.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Using Z3-Solver SMT based BP
++++++++++++++++++++++++++++

.. literalinclude :: ../../bppy/examples/hot_cold_smt.py
4 changes: 4 additions & 0 deletions docs/Examples/tic_tac_toe_priorities.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Using Priority Based Event Selection Strategy (Tic Tac Toe)
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

.. literalinclude :: ../../bppy/examples/tic_tac_toe_priorities.py
20 changes: 20 additions & 0 deletions docs/Makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
# Minimal makefile for Sphinx documentation
#

# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= sphinx-build
SOURCEDIR = .
BUILDDIR = _build

# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)

.PHONY: help Makefile

# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
27 changes: 27 additions & 0 deletions docs/conf.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html

# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information

project = 'BPpy'
copyright = '2023, Tom Yaacov'
author = 'Tom Yaacov'

# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

extensions = []

templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']



# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output

html_theme = 'sphinx_rtd_theme'
html_static_path = ['_static']
55 changes: 55 additions & 0 deletions docs/index.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
.. BPpy documentation master file, created by
sphinx-quickstart on Wed May 31 15:43:12 2023.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
Welcome to BPpy's documentation!
================================

BPpy is a Python implementation for the Behavioral Programming paradigm.

+++++++++++++++
Installation
+++++++++++++++
You can install ``bppy`` with pip:

.. code-block:: shell
pip install bppy
.. toctree::
:maxdepth: 2
:caption: Examples

Examples/hello_world
Examples/hot_cold
Examples/hot_cold_smt
Examples/hot_cold_dynamic
Examples/external_events
Examples/tic_tac_toe_priorities



+++++++++++++++
Citing BPpy
+++++++++++++++
To cite this repository in publications:

.. code-block:: none
@misc{bppy,
author = {Tom Yaacov},
title = {BPpy: Behavioral Programming In Python},
year = {2020},
publisher = {GitHub},
journal = {GitHub repository},
howpublished = {\url{https://github.com/bThink-BGU/BPpy}},
}
Indices and tables
==================

* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`
35 changes: 35 additions & 0 deletions docs/make.bat
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
@ECHO OFF

pushd %~dp0

REM Command file for Sphinx documentation

if "%SPHINXBUILD%" == "" (
set SPHINXBUILD=sphinx-build
)
set SOURCEDIR=.
set BUILDDIR=_build

%SPHINXBUILD% >NUL 2>NUL
if errorlevel 9009 (
echo.
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
echo.installed, then set the SPHINXBUILD environment variable to point
echo.to the full path of the 'sphinx-build' executable. Alternatively you
echo.may add the Sphinx directory to PATH.
echo.
echo.If you don't have Sphinx installed, grab it from
echo.https://www.sphinx-doc.org/
exit /b 1
)

if "%1" == "" goto help

%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
goto end

:help
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%

:end
popd
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
@@ -5,7 +5,7 @@

setuptools.setup(
name="bppy",
version="0.0.4",
version="0.0.5",
author="Tom Yaacov",
author_email="tomyaacov1210@gmail.com",
description="BPpy: Behavioral Programming In Python",
22 changes: 22 additions & 0 deletions tests/test_bprogram.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
import unittest
from bppy import *


class TestBProgram(unittest.TestCase):

def test_advance_bthreads(self):
@b_thread
def hello():
yield {request: BEvent("Hello")}

@b_thread
def world():
yield {request: BEvent("World")}

b_program = BProgram(bthreads=[hello(), world()],
event_selection_strategy=SimpleEventSelectionStrategy())

b_program.setup()
b_program.advance_bthreads(b_program.tickets, BEvent("Hello"))
b_program.advance_bthreads(b_program.tickets, BEvent("World"))
assert all([len(x) == 0 for x in b_program.tickets]) and len(b_program.tickets) == 2
269 changes: 269 additions & 0 deletions tests/test_examples.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,269 @@
import unittest
from bppy import *


class TestExamples(unittest.TestCase):

def test_hello_world(self):
@b_thread
def hello():
yield {request: BEvent("Hello")}

@b_thread
def world():
yield {request: BEvent("World")}
b_program = BProgram(bthreads=[hello(), world()],
event_selection_strategy=SimpleEventSelectionStrategy())
b_program.run()
assert True

def test_hot_cold_all(self):
@b_thread
def add_hot():
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}

@b_thread
def add_cold():
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}

@b_thread
def control_temp():
e = BEvent("Dummy")
while True:
e = yield {waitFor: All(), block: e}

b_program = BProgram(bthreads=[add_hot(), add_cold(), control_temp()],
event_selection_strategy=SimpleEventSelectionStrategy())
b_program.run()
assert True

def test_hot_cold_bath(self):
@b_thread
def add_hot():
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}

@b_thread
def add_cold():
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}

@b_thread
def control_temp():
while True:
yield {waitFor: BEvent("COLD"), block: BEvent("HOT")}
yield {waitFor: BEvent("HOT"), block: BEvent("COLD")}

b_program = BProgram(bthreads=[add_hot(), add_cold(), control_temp()],
event_selection_strategy=SimpleEventSelectionStrategy())
b_program.run()
assert True

def test_hot_cold_dynamic(self):
@b_thread
def add_hot():
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}
yield {request: BEvent("HOT")}

@b_thread
def add_cold():
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}
yield {request: BEvent("COLD")}

@b_thread
def control_temp(block_event):
block_event = yield {waitFor: All(), block: block_event}
b_program.add_bthread(control_temp(block_event))

b_program = BProgram(bthreads=[add_hot(), add_cold(), control_temp(BEvent("HOT"))],
event_selection_strategy=SimpleEventSelectionStrategy())
b_program.run()
assert True

def test_hot_cold_smt(self):
hot = Bool('hot')
cold = Bool('cold')

@b_thread
def three_hot():
for i in range(3):
yield {request: hot, waitFor: hot}

@b_thread
def three_cold():
for j in range(3):
yield {request: cold, waitFor: cold}

@b_thread
def control_temp():
m = yield {}
while True:
if is_true(m[cold]):
m = yield {block: cold}
if is_true(m[hot]):
m = yield {block: hot}

@b_thread
def mutual_exclusion():
yield {block: And(cold, hot), waitFor: false}

b_program = BProgram(bthreads=[three_cold(), three_hot(), mutual_exclusion(), control_temp()],
event_selection_strategy=SMTEventSelectionStrategy())
b_program.run()
assert True

def test_external_events(self):
class External(BEvent):
pass

any_external = EventSet(lambda event: isinstance(event, External))

@b_thread
def add_external():
b_program.enqueue_external_event(External("A"))
b_program.enqueue_external_event(External("B"))
b_program.enqueue_external_event(External("C"))
while True:
yield {waitFor: All()}

@b_thread
def act_on_external():
while True:
# triggers external events if exists, else terminates the bprogram
event = yield {block: All(), waitFor: any_external}
yield {request: BEvent(event.name)}

b_program = BProgram(bthreads=[add_external(), act_on_external()],
event_selection_strategy=SimpleEventSelectionStrategy())
b_program.run()
assert True

def test_tic_tac_toe_priorities(self):
x = lambda row, col: BEvent('X' + str(row) + str(col))
o = lambda row, col: BEvent('O' + str(row) + str(col))

LINES = [[(i, j) for j in range(3)] for i in range(3)] + [[(i, j) for i in range(3)] for j in range(3)] + [
[(i, i) for i in range(3)]] + [[(i, 3 - i - 1) for i in range(3)]]
x_lines = [[x(i, j) for (i, j) in line] for line in LINES]
o_lines = [[o(i, j) for (i, j) in line] for line in LINES]

any_x = [x(i, j) for i in range(3) for j in range(3)]
any_o = [o(i, j) for i in range(3) for j in range(3)]
move_events = EventSet(lambda e: e.name.startswith('X') or e.name.startswith('O'))

static_event = {
'OWin': BEvent('OWin'),
'XWin': BEvent('XWin'),
'draw': BEvent('Draw')
}

@b_thread
def square_taken(row, col):
yield {waitFor: [x(row, col), o(row, col)]}
yield {block: [x(row, col), o(row, col)]}

@b_thread
def enforce_turns():
while True:
yield {waitFor: any_x, block: any_o}
yield {waitFor: any_o, block: any_x}

@b_thread
def end_of_game():
yield {waitFor: list(static_event.values())}
yield {block: All()}

@b_thread
def detect_draw():
for r in range(3):
for c in range(3):
yield {waitFor: move_events}
yield {request: static_event['draw'], priority: 90}

@b_thread
def detect_x_win(line):
for i in range(3):
yield {waitFor: line}
yield {request: static_event['XWin'], priority: 100}

@b_thread
def detect_o_win(line):
for i in range(3):
yield {waitFor: line}
yield {request: static_event['OWin'], priority: 100}

# Preference to put O on the center
@b_thread
def center_preference():
while True:
yield {request: o(1, 1), priority: 35}

# Preference to put O on the corners
@b_thread
def corner_preference():
while True:
yield {request: [o(0, 0), o(0, 2), o(2, 0), o(2, 2)], priority: 20}

# Preference to put O on the sides
@b_thread
def side_preference():
while True:
yield {request: [o(0, 1), o(1, 0), o(1, 2), o(2, 1)], priority: 10}

# player O strategy to add a third O to win
@b_thread
def add_third_o(line):
for i in range(2):
yield {waitFor: line}
yield {request: line, priority: 50}

# player O strategy to prevent a third X
@b_thread
def prevent_third_x(xline, oline):
for i in range(2):
yield {waitFor: xline}
yield {request: oline, priority: 40}

@b_thread
def block_fork(xfork, ofork):
for i in range(2):
yield {waitFor: xfork}
yield {request: ofork, priority: 30}

forks22 = [[x(1, 2), x(2, 0)], [x(2, 1), x(0, 2)], [x(1, 2), x(2, 1)]], [o(2, 2), o(0, 2), o(2, 0)]
forks02 = [[x(1, 2), x(0, 0)], [x(0, 1), x(2, 2)], [x(1, 2), x(0, 1)]], [o(0, 2), o(0, 0), o(2, 2)]
forks20 = [[x(1, 0), x(2, 2)], [x(2, 1), x(0, 0)], [x(2, 1), x(1, 0)]], [o(2, 0), o(0, 0), o(2, 2)]
forks00 = [[x(0, 1), x(2, 0)], [x(1, 0), x(0, 2)], [x(0, 1), x(1, 0)]], [o(0, 0), o(0, 2), o(2, 0)]
forks_diag = [[x(0, 2), x(2, 0)], [x(0, 0), x(2, 2)]], [o(0, 1), o(1, 0), o(1, 2), o(2, 1)]

# simulate player X
@b_thread
def player_x():
while True:
yield {request: any_x}

bprog = BProgram(
bthreads=[square_taken(i, j) for i in range(3) for j in range(3)] +
[enforce_turns(), end_of_game(), detect_draw()] +
[detect_x_win(line) for line in x_lines] +
[detect_o_win(line) for line in o_lines] +
[center_preference(), corner_preference(), side_preference()] +
[add_third_o(line) for line in o_lines] +
[prevent_third_x(xline, oline) for (xline, oline) in zip(x_lines, o_lines)] +
[block_fork(xfork, forks22[1]) for xfork in forks22[0]] +
[block_fork(xfork, forks02[1]) for xfork in forks02[0]] +
[block_fork(xfork, forks20[1]) for xfork in forks20[0]] +
[block_fork(xfork, forks00[1]) for xfork in forks00[0]] +
[block_fork(xfork, forks_diag[1]) for xfork in forks_diag[0]] +
[player_x()],
event_selection_strategy=PriorityBasedEventSelectionStrategy(default_priority=0))
bprog.run()
assert True

0 comments on commit b02b1e9

Please sign in to comment.