Merge pull request #11 from SmokinCaterpillar/fix/resetting_of_steem_…

…on_error Fix/resetting of steem on error
SmokinCaterpillar · Mar 22, 2018 · f571704 · f571704
2 parents 3cfb28d + f199b9c
commit f571704
Show file tree

Hide file tree

Showing 26 changed files with 175 additions and 153 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+*0.6.1a* - 2018-03-22
+
+* New multiprocessing sage Steem object
+
 *0.6.0a* - 2018-03-12
 
 * Two new readability scores

diff --git a/integration_tests/bchain/checkops_test.py b/integration_tests/bchain/checkops_test.py
@@ -1,6 +1,6 @@
 import pandas as pd
 
-from trufflepig.testutils.pytest_fixtures import steem_kwargs, steem
+from trufflepig.testutils.pytest_fixtures import steem
 import trufflepig.bchain.checkops as tpcd
 
 
@@ -13,10 +13,10 @@ def test_check_all_ops_between(steem):
     assert comments
 
 
-def test_check_all_ops_between_parallel(steem_kwargs):
+def test_check_all_ops_between_parallel(steem):
     start = pd.to_datetime('2018-01-17-13:39:00')
     end = pd.to_datetime('2018-01-17-13:41:00')
-    comments = tpcd.check_all_ops_between_parallel(start, end, steem_kwargs,
+    comments = tpcd.check_all_ops_between_parallel(start, end, steem,
                                        account='originalworks',
                                        stop_after=25, ncores=4)
     assert comments

diff --git a/integration_tests/bchain/getaccountdata_test.py b/integration_tests/bchain/getaccountdata_test.py
@@ -1,6 +1,6 @@
 import pandas as pd
 
-from trufflepig.testutils.pytest_fixtures import steem_kwargs, steem
+from trufflepig.testutils.pytest_fixtures import steem
 import trufflepig.bchain.getaccountdata as tpac
 
 
@@ -45,12 +45,12 @@ def test_bidbot_test_max_time(steem):
     assert len(result) <= 1
 
 
-def test_get_upvote_payments_for_accounts(steem_kwargs):
+def test_get_upvote_payments_for_accounts(steem):
     min_datetime = pd.datetime.utcnow() - pd.Timedelta(days=14)
     max_datetime = min_datetime + pd.Timedelta(days=5)
     accounts = ['trufflepig', 'smcaterpillar', 'brittuf']
     result = tpac.get_upvote_payments_for_accounts(accounts,
-                                                   steem_kwargs,
+                                                   steem,
                                                    min_datetime=min_datetime,
                                                    max_datetime=max_datetime)
     assert result
diff --git a/integration_tests/bchain/getdata_test.py b/integration_tests/bchain/getdata_test.py
@@ -5,7 +5,7 @@
 from steem.post import Post
 
 import trufflepig.bchain.getdata as tpbg
-from trufflepig.testutils.pytest_fixtures import steem, steem_kwargs, \
+from trufflepig.testutils.pytest_fixtures import steem, \
     bchain, temp_dir
 
 
@@ -56,18 +56,18 @@ def test_scrape_date(steem, temp_dir):
     assert len(p1) > 0
 
 
-def test_scrape_or_load_data_parallel(temp_dir, steem_kwargs):
+def test_scrape_or_load_data_parallel(temp_dir, steem):
 
-    frame = tpbg.load_or_scrape_training_data(steem_kwargs,
+    frame = tpbg.load_or_scrape_training_data(steem,
                                               temp_dir,
                                               days=3,
                                               stop_after=10,
                                               ncores=5)
     assert len(frame) >= 30
 
 
-def test_scrape_recent_date(steem_kwargs):
-    frame = tpbg.scrape_hour_data(steem_kwargs,
+def test_scrape_recent_date(steem):
+    frame = tpbg.scrape_hour_data(steem,
                                   stop_after=50,
                                   ncores=1)
     assert len(frame)

diff --git a/integration_tests/bchain/paydelegates_test.py b/integration_tests/bchain/paydelegates_test.py
@@ -1,19 +1,19 @@
 import pytest
 import pandas as pd
 
-from trufflepig.testutils.pytest_fixtures import steem_kwargs
+from trufflepig.testutils.pytest_fixtures import steem
 import trufflepig.bchain.paydelegates as tppd
 import trufflepig.bchain.postdata as tpdd
 from trufflepig import config
 
 
 @pytest.mark.skipif(config.ACTIVE_KEY is None, reason="needs active key")
-def test_pay_delegates(steem_kwargs):
+def test_pay_delegates(steem):
 
-    tpdd.create_wallet(steem_kwargs, config.PASSWORD,
+    tpdd.create_wallet(steem, config.PASSWORD,
                        config.POSTING_KEY, config.ACTIVE_KEY)
 
     tppd.pay_delegates(account=config.ACCOUNT,
-                       steem_args=steem_kwargs,
+                       steem=steem,
                        current_datetime=pd.datetime.utcnow()#'2029-01-01'
                        )
diff --git a/integration_tests/bchain/postdata_test.py b/integration_tests/bchain/postdata_test.py
@@ -6,7 +6,7 @@
 import trufflepig.preprocessing as tppp
 from trufflepig import config
 from trufflepig.testutils import random_data
-from trufflepig.testutils.pytest_fixtures import steem_kwargs, steem
+from trufflepig.testutils.pytest_fixtures import steem
 
 
 @pytest.mark.skipif(config.PASSWORD is None, reason="needs posting key")
@@ -31,13 +31,11 @@ def test_test_top10post(steem):
 
 
 @pytest.mark.skipif(config.PASSWORD is None, reason="needs posting key")
-def test_test_all_top_with_real_data(steem_kwargs):
-
-    steem = tpbg.check_and_convert_steem(steem_kwargs)
+def test_test_all_top_with_real_data(steem):
 
     steem.wallet.unlock(config.PASSWORD)
 
-    df = tpbg.scrape_hour_data(steem_kwargs, stop_after=10)
+    df = tpbg.scrape_hour_data(steem, stop_after=10)
 
     df = tppp.preprocess(df)
 

diff --git a/integration_tests/bchain/postweeklyupdate_test.py b/integration_tests/bchain/postweeklyupdate_test.py
@@ -7,7 +7,7 @@
 import trufflepig.preprocessing as tppp
 from trufflepig import config
 from trufflepig.testutils.random_data import create_n_random_posts
-from trufflepig.testutils.pytest_fixtures import steem_kwargs
+from trufflepig.testutils.pytest_fixtures import steem
 
 
 def test_statistics():
@@ -43,15 +43,15 @@ def test_statistics():
     assert body
 
 
-def test_existence(steem_kwargs):
+def test_existence(steem):
     result = tppw.return_overview_permalink_if_exists(account=config.ACCOUNT,
-                                                      steem_args=steem_kwargs,
+                                                      steem=steem,
                                                       current_datetime=pd.datetime.utcnow())
     assert isinstance(result, str)
 
 
 @pytest.mark.skipif(config.PASSWORD is None, reason="needs posting key")
-def test_weekly_post(steem_kwargs):
+def test_weekly_post(steem):
     posts = create_n_random_posts(300)
 
     post_frame = pd.DataFrame(posts)
@@ -73,7 +73,7 @@ def test_weekly_post(steem_kwargs):
 
     permalink = tppw.post_weakly_update(pipeline, post_frame,
                                         account=config.ACCOUNT,
-                                        steem_args=steem_kwargs,
+                                        steem=steem,
                                         current_datetime=current_date)
 
     assert permalink
diff --git a/integration_tests/model_test.py b/integration_tests/model_test.py
@@ -7,7 +7,7 @@
 import trufflepig.model as tpmo
 import trufflepig.preprocessing as tppp
 from trufflepig.testutils.random_data import create_n_random_posts
-from trufflepig.testutils.pytest_fixtures import steem_kwargs, temp_dir
+from trufflepig.testutils.pytest_fixtures import steem, temp_dir
 
 
 def test_pipeline_model():
@@ -146,8 +146,8 @@ def predict(self, df):
                                predicted_votes[:, np.newaxis]], axis=1)
 
 
-def test_find_truffles_with_real_data(steem_kwargs):
-    df = tpbg.scrape_hour_data(steem_kwargs, stop_after=20)
+def test_find_truffles_with_real_data(steem):
+    df = tpbg.scrape_hour_data(steem, stop_after=20)
 
     df = tppp.preprocess(df)
 

diff --git a/integration_tests/pigonduty_test.py b/integration_tests/pigonduty_test.py
@@ -1,18 +1,18 @@
 import pytest
 
-from trufflepig.testutils.pytest_fixtures import steem_kwargs, steem
+from trufflepig.testutils.pytest_fixtures import steem
 from integration_tests.model_test import MockPipeline
 
 from trufflepig import config
 import trufflepig.pigonduty as tppd
 
 
 @pytest.mark.skipif(config.PASSWORD is None, reason="needs posting key")
-def test_call_a_pig(steem_kwargs):
+def test_call_a_pig(steem):
     current_datetime = '2018-03-03-18:21:30'
 
     pipeline = MockPipeline()
-    tppd.call_a_pig(steem_kwargs=steem_kwargs,account='trufflepig',
+    tppd.call_a_pig(steem=steem,account='trufflepig',
                     pipeline=pipeline, topN_permalink='www.test.com',
                     current_datetime=current_datetime, hours=0.1,
                     sleep_time=0.1, overview_permalink='dsfd')

diff --git a/integration_tests/preprocessing_test.py b/integration_tests/preprocessing_test.py
@@ -6,7 +6,7 @@
 import trufflepig.preprocessing as tppp
 import trufflepig.bchain.getdata as tpgd
 from trufflepig.testutils.random_data import create_n_random_posts
-from trufflepig.testutils.pytest_fixtures import temp_dir, steem_kwargs
+from trufflepig.testutils.pytest_fixtures import temp_dir, steem
 
 
 def test_load_or_preproc(temp_dir):
@@ -26,25 +26,25 @@ def test_load_or_preproc(temp_dir):
     assert_frame_equal(frame, frame2)
 
 
-def test_load_or_preproc_with_real_data(steem_kwargs, temp_dir):
+def test_load_or_preproc_with_real_data(steem, temp_dir):
     filename = os.path.join(temp_dir, 'pptest.gz')
 
     start_datetime = pd.datetime.utcnow() - pd.Timedelta(days=14)
     end_datetime = start_datetime + pd.Timedelta(hours=2)
     posts = tpgd.get_all_posts_between_parallel(start_datetime,
                                                      end_datetime,
-                                                     steem_kwargs,
+                                                     steem,
                                                      stop_after=15)
     post_frame = pd.DataFrame(posts)
     bots = ['okankarol', 'bidseption', 'highvote', 'oguzhangazi', 'ottoman',]
     frame = tppp.load_or_preprocess(post_frame, filename,
-                                    steem_args_for_upvote=steem_kwargs,
+                                    steem_args_for_upvote=steem,
                                     ncores=5, chunksize=20, bots=bots)
 
     assert len(os.listdir(temp_dir)) == 1
 
     frame2 = tppp.load_or_preprocess(post_frame, filename,
-                                    steem_args_for_upvote=steem_kwargs,
+                                    steem_args_for_upvote=steem,
                                     ncores=5, chunksize=20, bots=bots)
 
     assert len(os.listdir(temp_dir)) == 1

diff --git a/scripts/do_cross_val.py b/scripts/do_cross_val.py
@@ -8,7 +8,7 @@
 import trufflepig.model as tpmo
 import trufflepig.preprocessing as tppp
 import trufflepig.bchain.getdata as tpgd
-import trufflepig.bchain.postdata as tppd
+from trufflepig.bchain.mpsteem import MPSteem
 from trufflepig import config
 
 
@@ -19,7 +19,7 @@ def main():
     logging.basicConfig(level=logging.INFO, format=format)
     directory = os.path.join(config.PROJECT_DIRECTORY, 'scraped_data')
 
-    steem = dict(nodes=config.NODES, no_broadcast=True)
+    steem = MPSteem(nodes=config.NODES, no_broadcast=True)
     current_datetime = pd.to_datetime('2018-02-01')
 
     crossval_filename = os.path.join(directory, 'xval_{}.gz'.format(current_datetime.date()))

diff --git a/scripts/doc2vec_cross_val.py b/scripts/doc2vec_cross_val.py
@@ -6,6 +6,7 @@
 import trufflepig.model as tpmo
 import trufflepig.preprocessing as tppp
 import trufflepig.bchain.getdata as tpgd
+from trufflepig.bchain.mpsteem import MPSteem
 from trufflepig import config
 
 
@@ -16,7 +17,7 @@ def main():
     logging.basicConfig(level=logging.INFO, format=format)
     directory = os.path.join(config.PROJECT_DIRECTORY, 'scraped_data')
 
-    steem = dict(nodes=config.NODES)
+    steem = MPSteem(nodes=config.NODES)
     current_datetime = '2018-02-01'
 
     crossval_filename = os.path.join(directory, 'xval_{}.gz'.format(current_datetime))

diff --git a/scripts/predict_post.py b/scripts/predict_post.py
@@ -7,6 +7,7 @@
 import trufflepig.model as tpmo
 import trufflepig.preprocessing as tppp
 import trufflepig.bchain.getdata as tpgd
+from trufflepig.bchain.mpsteem import MPSteem
 from trufflepig import config
 
 
@@ -43,8 +44,7 @@ def main():
     pipeline = tpmo.load_or_train_pipeline(None, model_directoy,
                                            current_datetime)
 
-    steem_kwargs = dict(nodes=config.NODES, no_broadcast=True)
-    steem = tpgd.check_and_convert_steem(steem_kwargs)
+    steem = MPSteem(nodes=config.NODES, no_broadcast=True)
     posts = tpgd.get_post_data([(author, permalink)], steem, {})
 
     posts = pd.DataFrame(posts)

diff --git a/trufflepig/__init__.py b/trufflepig/__init__.py
@@ -1 +1 @@
-__version__ = '0.5.0a'
+__version__ = '0.6.1a'
diff --git a/trufflepig/bchain/checkops.py b/trufflepig/bchain/checkops.py
@@ -97,10 +97,9 @@ def check_all_ops_between(start_datetime, end_datetime, steem,
     return comment_authors_and_permalinks
 
 
-def _check_all_ops_parallel(block_nums, steem_args, account,
+def _check_all_ops_parallel(block_nums, steem, account,
                                        stop_after=None):
     """Helper wrapper for multiprocessing"""
-    steem = tpbg.check_and_convert_steem(steem_args)
     comment_authors_and_permalinks = []
     for idx, block_num in enumerate(block_nums):
         authors_and_permalinks = check_all_ops_in_block(block_num, steem, account)
@@ -110,14 +109,13 @@ def _check_all_ops_parallel(block_nums, steem_args, account,
     return comment_authors_and_permalinks
 
 
-def check_all_ops_between_parallel(start_datetime, end_datetime, steem_args,
+def check_all_ops_between_parallel(start_datetime, end_datetime, steem,
                                    account, stop_after=None, ncores=8,
                                    chunksize=20, timeout=1200):
     """As above but in parallel with `ncores` jobs of `chunksize`.
 
     Waits for comments unitl `timeout`.
     """
-    steem = tpbg.check_and_convert_steem(steem_args)
     start_num, block_start_datetime = tpbg.find_nearest_block_num(start_datetime, steem)
     end_num, block_end_datetime = tpbg.find_nearest_block_num(end_datetime, steem)
 
@@ -137,7 +135,7 @@ def check_all_ops_between_parallel(start_datetime, end_datetime, steem_args,
     async_results = []
     for idx, chunk in enumerate(chunks):
         result = pool.apply_async(_check_all_ops_parallel,
-                                  args=(chunk, steem_args,
+                                  args=(chunk, steem,
                                         account,
                                         stop_after))
         async_results.append(result)
@@ -209,6 +207,8 @@ def get_parent_posts(comment_authors_and_permalinks, steem):
 
         except Exception as e:
             logger.exception(('Could not work with comment {} by '
-                              '{}').format(comment_permalink, comment_author))
+                              '{}. Reconnecting...'
+                              '').format(comment_permalink, comment_author))
+            steem.reconnect()
 
     return posts