<s> import binascii <EOL> import hashlib <EOL> import io <EOL> import os <EOL> import re <EOL> import struct <EOL> import zlib <EOL> from . import btreedb4 <EOL> from . import sbvj01 <EOL> from . import sbon <EOL> try : <EOL> import builtins <EOL> except : <EOL> import __builtin__ as builtins <EOL> try : <EOL> range = xrange <EOL> except : <EOL> pass <EOL> class KeyStore ( btreedb4 . FileBTreeDB4 ) : <EOL> """<STR_LIT>""" <EOL> def encode_key ( self , key ) : <EOL> return hashlib . sha256 ( key . encode ( '<STR_LIT:utf-8>' ) ) . digest ( ) <EOL> class KeyStoreCompressed ( KeyStore ) : <EOL> """<STR_LIT>""" <EOL> def deserialize_data ( self , data ) : <EOL> return zlib . decompress ( data ) <EOL> class CelestialChunks ( KeyStoreCompressed ) : <EOL> def deserialize_data ( self , data ) : <EOL> data = super ( CelestialChunks , self ) . deserialize_data ( data ) <EOL> stream = io . BytesIO ( data ) <EOL> return sbon . read_document ( stream ) <EOL> def initialize ( self ) : <EOL> super ( CelestialChunks , self ) . initialize ( ) <EOL> assert self . identifier == '<STR_LIT>' , '<STR_LIT>' <EOL> class Package ( KeyStore ) : <EOL> """<STR_LIT>""" <EOL> DIGEST_KEY = '<STR_LIT>' <EOL> INDEX_KEY = '<STR_LIT>' <EOL> def __init__ ( self , path ) : <EOL> super ( Package , self ) . __init__ ( path ) <EOL> self . _index = None <EOL> def encode_key ( self , key ) : <EOL> return super ( Package , self ) . encode_key ( key . lower ( ) ) <EOL> def get_digest ( self ) : <EOL> return self . get ( Package . DIGEST_KEY ) <EOL> def get_index ( self ) : <EOL> if self . _index : <EOL> return self . _index <EOL> stream = io . BytesIO ( self . get ( Package . INDEX_KEY ) ) <EOL> if self . identifier == '<STR_LIT>' : <EOL> self . _index = sbon . read_string_list ( stream ) <EOL> elif self . identifier == '<STR_LIT>' : <EOL> self . _index = sbon . read_string_digest_map ( stream ) <EOL> return self . _index <EOL> class VariantDatabase ( KeyStoreCompressed ) : <EOL> """<STR_LIT>""" <EOL> def deserialize_data ( self , data ) : <EOL> data = super ( VariantDatabase , self ) . deserialize_data ( data ) <EOL> stream = io . BytesIO ( data ) <EOL> return sbon . read_dynamic ( stream ) <EOL> def encode_key ( self , key ) : <EOL> return super ( VariantDatabase , self ) . encode_key ( key ) <EOL> def initialize ( self ) : <EOL> super ( VariantDatabase , self ) . initialize ( ) <EOL> assert self . identifier == '<STR_LIT>' , '<STR_LIT>' <EOL> class Player ( sbvj01 . FileSBVJ01 ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path ) : <EOL> super ( Player , self ) . __init__ ( path ) <EOL> self . name = None <EOL> def initialize ( self ) : <EOL> super ( Player , self ) . initialize ( ) <EOL> assert self . identifier == '<STR_LIT>' , '<STR_LIT>' <EOL> self . name = self . data [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] <EOL> class World ( btreedb4 . FileBTreeDB4 ) : <EOL> """<STR_LIT>""" <EOL> METADATA_KEY = ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> TILES_X = <NUM_LIT:32> <EOL> TILES_Y = <NUM_LIT:32> <EOL> TILES_PER_REGION = TILES_X * TILES_Y <EOL> def __init__ ( self , stream ) : <EOL> super ( World , self ) . __init__ ( stream ) <EOL> self . width = None <EOL> self . height = None <EOL> self . _metadata = None <EOL> self . _metadata_version = None <EOL> def deserialize_data ( self , data ) : <EOL> return zlib . decompress ( data ) <EOL> def encode_key ( self , key ) : <EOL> return struct . pack ( '<STR_LIT>' , * key ) <EOL> def get_entities ( self , x , y ) : <EOL> stream = io . BytesIO ( self . get ( ( <NUM_LIT:2> , x , y ) ) ) <EOL> return sbon . read_document_list ( stream ) <EOL> def get_metadata ( self ) : <EOL> if self . _metadata : <EOL> return self . _metadata , self . _metadata_version <EOL> data = self . get_raw ( World . METADATA_KEY ) <EOL> if self . _world_version >= <NUM_LIT:4> : <EOL> data = zlib . decompress ( data ) <EOL> stream = io . BytesIO ( data ) <EOL> self . width , self . height = struct . unpack ( '<STR_LIT>' , stream . read ( <NUM_LIT:8> ) ) <EOL> name , version , data = sbon . read_document ( stream ) <EOL> assert name == '<STR_LIT>' , '<STR_LIT>' <EOL> self . _metadata = data <EOL> self . _metadata_version = version <EOL> return data , version <EOL> def get_tiles ( self , x , y ) : <EOL> stream = io . BytesIO ( self . get ( ( <NUM_LIT:1> , x , y ) ) ) <EOL> unknown = stream . read ( <NUM_LIT:3> ) <EOL> return [ sbon . read_tile ( self . _metadata_version , stream ) <EOL> for _ in range ( World . TILES_PER_REGION ) ] <EOL> def initialize ( self ) : <EOL> super ( World , self ) . initialize ( ) <EOL> match = re . match ( '<STR_LIT>' , self . identifier ) <EOL> if not match : <EOL> raise ValueError ( '<STR_LIT>' % ( self . identifier , ) ) <EOL> self . _world_version = int ( match . group ( <NUM_LIT:1> ) ) <EOL> class FailedWorld ( World ) : <EOL> def __init__ ( self , stream ) : <EOL> super ( FailedWorld , self ) . __init__ ( stream ) <EOL> self . repair = True <EOL> def get_metadata ( self ) : <EOL> try : <EOL> stream = io . BytesIO ( self . get_raw ( World . METADATA_KEY ) ) <EOL> except : <EOL> stream = btreedb4 . LeafReader . last_buffer <EOL> stream . seek ( <NUM_LIT:0> ) <EOL> unknown_1 , unknown_2 = struct . unpack ( '<STR_LIT>' , stream . read ( <NUM_LIT:8> ) ) <EOL> name , version , data = sbon . read_document ( stream , True ) <EOL> assert name == '<STR_LIT>' , '<STR_LIT>' <EOL> self . _metadata = data <EOL> self . _metadata_version = version <EOL> return data , version <EOL> EXTENSION_TO_CLASS = dict ( <EOL> chunks = CelestialChunks , <EOL> clientcontext = sbvj01 . FileSBVJ01 , <EOL> dat = sbvj01 . FileSBVJ01 , <EOL> db = VariantDatabase , <EOL> fail = FailedWorld , <EOL> modpak = Package , <EOL> pak = Package , <EOL> player = Player , <EOL> shipworld = World , <EOL> world = World , <EOL> ) <EOL> def open ( path , override_extension = None ) : <EOL> """<STR_LIT>""" <EOL> extension = override_extension or os . path . splitext ( path ) [ <NUM_LIT:1> ] [ <NUM_LIT:1> : ] <EOL> return read_stream ( builtins . open ( path , '<STR_LIT:rb>' ) , extension ) <EOL> def read_stream ( stream , extension ) : <EOL> cls = EXTENSION_TO_CLASS . get ( extension ) <EOL> if not cls : <EOL> raise ValueError ( '<STR_LIT>' % extension ) <EOL> file = cls ( stream ) <EOL> file . initialize ( ) <EOL> return file </s>
