Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
21:0x6B9A,
22:0xBFBF,
#MLE
23:0xF27C,#PDU22
24:0x50B0,
25:0x9A6B,
26:0xBFBF,
#MBE
27:0xBFBF,#PDU26
28:0x6B9A,
29:0xB050,
30:0x7CF2
})
store = ModbusSlaveContext(di=block, co=block, hr=block, ir=block)
context = ModbusServerContext(slaves=store, single=True)
# ----------------------------------------------------------------------- #
# initialize the server information
# ----------------------------------------------------------------------- #
# If you don't set this or any fields, they are defaulted to empty strings.
# ----------------------------------------------------------------------- #
identity = ModbusDeviceIdentification()
identity.VendorName = 'Pymodbus'
identity.ProductCode = 'PM'
identity.VendorUrl = 'http://github.com/riptideio/pymodbus/'
identity.ProductName = 'Pymodbus Server'
identity.ModelName = 'Pymodbus Server'
identity.MajorMinorRevision = '1.5'
def _parse(self):
""" Parses the config file and creates a server context """
try:
handle = pickle.load(self.file)
dsd = handle['di']
csd = handle['ci']
hsd = handle['hr']
isd = handle['ir']
except KeyError:
raise ConfigurationException("Invalid Configuration")
slave = ModbusSlaveContext(d=dsd, c=csd, h=hsd, i=isd)
return ModbusServerContext(slaves=slave)
def _parse(self):
""" Parses the config file and creates a server context """
try:
handle = pickle.load(self.file)
dsd = handle['di']
csd = handle['ci']
hsd = handle['hr']
isd = handle['ir']
except KeyError:
raise ConfigurationException("Invalid Configuration")
slave = ModbusSlaveContext(d=dsd, c=csd, h=hsd, i=isd)
return ModbusServerContext(slaves=slave)
def run_update_server():
# ----------------------------------------------------------------------- #
# initialize your data store
# ----------------------------------------------------------------------- #
store = ModbusSlaveContext(
di=ModbusSequentialDataBlock(0,range(1,101)),
co=ModbusSequentialDataBlock(0,range(101,201)),
hr=ModbusSequentialDataBlock(0,range(201,301)),
ir=ModbusSequentialDataBlock(0,range(301,401)))
context = ModbusServerContext(slaves=store, single=True)
# ----------------------------------------------------------------------- #
# initialize the server information
# ----------------------------------------------------------------------- #
identity = ModbusDeviceIdentification()
identity.VendorName = 'pymodbus'
identity.ProductCode = 'PM'
identity.VendorUrl = 'http://github.com/bashwork/pymodbus/'
identity.ProductName = 'pymodbus Server'
identity.ModelName = 'pymodbus Server'
def _parse(self):
""" Parses the config file and creates a server context """
try:
handle = pickle.load(self.file)
dsd = handle['di']
csd = handle['ci']
hsd = handle['hr']
isd = handle['ir']
except KeyError:
raise ConfigurationException("Invalid Configuration")
slave = ModbusSlaveContext(d=dsd, c=csd, h=hsd, i=isd)
return ModbusServerContext(slaves=slave)
eg.:
store = ModbusSlaveContext(
di = ModbusSequentialDataBlock(0, [0]*1000),
co = ModbusSequentialDataBlock(0, [0]*1000),
hr = ModbusSequentialDataBlock(0, [0]*10000),
ir = ModbusSequentialDataBlock(0, [0]*1000))
context = ModbusServerContext(slaves=store, single=True)
"""
definitions = register_definitions( registers )
did = definitions.get( 'di' )
cod = definitions.get( 'co' )
hrd = definitions.get( 'hr' )
ird = definitions.get( 'ir' )
store = ModbusSlaveContext(
di = modbus_sparse_data_block( did ) if did else None,
co = modbus_sparse_data_block( cod ) if cod else None,
hr = modbus_sparse_data_block( hrd ) if hrd else None,
ir = modbus_sparse_data_block( ird ) if ird else None )
# If slaves is None, then just pass the store with single=True; it will be
# used for every slave. Otherwise, map all the specified slave IDs to the
# same store and pass single=False. This would be used for Serial Modbus
# protocols, and you should probably also specify ignore_missing_slaves=True
# so that the simulator acts like a multi-drop serial PLC arrangement.
try:
if slaves is None:
return ModbusServerContext( slaves=store, single=True )
else:
if not hasattr( slaves, '__iter__' ):
slaves = [ slaves ] # Convert a single value to an iterable
def parse(self):
""" Parses the config file and creates a server context
"""
handle = pickle.load(self.file)
try: # test for existance, or bomb
dsd = handle['di']
csd = handle['ci']
hsd = handle['hr']
isd = handle['ir']
except Exception:
raise ConfigurationException("Invalid Configuration")
slave = ModbusSlaveContext(d=dsd, c=csd, h=hsd, i=isd)
return ModbusServerContext(slaves=slave)
def run_server():
store = ModbusSlaveContext(
di=ModbusSequentialDataBlock(0, [17] * 100),
co=ModbusSequentialDataBlock(0, [17] * 100),
hr=ModbusSequentialDataBlock(0, [17] * 100),
ir=ModbusSequentialDataBlock(0, [17] * 100))
store.register(CustomModbusRequest.function_code, 'cm',
ModbusSequentialDataBlock(0, [17] * 100))
context = ModbusServerContext(slaves=store, single=True)
# ----------------------------------------------------------------------- #
# initialize the server information
# ----------------------------------------------------------------------- #
# If you don't set this or any fields, they are defaulted to empty strings.
# ----------------------------------------------------------------------- #
identity = ModbusDeviceIdentification()
identity.VendorName = 'Pymodbus'
def run_callback_server():
# ----------------------------------------------------------------------- #
# initialize your data store
# ----------------------------------------------------------------------- #
queue = Queue()
devices = read_device_map("device-mapping")
block = CallbackDataBlock(devices, queue)
store = ModbusSlaveContext(di=block, co=block, hr=block, ir=block)
context = ModbusServerContext(slaves=store, single=True)
# ----------------------------------------------------------------------- #
# initialize the server information
# ----------------------------------------------------------------------- #
identity = ModbusDeviceIdentification()
identity.VendorName = 'pymodbus'
identity.ProductCode = 'PM'
identity.VendorUrl = 'http://github.com/bashwork/pymodbus/'
identity.ProductName = 'pymodbus Server'
identity.ModelName = 'pymodbus Server'
identity.MajorMinorRevision = '2.2.0'
# ----------------------------------------------------------------------- #
# run the server you want
# ----------------------------------------------------------------------- #
def __init__(self, address, blocks):
Thread.__init__(self)
store = ModbusSlaveContext(
di=blocks.get('di', ModbusSequentialDataBlock(0, [0])),
co=blocks.get('co', ModbusSequentialDataBlock(0, [0])),
hr=blocks.get('hr', ModbusSequentialDataBlock(0, [0])),
ir=blocks.get('ir', ModbusSequentialDataBlock(0, [0])))
self.context = ModbusServerContext(slaves=store, single=True)
self.server = ModbusTcpServer(context=self.context, framer=None, identity=None, address=address)