<s> """<STR_LIT>""" <EOL> import re <EOL> from ctypes import c_int , c_char_p , byref , sizeof , c_uint16 , c_int32 , c_byte <EOL> from ctypes import c_void_p <EOL> import logging <EOL> import snap7 <EOL> from snap7 import six <EOL> from snap7 . snap7types import S7Object , buffer_type , buffer_size , BlocksList <EOL> from snap7 . snap7types import TS7BlockInfo , param_types , cpu_statuses <EOL> from snap7 . common import check_error , load_library , ipv4 <EOL> from snap7 . snap7exceptions import Snap7Exception <EOL> logger = logging . getLogger ( __name__ ) <EOL> def error_wrap ( func ) : <EOL> """<STR_LIT>""" <EOL> def f ( * args , ** kw ) : <EOL> code = func ( * args , ** kw ) <EOL> check_error ( code , context = "<STR_LIT>" ) <EOL> return f <EOL> class Client ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . library = load_library ( ) <EOL> self . pointer = False <EOL> self . create ( ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> self . library . Cli_Create . restype = c_void_p <EOL> self . pointer = S7Object ( self . library . Cli_Create ( ) ) <EOL> def destroy ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> return self . library . Cli_Destroy ( byref ( self . pointer ) ) <EOL> def plc_stop ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> return self . library . Cli_PlcStop ( self . pointer ) <EOL> def plc_cold_start ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> return self . library . Cli_PlcColdStart ( self . pointer ) <EOL> def plc_hot_start ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> return self . library . Cli_PlcColdStart ( self . pointer ) <EOL> def get_cpu_state ( self ) : <EOL> """<STR_LIT>""" <EOL> state = c_int ( <NUM_LIT:0> ) <EOL> self . library . Cli_GetPlcStatus ( self . pointer , byref ( state ) ) <EOL> try : <EOL> status_string = cpu_statuses [ state . value ] <EOL> except KeyError : <EOL> status_string = None <EOL> if not status_string : <EOL> raise Snap7Exception ( "<STR_LIT>" % state . value ) <EOL> logging . debug ( "<STR_LIT>" % status_string ) <EOL> return status_string <EOL> def get_cpu_info ( self ) : <EOL> """<STR_LIT>""" <EOL> info = snap7 . snap7types . S7CpuInfo ( ) <EOL> result = self . library . Cli_GetCpuInfo ( self . pointer , byref ( info ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return info <EOL> @ error_wrap <EOL> def disconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> return self . library . Cli_Disconnect ( self . pointer ) <EOL> @ error_wrap <EOL> def connect ( self , address , rack , slot , tcpport = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" % ( address , tcpport , <EOL> rack , slot ) ) <EOL> self . set_param ( snap7 . snap7types . RemotePort , tcpport ) <EOL> return self . library . Cli_ConnectTo ( <EOL> self . pointer , c_char_p ( six . b ( address ) ) , <EOL> c_int ( rack ) , c_int ( slot ) ) <EOL> def db_read ( self , db_number , start , size ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % <EOL> ( db_number , start , size ) ) <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ snap7 . snap7types . S7WLByte ] <EOL> data = ( type_ * size ) ( ) <EOL> result = ( self . library . Cli_DBRead ( <EOL> self . pointer , db_number , start , size , <EOL> byref ( data ) ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( data ) <EOL> @ error_wrap <EOL> def db_write ( self , db_number , start , data ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> size = len ( data ) <EOL> cdata = ( type_ * size ) . from_buffer ( data ) <EOL> logger . debug ( "<STR_LIT>" % <EOL> ( db_number , start , size , data ) ) <EOL> return self . library . Cli_DBWrite ( self . pointer , db_number , start , size , <EOL> byref ( cdata ) ) <EOL> def full_upload ( self , _type , block_num ) : <EOL> """<STR_LIT>""" <EOL> _buffer = buffer_type ( ) <EOL> size = c_int ( sizeof ( _buffer ) ) <EOL> block_type = snap7 . snap7types . block_types [ _type ] <EOL> result = self . library . Cli_FullUpload ( self . pointer , block_type , <EOL> block_num , byref ( _buffer ) , <EOL> byref ( size ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( _buffer ) , size . value <EOL> def upload ( self , block_num ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % ( block_num ) ) <EOL> block_type = snap7 . snap7types . block_types [ '<STR_LIT>' ] <EOL> _buffer = buffer_type ( ) <EOL> size = c_int ( sizeof ( _buffer ) ) <EOL> result = self . library . Cli_Upload ( self . pointer , block_type , block_num , <EOL> byref ( _buffer ) , byref ( size ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> logger . info ( '<STR_LIT>' % size ) <EOL> return bytearray ( _buffer ) <EOL> @ error_wrap <EOL> def download ( self , data , block_num = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> type_ = c_byte <EOL> size = len ( data ) <EOL> cdata = ( type_ * len ( data ) ) . from_buffer ( data ) <EOL> result = self . library . Cli_Download ( self . pointer , block_num , <EOL> byref ( cdata ) , size ) <EOL> return result <EOL> def db_get ( self , db_number ) : <EOL> """<STR_LIT>""" <EOL> logging . debug ( "<STR_LIT>" % db_number ) <EOL> _buffer = buffer_type ( ) <EOL> result = self . library . Cli_DBGet ( <EOL> self . pointer , db_number , byref ( _buffer ) , <EOL> byref ( c_int ( buffer_size ) ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( _buffer ) <EOL> def read_area ( self , area , dbnumber , start , size ) : <EOL> """<STR_LIT>""" <EOL> assert area in snap7 . snap7types . areas . values ( ) <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( area , dbnumber , start , size , wordlen ) ) <EOL> data = ( type_ * size ) ( ) <EOL> result = self . library . Cli_ReadArea ( self . pointer , area , dbnumber , start , <EOL> size , wordlen , byref ( data ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( data ) <EOL> @ error_wrap <EOL> def write_area ( self , area , dbnumber , start , data ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> size = len ( data ) <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( area , dbnumber , start , size , type_ ) ) <EOL> cdata = ( type_ * len ( data ) ) . from_buffer ( data ) <EOL> return self . library . Cli_WriteArea ( self . pointer , area , dbnumber , start , <EOL> size , wordlen , byref ( cdata ) ) <EOL> def read_multi_vars ( self , items ) : <EOL> """<STR_LIT>""" <EOL> result = self . library . Cli_ReadMultiVars ( self . pointer , byref ( items ) , <EOL> c_int32 ( len ( items ) ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return result , items <EOL> def list_blocks ( self ) : <EOL> """<STR_LIT>""" <EOL> logging . debug ( "<STR_LIT>" ) <EOL> blocksList = BlocksList ( ) <EOL> result = self . library . Cli_ListBlocks ( self . pointer , byref ( blocksList ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" % blocksList ) <EOL> return blocksList <EOL> def list_blocks_of_type ( self , blocktype , size ) : <EOL> """<STR_LIT>""" <EOL> blocktype = snap7 . snap7types . block_types . get ( blocktype ) <EOL> if not blocktype : <EOL> raise Snap7Exception ( "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" % <EOL> ( blocktype , size ) ) <EOL> data = ( c_int * <NUM_LIT:10> ) ( ) <EOL> count = c_int ( size ) <EOL> result = self . library . Cli_ListBlocksOfType ( <EOL> self . pointer , blocktype , <EOL> byref ( data ) , <EOL> byref ( count ) ) <EOL> logging . debug ( "<STR_LIT>" % count ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return data <EOL> def get_block_info ( self , blocktype , db_number ) : <EOL> """<STR_LIT>""" <EOL> blocktype = snap7 . snap7types . block_types . get ( blocktype ) <EOL> if not blocktype : <EOL> raise Snap7Exception ( "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" % <EOL> ( db_number , blocktype ) ) <EOL> data = TS7BlockInfo ( ) <EOL> result = self . library . Cli_GetAgBlockInfo ( <EOL> self . pointer , blocktype , <EOL> db_number , byref ( data ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return data <EOL> @ error_wrap <EOL> def set_session_password ( self , password ) : <EOL> """<STR_LIT>""" <EOL> assert len ( password ) <= <NUM_LIT:8> , '<STR_LIT>' <EOL> return self . library . Cli_SetSessionPassword ( self . pointer , <EOL> c_char_p ( six . b ( password ) ) ) <EOL> @ error_wrap <EOL> def clear_session_password ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . library . Cli_ClearSessionPassword ( self . pointer ) <EOL> def set_connection_params ( self , address , local_tsap , remote_tsap ) : <EOL> """<STR_LIT>""" <EOL> assert re . match ( ipv4 , address ) , '<STR_LIT>' % address <EOL> result = self . library . Cli_SetConnectionParams ( self . pointer , address , <EOL> c_uint16 ( local_tsap ) , <EOL> c_uint16 ( remote_tsap ) ) <EOL> if result != <NUM_LIT:0> : <EOL> raise Snap7Exception ( "<STR_LIT>" ) <EOL> def set_connection_type ( self , connection_type ) : <EOL> """<STR_LIT>""" <EOL> result = self . library . Cli_SetConnectionType ( self . pointer , <EOL> c_uint16 ( connection_type ) ) <EOL> if result != <NUM_LIT:0> : <EOL> raise Snap7Exception ( "<STR_LIT>" ) <EOL> def get_connected ( self ) : <EOL> """<STR_LIT>""" <EOL> connected = c_int32 ( ) <EOL> result = self . library . Cli_GetConnected ( self . pointer , byref ( connected ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bool ( connected ) <EOL> def ab_read ( self , start , size ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> data = ( type_ * size ) ( ) <EOL> logging . debug ( "<STR_LIT>" % ( start , size ) ) <EOL> result = self . library . Cli_ABRead ( self . pointer , start , size , <EOL> byref ( data ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( data ) <EOL> def ab_write ( self , start , data ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> size = len ( data ) <EOL> cdata = ( type_ * size ) . from_buffer ( data ) <EOL> logging . debug ( "<STR_LIT>" % ( start , size ) ) <EOL> return self . library . Cli_ABWrite ( <EOL> self . pointer , start , size , byref ( cdata ) ) <EOL> def as_ab_read ( self , start , size ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> data = ( type_ * size ) ( ) <EOL> logging . debug ( "<STR_LIT>" % ( start , size ) ) <EOL> result = self . library . Cli_AsABRead ( self . pointer , start , size , <EOL> byref ( data ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( data ) <EOL> def as_ab_write ( self , start , data ) : <EOL> """<STR_LIT>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> size = len ( data ) <EOL> cdata = ( type_ * size ) . from_buffer ( data ) <EOL> logging . debug ( "<STR_LIT>" % ( start , size ) ) <EOL> return self . library . Cli_AsABWrite ( <EOL> self . pointer , start , size , byref ( cdata ) ) <EOL> @ error_wrap <EOL> def as_compress ( self , time ) : <EOL> """<STR_LIT>""" <EOL> return self . library . Cli_AsCompress ( self . pointer , time ) <EOL> def copy_ram_to_rom ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . library . Cli_AsCopyRamToRom ( self . pointer ) <EOL> def as_ct_read ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . library . Cli_AsCTRead ( self . pointer ) <EOL> def as_ct_write ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . library . Cli_AsCTWrite ( self . pointer ) <EOL> def as_db_fill ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> return self . library . Cli_AsDBFill ( self . pointer ) <EOL> def as_db_get ( self , db_number ) : <EOL> """<STR_LIT>""" <EOL> logging . debug ( "<STR_LIT>" % db_number ) <EOL> _buffer = buffer_type ( ) <EOL> result = self . library . Cli_AsDBGet ( self . pointer , db_number , <EOL> byref ( _buffer ) , <EOL> byref ( c_int ( buffer_size ) ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( _buffer ) <EOL> def as_db_read ( self , db_number , start , size ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % <EOL> ( db_number , start , size ) ) <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ snap7 . snap7types . S7WLByte ] <EOL> data = ( type_ * size ) ( ) <EOL> result = ( self . library . Cli_AsDBRead ( self . pointer , db_number , start , <EOL> size , byref ( data ) ) ) <EOL> check_error ( result , context = "<STR_LIT>" ) <EOL> return bytearray ( data ) <EOL> def as_db_write ( self , db_number , start , data ) : <EOL> """<STR_LIT:U+0020>""" <EOL> wordlen = snap7 . snap7types . S7WLByte <EOL> type_ = snap7 . snap7types . wordlen_to_ctypes [ wordlen ] <EOL> size = len ( data ) <EOL> cdata = ( type_ * size ) . from_buffer ( data ) <EOL> logger . debug ( "<STR_LIT>" % <EOL> ( db_number , start , size , data ) ) <EOL> return self . library . Cli_AsDBWrite ( <EOL> self . pointer , db_number , start , size , <EOL> byref ( cdata ) ) <EOL> @ error_wrap <EOL> def as_download ( self , data , block_num = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> size = len ( data ) <EOL> type_ = c_byte * len ( data ) <EOL> cdata = type_ . from_buffer ( data ) <EOL> return self . library . Cli_AsDownload ( self . pointer , block_num , <EOL> byref ( cdata ) , size ) <EOL> @ error_wrap <EOL> def compress ( self , time ) : <EOL> """<STR_LIT>""" <EOL> return self . library . Cli_Compress ( self . pointer , time ) <EOL> @ error_wrap <EOL> def set_param ( self , number , value ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % ( number , value ) ) <EOL> type_ = param_types [ number ] <EOL> return self . library . Cli_SetParam ( self . pointer , number , <EOL> byref ( type_ ( value ) ) ) <EOL> def get_param ( self , number ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" % number ) <EOL> type_ = param_types [ number ] <EOL> value = type_ ( ) <EOL> code = self . library . Cli_GetParam ( self . pointer , c_int ( number ) , <EOL> byref ( value ) ) <EOL> check_error ( code ) <EOL> return value . value <EOL> def get_pdu_length ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> requested_ = c_uint16 ( ) <EOL> negotiated_ = c_uint16 ( ) <EOL> code = self . library . Cli_GetPduLength ( self . pointer , byref ( requested_ ) , byref ( negotiated_ ) ) <EOL> check_error ( code ) <EOL> return negotiated_ . value </s>
