Coverage for src / utils / store_util.py: 0%

1import logging 

2import os 

3 

4import pandas as pd 

5 

6from src import portfolio, collection_store, battle_store, season_balances_info, season_battle_info, land, land_pools 

7from src.api import spl 

8from src.configuration import store, config 

9from src.static.static_values_enum import Format 

10from src.utils import progress_util, spl_util 

11 

12 

13def update_season_end_dates(): 

14 if store.season_end_dates.empty: 

15 from_season_id = 1 

16 else: 

17 from_season_id = store.season_end_dates.id.max() + 1 

18 

19 till_season_id = spl.get_current_season()['id'] 

20 # logging.info('Update season end dates for '' + str(till_season_id) + '' seasons') 

21 for season_id in range(from_season_id, till_season_id + 1): 

22 logging.info('Update season end date for season: ' + str(season_id)) 

23 

24 store.season_end_dates = pd.concat([store.season_end_dates, 

25 spl.get_season_end_time(season_id)], 

26 ignore_index=True) 

27 save_stores() 

28 

29 

30def get_store_names(): 

31 stores_arr = [] 

32 for store_name, _store in store.__dict__.items(): 

33 if isinstance(_store, pd.DataFrame): 

34 stores_arr.append(store_name) 

35 return stores_arr 

36 

37 

38def validate_store_name(name): 

39 for store_name, _store in store.__dict__.items(): 

40 if isinstance(_store, pd.DataFrame): 

41 if name == store_name: 

42 return True 

43 return False 

44 

45 

46def get_store_file(name): 

47 return os.path.join(config.store_dir, str(name + config.file_extension)) 

48 

49 

50def load_stores(): 

51 for store_name in get_store_names(): 

52 store_file = get_store_file(store_name) 

53 if os.path.isfile(store_file): 

54 # TODO investigate the low_memory 

55 # DtypeWarning: Columns (6,14) have mixed types. Specify dtype option on import or set low_memory=False. 

56 # store.__dict__[store_name] = pd.read_csv(store_file, index_col=0) 

57 store.__dict__[store_name] = pd.read_csv(store_file, index_col=0, low_memory=False) 

58 

59 

60def save_stores(): 

61 for store_name in get_store_names(): 

62 store_file = get_store_file(store_name) 

63 store.__dict__[store_name].sort_index().to_csv(store_file) 

64 

65 

66def save_single_store(store_name): 

67 if validate_store_name(store_name): 

68 store_file = get_store_file(store_name) 

69 store.__dict__[store_name].sort_index().to_csv(store_file) 

70 else: 

71 logging.error('Invalid store name') 

72 

73 

74def get_account_names(): 

75 if store.accounts.empty: 

76 return list() 

77 else: 

78 return store.accounts.account_name.tolist() 

79 

80 

81def get_played_players(account): 

82 df = store.battle_big.copy() 

83 

84 if not df.empty: 

85 df = df.loc[df.account == account] 

86 if not df.empty: 

87 return df.opponent.sort_values().unique().tolist() 

88 

89 return list() 

90 

91 

92def get_first_account_name(): 

93 if store.accounts.empty: 

94 return '' 

95 else: 

96 return store.accounts.values[0][0] 

97 

98 

99def add_account(account_name): 

100 new_account = pd.DataFrame({'account_name': account_name}, index=[0]) 

101 if store.accounts.empty: 

102 store.accounts = pd.concat([store.accounts, new_account], ignore_index=True) 

103 else: 

104 if store.accounts.loc[(store.accounts.account_name == account_name)].empty: 

105 store.accounts = pd.concat([store.accounts, new_account], ignore_index=True) 

106 save_stores() 

107 return store.accounts.account_name.tolist() 

108 

109 

110def remove_account_from_store(store_name, search_column, account_name): 

111 _store = store.__dict__[store_name] 

112 if search_column in _store.columns.tolist(): 

113 rows = _store.loc[(_store[search_column] == account_name)] 

114 if not rows.empty: 

115 _store = _store.drop(rows.index) 

116 return _store 

117 

118 

119def remove_data(account_name): 

120 for store_name in get_store_names(): 

121 store.__dict__[store_name] = remove_account_from_store(store_name, 'account_name', account_name) 

122 store.__dict__[store_name] = remove_account_from_store(store_name, 'account', account_name) 

123 store.__dict__[store_name] = remove_account_from_store(store_name, 'player', account_name) 

124 store.__dict__[store_name] = remove_account_from_store(store_name, 'username', account_name) 

125 

126 save_stores() 

127 

128 

129def remove_account(account_name): 

130 if store.accounts.empty: 

131 return list() 

132 else: 

