@ -33,14 +33,26 @@ STRUCT_DOUBLE = struct.Struct('<d')
# Functions to retrieve data from the Steam client's local filesystem cache
def _stat_default ( stat ) :
mappers = {
STAT_TYPE_INT : int ,
STAT_TYPE_FLOAT : float ,
STAT_TYPE_AVGRATE : float ,
STAT_TYPE_BITS : int
}
return mappers [ stat [ ' type_int ' ] ] ( stat . get ( ' Default ' , ' 0 ' ) )
def load_schema ( base_dir , appid ) :
path_name = pathlib . Path ( base_dir , ' appcache ' , ' stats ' )
schema_file = ' UserGameStatsSchema_{}.bin ' . format ( appid )
with path_name . joinpath ( schema_file ) . open ( ' rb ' ) as fp :
return vdf . binary_loads ( fp . read ( ) )
def load_cache ( base_dir , appid , userid = None ) :
if userid is None :
return { ' cache ' : { ' crc ' : 0 , ' PendingChanges ' : 0 } }
path_name = pathlib . Path ( base_dir , ' appcache ' , ' stats ' )
cache_file = ' UserGameStats_{}_{}.bin ' . format ( userid , appid )
with path_name . joinpath ( cache_file ) . open ( ' rb ' ) as fp :
return vdf . binary_loads ( fp . read ( ) )
def get_global_stat_names ( appid , schema ) :
global_stat_names = [ ]
for stat in schema [ appid ] [ ' stats ' ] . values ( ) :
if stat . get ( ' aggregated ' , ' 0 ' ) == ' 1 ' :
global_stat_names . append ( stat [ ' name ' ] )
return global_stat_names
def get_local_stats ( base_dir , appid , userid = None ) :
"""
@ -65,20 +77,17 @@ def get_local_stats(base_dir, appid, userid=None):
"""
local_stats = { FOLDER_ACH : { } , FOLDER_GLOBAL : { } , FOLDER_STAT : { } }
path_name = pathlib . Path ( base_dir , ' appcache ' , ' stats ' )
# Read in schema file
schema_file = ' UserGameStatsSchema_{}.bin ' . format ( appid )
with path_name . joinpath ( schema_file ) . open ( ' rb ' ) as fh :
schema = vdf . binary_loads ( fh . read ( ) )
schema = load_schema ( base_dir , appid )
cache = load_cache ( base_dir , appid , userid )
# Read in cache file
if userid is not None :
cache_file = ' UserGameStats_{}_{}.bin ' . format ( userid , appid )
with path_name . joinpath ( cache_file ) . open ( ' rb ' ) as fh :
cache = vdf . binary_loads ( fh . read ( ) )
else :
cache = { ' cache ' : { ' crc ' : 0 , ' PendingChanges ' : 0 } }
def _stat_default ( stat ) :
mappers = {
STAT_TYPE_INT : int ,
STAT_TYPE_FLOAT : float ,
STAT_TYPE_AVGRATE : float ,
STAT_TYPE_BITS : int
}
return mappers [ stat [ ' type_int ' ] ] ( stat . get ( ' Default ' , ' 0 ' ) )
for stat in schema [ appid ] [ ' stats ' ] . values ( ) :
# Get default and cached stat values
@ -270,18 +279,35 @@ def dump_history(data, startdate, enddate):
f [ dates . index ( record [ ' date ' ] ) ] = int ( record [ ' total ' ] )
csvwriter . writerow ( [ stat ] + f )
def dump_stat_defaults ( appid , schema ) :
type_labels = {
STAT_TYPE_INT : ' int ' ,
STAT_TYPE_FLOAT : ' float ' ,
STAT_TYPE_AVGRATE : ' avgrate '
}
with open ( ' stats.txt ' , ' w ' ) as fp :
for stat in schema [ appid ] [ ' stats ' ] . values ( ) :
s_type = type_labels . get ( stat [ ' type_int ' ] )
if s_type is None :
continue
s_default = stat . get ( ' Default ' , ' 0 ' )
fp . write ( ' {}={}={} \n ' . format ( stat [ ' name ' ] , s_type , s_default ) )
# Program interface
def main ( ) :
import sys
if len ( sys . argv ) > 2 :
# Test writing default stat values
schema = load_schema ( sys . argv [ 1 ] , sys . argv [ 2 ] )
dump_stat_defaults ( sys . argv [ 2 ] , schema )
# Test fetching local player stats
if len ( sys . argv ) > 3 :
stats = get_local_stats ( sys . argv [ 1 ] , sys . argv [ 2 ] , sys . argv [ 3 ] )
else :
stats = get_local_stats ( sys . argv [ 1 ] , sys . argv [ 2 ] )
# Test fetching global stats Web API use
global_stat_names = [ n for n in stats [ FOLDER_GLOBAL ] . keys ( ) ]
global_stat_names = get_global_stat_names ( sys . argv [ 2 ] , schema )
global_stats = get_global_stats ( sys . argv [ 2 ] , global_stat_names )
for stat_name , stat_data in global_stats . items ( ) :
if stat_data is None :
@ -299,8 +325,8 @@ def main():
def main_history ( ) :
import sys
if len ( sys . argv ) > 4 :
stats = get_local_stats ( sys . argv [ 1 ] , sys . argv [ 2 ] )
global_stat_names = [ n for n in stats [ FOLDER_GLOBAL ] . keys ( ) ]
schema = load_schema ( sys . argv [ 1 ] , sys . argv [ 2 ] )
global_stat_names = get_global_stat_names ( sys . argv [ 2 ] , schema )
# Make multiple calls to get daily totals
global_stat_hist = { }
def divide_names ( l , n ) :