boseji · September 15, 2024 06:54
diff --git a/README.md b/README.md
diff --git a/clear-env.sh b/clear-env.sh
 #!/usr/bin/env bash
 # Copyright (C) 2024 Abhijit Bose (aka. Boseji) 
 # SPDX: MIT

 rm -rf venv
diff --git a/client.py b/client.py
 #!/usr/bin/env python
 # Copyright (C) 2024 Abhijit Bose (aka. Boseji) 
 # SPDX: MIT

 import socket

 from umodbus import conf
 from umodbus.client import tcp

 # Enable values to be signed (default is False).
 conf.SIGNED_VALUES = True
 # Server Detail
 svr = ('localhost', 50222)
 # Create the Socket
 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 print("\n Connecting to", svr)
 sock.connect(svr)

 # Returns a message or Application Data Unit (ADU) specific for doing
 # Modbus TCP/IP.
 message = tcp.write_single_register(
    slave_id=1, 
    address=1, 
    value=23,
    )
 # Return a Write Registers
 # message = tcp.write_multiple_registers(
 #     slave_id=1,
 #     starting_address=2,
 #     values=[14, 13, 67],
 # )

 # Response depends on Modbus function code. This particular returns the
 # amount of coils written, in this case it is.
 response = tcp.send_message(message, sock)

 sock.close()
diff --git a/Python_MODBUS-IP_Server_and_Client.md b/Python_MODBUS-IP_Server_and_Client.md
diff --git a/requirements.txt b/requirements.txt
 pyserial==3.5
 uModbus==1.0.4
diff --git a/server.py b/server.py
 #!/usr/bin/env python
 # Copyright (C) 2024 Abhijit Bose (aka. Boseji) 
 # SPDX: MIT
 import logging
 from socketserver import TCPServer
 from collections import defaultdict

 from umodbus import conf
 from umodbus.server.tcp import RequestHandler, get_server
 from umodbus.utils import log_to_stream

 # Add stream handler to logger 'uModbus'.
 log_to_stream(level=logging.DEBUG)
 # Server Detail
 svr = ('localhost', 50222)

 # A very simple data store which maps address against their values.
 data_store = defaultdict(int)

 # Enable values to be signed (default is False).
 conf.SIGNED_VALUES = True

 TCPServer.allow_reuse_address = True
 # Open Socket
 app = get_server(TCPServer, svr, RequestHandler)


 @app.route(slave_ids=[1], function_codes=[3, 4], addresses=list(range(0, 10)))
 def read_data_store(slave_id, function_code, address):
    """" Return value of address. """
    print("Fetch ADDR:", address)
    return data_store[address]


 @app.route(slave_ids=[1], function_codes=[6, 16], addresses=list(range(0, 10)))
 def write_data_store(slave_id, function_code, address, value):
    """" Set value for address. """
    print("Store ADDR:", address, " Data:", value)
    data_store[address] = value

 if __name__ == '__main__':
    print("\n Starting Sever ", svr)
    try:
        app.serve_forever()
    finally:
        print("\n\n Closing down server...\n")
        app.shutdown()
        app.server_close()
diff --git a/setup-env.sh b/setup-env.sh
 #!/usr/bin/env bash
 # Copyright (C) 2024 Abhijit Bose (aka. Boseji) 
 # SPDX: MIT

 python3 -m venv venv
 source venv/bin/activate
 pip install -r requirements.txt
	#!/usr/bin/env bash
	# Copyright (C) 2024 Abhijit Bose (aka. Boseji)
	# SPDX: MIT

	rm -rf venv
	#!/usr/bin/env python
	# Copyright (C) 2024 Abhijit Bose (aka. Boseji)
	# SPDX: MIT

	import socket

	from umodbus import conf
	from umodbus.client import tcp

	# Enable values to be signed (default is False).
	conf.SIGNED_VALUES = True
	# Server Detail
	svr = ('localhost', 50222)
	# Create the Socket
	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	print("\n Connecting to", svr)
	sock.connect(svr)

	# Returns a message or Application Data Unit (ADU) specific for doing
	# Modbus TCP/IP.
	message = tcp.write_single_register(
	slave_id=1,
	address=1,
	value=23,
	)
	# Return a Write Registers
	# message = tcp.write_multiple_registers(
	# slave_id=1,
	# starting_address=2,
	# values=[14, 13, 67],
	# )

	# Response depends on Modbus function code. This particular returns the
	# amount of coils written, in this case it is.
	response = tcp.send_message(message, sock)

	sock.close()