133 account_row = store.accounts.loc[(store.accounts.account_name == account_name)] 

134 if not account_row.empty: 

135 remove_data(account_name) 

136 

137 return store.accounts.account_name.tolist() 

138 

139 

140def get_last_portfolio_selection(): 

141 if store.portfolio.empty or store.view_portfolio_accounts.empty: 

142 return list() 

143 else: 

144 # Remove users that are no longer in 

145 curr_users = store.portfolio.account_name.unique().tolist() 

146 mask = (store.view_portfolio_accounts.account_name.isin(curr_users)) 

147 return store.view_portfolio_accounts.loc[mask].account_name.tolist() 

148 

149 

150def get_seasons_played_list(): 

151 input_df = store.battle_big.copy() 

152 if not input_df.empty: 

153 first_date = pd.to_datetime(input_df.created_date).min() 

154 

155 temp_end_dates = store.season_end_dates.copy() 

156 temp_end_dates.end_date = pd.to_datetime(temp_end_dates.end_date) 

157 

158 last_id = temp_end_dates.loc[(temp_end_dates.end_date > first_date)].id.min() 

159 return temp_end_dates.sort_values('id', ascending=False).loc[(temp_end_dates.id >= last_id - 1)].id.to_list() 

160 else: 

161 return list() 

162 

163 

164def get_season_values(df, season_id, users, season_id_column='season_id'): 

165 if not df.empty: 

166 return df.loc[(df.player.isin(users)) & (df[season_id_column] == season_id)].copy() 

167 else: 

168 return df 

169 

170 

171def is_last_season_processed(account, season_id): 

172 if not (store.season_sps.empty or store.season_sps.loc[store.season_sps.player == account].empty): 

173 last_season = store.season_sps.loc[store.season_sps.player == account].season_id.max() 

174 if last_season == season_id: 

175 return True 

176 return False 

177 

178 

179def is_new_account(account): 

180 if store.season_sps.empty or store.season_sps.loc[store.season_sps.player == account].empty: 

181 return True 

182 return False 

183 

184 

185def update_battle_log(): 

186 progress_util.set_daily_title('Update collection') 

187 collection_store.update_collection() 

188 

189 progress_util.set_daily_title('Update battles') 

190 battle_store.process_battles() 

191 

192 progress_util.set_daily_title('Update portfolio') 

193 portfolio.update_portfolios() 

194 

195 progress_util.set_daily_title('Update land') 

196 land.update_land_data() 

197 land_pools.update_pools() 

198 

199 save_stores() 

200 progress_util.update_daily_msg('Done') 

201 

202 

203def update_season_log(): 

204 progress_util.set_season_title('Season update process initiated') 

205 update_season_end_dates() 

206 current_season_data = spl.get_current_season() 

207 for account in get_account_names(): 

208 if get_token_dict(account): 

209 progress_util.update_season_msg('Start season update for: ' + str(account)) 

210 last_season_id = current_season_data['id'] - 1 

211 current_season_id = current_season_data['id'] 

212 if not is_last_season_processed(account, last_season_id): 

213 if spl_util.is_season_reward_claimed(account, last_season_id): 

214 season_balances_info.update_balances_store(account, current_season_id) 

215 store.season_modern_battle_info = season_battle_info.get_season_battles( 

216 account, 

217 store.season_modern_battle_info.copy(), 

218 Format.modern, 

219 current_season_id) 

220 store.season_wild_battle_info = season_battle_info.get_season_battles( 

221 account, 

222 store.season_wild_battle_info.copy(), 

223 Format.wild, 

224 current_season_id) 

225 else: 

226 progress_util.update_season_msg('No seasons to process for: ' + str(account)) 

227 else: 

228 progress_util.update_season_msg('Skip... no token found. Check config page.') 

229 

230 save_stores() 

231 progress_util.set_season_title('Season update done') 

232 progress_util.update_season_msg('Done') 

233 

234 

235def update_data(battle_update=True, season_update=False): 

236 try: 

237 if not spl.is_maintenance_mode(): 

238 if battle_update: 

239 update_battle_log() 

240 

241 if season_update: 

242 update_season_log() 

243 else: 

244 logging.info('Splinterlands server is in maintenance mode skip this update cycle') 

245 except Exception as e: 

246 logging.error('Exception during update data') 

247 logging.exception(e) 

248 

249 

250def get_token_dict(username): 

251 if not store.secrets.empty: 

252 df = store.secrets.loc[store.secrets.username == username] 

253 if not df.empty: 

254 row = df.iloc[0] 

255 params = { 

256 'username': row.username, 

257 'token': row.token, 

258 } 

259 return params 

260 

261 return None