eth-stat-syncer/eth_stat_syncer/runnable/tracker.py

137 lines
4.5 KiB
Python

# standard imports
import sys
import os
import logging
import argparse
import uuid
# external imports
import confini
from chainsyncer.store.fs import SyncFsStore
from chainsyncer.driver.chain_interface import ChainInterfaceDriver
from chainsyncer.filter import SyncFilter
from chainsyncer.error import NoBlockForYou
from chainlib.chain import ChainSpec
from chainlib.eth.connection import EthHTTPConnection
from chainlib.interface import ChainInterface
from chainlib.eth.block import (
block_by_number,
Block,
block_latest,
)
from chainlib.eth.tx import (
receipt,
Tx,
)
# local imports
from eth_stat_syncer.store import (
GasAggregator,
RunStore,
)
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
script_dir = os.path.realpath(os.path.dirname(__file__))
exec_dir = os.path.realpath(os.getcwd())
default_config_dir = os.environ.get('confini_dir', os.path.join(exec_dir, 'config'))
argparser = argparse.ArgumentParser()
argparser.add_argument('-p', '--provider', dest='p', type=str, help='rpc provider')
argparser.add_argument('-c', '--config', dest='c', default=default_config_dir, type=str, help='rpc provider')
argparser.add_argument('-i', '--chain-spec', dest='i', default='evm:ethereum:1', type=str, help='chain spec')
argparser.add_argument('--moving', action='append', default=[], type=int, help='add moving average')
argparser.add_argument('--offset', type=int, default=0, help='Start sync on this block')
argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
argparser.add_argument('--cache-dir', dest='cache_dir', type=str, help='Directory to store gas cache')
argparser.add_argument('--state-dir', dest='state_dir', default=exec_dir, type=str, help='Directory to store sync state')
argparser.add_argument('--session-id', dest='session_id', type=str, help='Use state from specified session id')
argparser.add_argument('-v', action='store_true', help='be verbose')
argparser.add_argument('-vv', action='store_true', help='be more verbose')
args = argparser.parse_args()
if args.vv:
logging.getLogger().setLevel(logging.DEBUG)
elif args.v:
logging.getLogger().setLevel(logging.INFO)
config = confini.Config(args.c, args.env_prefix)
config.process()
# override args
args_override = {
'CHAIN_SPEC': getattr(args, 'i'),
'RPC_PROVIDER': getattr(args, 'p'),
}
config.dict_override(args_override, 'cli flag')
config.add(args.offset, '_SYNC_OFFSET', True)
config.add(os.path.realpath(args.state_dir), '_STATE_DIR', True)
config.add(args.cache_dir, '_CACHE_DIR', True)
config.add(args.session_id, '_SESSION_ID', True)
config.add(args.moving, '_MOVING', True)
logg.debug('loaded config: {}\n'.format(config))
chain_spec = ChainSpec.from_chain_str(config.get('CHAIN_SPEC'))
conn = EthHTTPConnection(args.p)
if config.get('_SESSION_ID') == None:
config.add(str(uuid.uuid4()), '_SESSION_ID', True)
class GasPriceFilter(SyncFilter):
def __init__(self, chain_spec, gas_aggregator):
self.chain_spec = chain_spec
self.gas_aggregator = gas_aggregator
def filter(self, conn, block, tx, db_session=None):
self.gas_aggregator.put(tx.gas_price)
return False
class EthChainInterface(ChainInterface):
def __init__(self):
self._block_by_number = block_by_number
self._block_from_src = Block.from_src
self._tx_receipt = receipt
self._src_normalize = Tx.src_normalize
def main():
gas_store = RunStore(basedir=config.get('_CACHE_DIR'))
cap = 360
try:
v = max(config.get('_MOVING'))
if v > cap:
cap = v
except ValueError:
pass
gas_aggregator = GasAggregator(gas_store, cap, moving=config.get('_MOVING'))
gas_filter = GasPriceFilter(chain_spec, gas_aggregator)
start_block = 0
if config.get('_SYNC_OFFSET') != None:
start_block = config.get('_SYNC_OFFSET')
else:
o = block_latest()
r = conn.do(o)
n = int(r, 16)
start_block = n
logg.info('block height at start {}'.format(start_block))
chain_interface = EthChainInterface()
sync_store = SyncFsStore(config.get('_STATE_DIR'), session_id=config.get('_SESSION_ID'))
sync_store.register(gas_filter)
drv = ChainInterfaceDriver(sync_store, chain_interface, offset=start_block, target=-1, block_callback=gas_aggregator.block_callback)
r = drv.run(conn)
if __name__ == '__main__':
main()