Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#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'
# ----------------------------------------------------------------------- #
# run the server you want
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)
# context = ModbusServerContext(slaves=slaves, single=False)
#
# The slave context can also be initialized in zero_mode which means that a
# request to address(0-7) will map to the address (0-7). The default is
# False which is based on section 4.4 of the specification, so address(0-7)
# will map to (1-8)::
#
# store = ModbusSlaveContext(..., zero_mode=True)
# ----------------------------------------------------------------------- #
store = ModbusSlaveContext(
di=ModbusSequentialDataBlock(0, [17]*100),
co=ModbusSequentialDataBlock(0, [17]*100),
hr=ModbusSequentialDataBlock(0, [17]*100),
ir=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'
identity.ProductCode = 'PM'
identity.VendorUrl = 'http://github.com/riptideio/pymodbus/'
identity.ProductName = 'Pymodbus Server'
identity.ModelName = 'Pymodbus Server'
identity.MajorMinorRevision = '2.2.0'
# ----------------------------------------------------------------------- #
# run the server you want
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_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'
identity.MajorMinorRevision = '1.0'
# connect to simulation
HOST = '127.0.0.1'
PORT = 55555
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
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
return ModbusServerContext( slaves=dict( (uid,store) for uid in slaves ), single=False )
finally:
log.info( "Modbus Slave IDs: %s", slaves or "(all)" )
#
# The slave context can also be initialized in zero_mode which means that a
# request to address(0-7) will map to the address (0-7). The default is
# False which is based on section 4.4 of the specification, so address(0-7)
# will map to (1-8)::
#
# store = ModbusSlaveContext(..., zero_mode=True)
# ----------------------------------------------------------------------- #
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'
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 run_custom_db_server():
# ----------------------------------------------------------------------- #
# initialize your data store
# ----------------------------------------------------------------------- #
block = CustomDataBlock([0]*100)
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 run_updating_server():
# ----------------------------------------------------------------------- #
# initialize your data store
# ----------------------------------------------------------------------- #
store = ModbusSlaveContext(
di=ModbusSequentialDataBlock(0, [17]*100),
co=ModbusSequentialDataBlock(0, [17]*100),
hr=ModbusSequentialDataBlock(0, [17]*100),
ir=ModbusSequentialDataBlock(0, [17]*100))
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
# ----------------------------------------------------------------------- #
time = 5 # 5 seconds delay
# for each device (unit), create an EmulatorDataBlock that has the register map defined in config
def _convert_register(v):
if isinstance(v, list):
# return a list of converted values
return [int(x, 16) for x in v]
else:
# convert single value to int
return int(v, 16)
block = EmulatorDataBlock({int(k, 16): _convert_register(v) for k, v in device['holding_registers'].items()})
logger.error('Device: {} Block: {}'.format(device['device_address'], block.values))
store = ModbusSlaveContext(di=block, co=block, hr=block, ir=block)
slaves[device['device_address']] = store
context = ModbusServerContext(slaves=slaves, single=False)
for slave in context:
logger.error('Slaves: {}'.format(slave[0]))
identity = ModbusDeviceIdentification()
identity.VendorName = 'Vapor IO'
identity.ProductName = 'Synse Modbus Emulator'
identity.MajorMinorRevision = '1.4'
# this is done to allow us to actually use RTU framing which the Start shortcut prevents
server = ModbusSerialServer(context, framer=ModbusRtuFramer, identity=identity,
port=emulator_config['serial_device'], timeout=0.01)
server.serve_forever()