# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
# !/usr/bin/env python
import greengrasssdk
import json
import logging
import os
import sys
import time
from pymodbus.client.sync import ModbusSerialClient as ModbusClient
import connector_lambda_python_commons.response as PublishResponse
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder
# NOTE: Must specify the serial port the GGC is connected to.
DEFAULT_SERIAL_PORT = '/dev/ttyUSB0'
METHOD = 'rtu'
DEFAULT_TIMEOUT_SECONDS = 1
DEFAULT_TOPIC_RESPONSE = 'modbus/adapter/response'
DEFAULT_BAUDRATE = 115200
UTF_8 = "utf-8"
#extra added ones
PARITY = 'N'
BYTESIZE = 8
STOPBITS = 1
FIELD_FUNCTION_CODE = "function_code"
FIELD_EXCEPTION_CODE = "exception_code"
FIELD_RESPONSE = "response"
FIELD_PAYLOAD = "payload"
FIELD_DEVICE = "device"
FIELD_OPERATION = "operation"
FIELD_ERROR = "error"
FIELD_VALUE = "value"
FIELD_VALUES = "values"
FIELD_ADDRESS = "address"
FIELD_COUNT = "count"
FIELD_READ_ADDRESS = "read_address"
FIELD_READ_COUNT = "read_count"
FIELD_WRITE_ADDRESS = "write_address"
FIELD_WRITE_REGISTERS = "write_registers"
FIELD_AND_MASK = "and_mask"
FIELD_OR_MASK = "or_mask"
FIELD_BITS = "bits"
FIELD_REGISTERS = "registers"
FIELD_REQUEST_ID = "id"
FIELD_REQUEST = "request"
FIELD_NUMBER_OF_PHASES = "number_of_phases"
FIELD_CHANNEL_FOR_FIRST_CT = "channel_for_first_CT"
FIELD_CHANNEL_FOR_SECOND_CT = "channel_for_second_CT"
FIELD_CHANNEL_FOR_THIRD_CT = "channel_for_third_CT"
FIELD_VIRTUAL_CHANNEL = "virtual_channel"
NAME_CURRENT_A = "current_a"
NAME_CURRENT_B = "current_b"
NAME_CURRENT_C = "current_c"
VALUE_REQUEST_ID_DEFAULT = "Missing"
VALUE_OPERATION_DEFAULT = "Missing"
VALUE_DEVICE_DEFAULT = -1
STATUS_NO_RESPONSE = "No Response"
STATUS_EXCEPTION = "Exception"
STATUS_SUCCESS = "Success"
SERIAL_PORT_ENV_VAR = "ModbusLocalPort"
NAMES_SINGLE_PHASE = ['power_total','power_reac_sum','power_app_sum','energy_total_pos','energy_total_neg','energy_reac_sum_import','energy_reac_sum_export']
NAMES_MULTIPHASE_A = ['power_a','power_reac_a','power_app_a','energy_a_pos','energy_a_neg','energy_a_sum_import','energy_reac_a_export']
NAMES_MULTIPHASE_B = ['power_b','power_reac_b','power_app_b','energy_b_pos','energy_b_neg','energy_b_sum_import','energy_reac_b_export']
NAMES_MULTIPHASE_C = ['power_a','power_reac_a','power_app_c','energy_c_pos','energy_c_neg','energy_c_sum_import','energy_reac_c_export']
# Convert string to unicode. In Python 3.x, string is by default unicode.
def convert_unicode(string):
if sys.version_info[0] < 3:
return unicode(string, UTF_8)
else:
return string
OPERATION_READ_COILS = convert_unicode("ReadCoilsRequest")
OPERATION_READ_INPUT_REGISTERS = convert_unicode("ReadInputRegistersRequest")
OPERATION_READ_DISCRETE_INPUTS = convert_unicode("ReadDiscreteInputsRequest")
OPERATION_READ_HOLDING_REGISTERS = convert_unicode("ReadHoldingRegistersRequest")
OPERATION_READ_ALL_HOLDING_REGISTERS = convert_unicode("ReadAllHoldingRegistersRequest")
OPERATION_WRITE_SINGLE_COIL = convert_unicode("WriteSingleCoilRequest")
OPERATION_WRITE_SINGLE_REGISTER = convert_unicode("WriteSingleRegisterRequest")
OPERATION_WRITE_MULTIPLE_COILS = convert_unicode("WriteMultipleCoilsRequest")
OPERATION_WRITE_MULTIPLE_REGISTERS = convert_unicode("WriteMultipleRegistersRequest")
OPERATION_READ_WRITE_MULTIPLE_REGISTERS = convert_unicode("ReadWriteMultipleRegistersRequest")
OPERATION_MASK_WRITE_REGISTER = convert_unicode("MaskWriteRegisterRequest")
# Run for all read requests
# Read requests require an address and count
def read_request(client, request, op):
modbus_response = {}
# Ensure address and count exist, and are both of type int
error = check_fields(request, [FIELD_ADDRESS, FIELD_COUNT], [int, int])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
address = request[FIELD_ADDRESS]
count = request[FIELD_COUNT]
# The device field has been checked earlier
unit = request[FIELD_DEVICE]
# Count has to be greater than 1.
if count < 1:
modbus_response[FIELD_ERROR] = "Invalid count {}".format(count)
return modbus_response
# The current supported read requests.
if op == OPERATION_READ_COILS:
modbus_response = client.read_coils(address, count, unit=unit)
elif op == OPERATION_READ_DISCRETE_INPUTS:
modbus_response = client.read_discrete_inputs(address, count, unit=unit)
elif op == OPERATION_READ_HOLDING_REGISTERS:
modbus_response = client.read_holding_registers(address, count, unit=unit)
elif op == OPERATION_READ_INPUT_REGISTERS:
modbus_response = client.read_input_registers(address, count, unit=unit)
else:
# Unexpected request
modbus_response[FIELD_ERROR] = "Unexpected request {}".format(op)
# Add address to response.
setattr(modbus_response, FIELD_ADDRESS, address)
return modbus_response
# Run for read all holding registers request
# Read all holding registers requests require number_of_phases
def read_all_request(client, request, op):
modbus_response = {}
# Ensure number_of_phases exists, and is of type int
error = check_fields(request, [FIELD_NUMBER_OF_PHASES], [int])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
address = request[FIELD_ADDRESS]
numberOfPhases = request[FIELD_NUMBER_OF_PHASES]
# The device field has been checked earlier
unit = request[FIELD_DEVICE]
virtualChannel = request[FIELD_VIRTUAL_CHANNEL]
# numberOfPhases has to be greater than 0.
if numberOfPhases < 1:
modbus_response[FIELD_ERROR] = "Invalid numberOfPhases {}".format(numberOfPhases)
return modbus_response
# numberOfPhases has to be greater than 0.
if numberOfPhases == 1:
offsetPhaseA = request[FIELD_CHANNEL_FOR_FIRST_CT]
getCtInfo(client, unit, offsetPhaseA, modbus_response, NAMES_SINGLE_PHASE)
getCurrentValues(client, unit, offsetPhaseA, modbus_response, NAME_CURRENT_A)
elif numberOfPhases == 2:
offsetPhaseA = request[FIELD_CHANNEL_FOR_FIRST_CT]
offsetPhaseB = request[FIELD_CHANNEL_FOR_SECOND_CT]
getVcInfo(client, unit, virtualChannel, modbus_response)
getCtInfo(client, unit, offsetPhaseA, modbus_response, NAMES_MULTIPHASE_A)
getCtInfo(client, unit, offsetPhaseB, modbus_response, NAMES_MULTIPHASE_B)
getCurrentValues(client, unit, offsetPhaseA, modbus_response, NAME_CURRENT_A)
getCurrentValues(client, unit, offsetPhaseB, modbus_response, NAME_CURRENT_B)
elif numberOfPhases == 3:
offsetPhaseA = request[FIELD_CHANNEL_FOR_FIRST_CT]
offsetPhaseB = request[FIELD_CHANNEL_FOR_SECOND_CT]
offsetPhaseC = request[FIELD_CHANNEL_FOR_THIRD_CT]
getVcInfo(client, unit, virtualChannel, modbus_response)
getCtInfo(client, unit, offsetPhaseA, modbus_response, NAMES_MULTIPHASE_A)
getCtInfo(client, unit, offsetPhaseB, modbus_response, NAMES_MULTIPHASE_B)
getCtInfo(client, unit, offsetPhaseC, modbus_response, NAMES_MULTIPHASE_C)
getCurrentValues(client, unit, offsetPhaseA, modbus_response, NAME_CURRENT_A)
getCurrentValues(client, unit, offsetPhaseB, modbus_response, NAME_CURRENT_B)
getCurrentValues(client, unit, offsetPhaseC, modbus_response, NAME_CURRENT_C)
else:
# Unexpected request
modbus_response[FIELD_ERROR] = "Invalid numberOfPhases {}".format(numberOfPhases)
getVoltageValues(client, unit, modbus_response)
getFrequencyValue(client, unit, modbus_response)
return modbus_response
# Run for all write single requests
# Single write requests require an address and value
def write_single_request(client, request, op):
modbus_response = {}
error = check_fields(request, [FIELD_ADDRESS, FIELD_VALUE], [int, int])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
address = request[FIELD_ADDRESS]
value = request[FIELD_VALUE]
unit = request[FIELD_DEVICE]
# The current supported write single requests
if op == OPERATION_WRITE_SINGLE_COIL:
modbus_response = client.write_coil(address, value, unit=unit)
elif op == OPERATION_WRITE_SINGLE_REGISTER:
modbus_response = client.write_register(address, value, unit=unit)
else:
# Unexpected request
modbus_response[FIELD_ERROR] = "Unexpected request {}".format(op)
return modbus_response
# Run for all write multiple requests
# Multiple write requests require an address and values
def write_multiple_request(client, request, op):
modbus_response = {}
error = check_fields(request, [FIELD_ADDRESS, FIELD_VALUES], [int, list])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
values = request[FIELD_VALUES]
address = request[FIELD_ADDRESS]
unit = request[FIELD_DEVICE]
if not all(isinstance(x, int) for x in values):
modbus_response[FIELD_ERROR] = "Values must be a non-empty list of integers"
return modbus_response
# The current supported write multiple requests
if op == OPERATION_WRITE_MULTIPLE_COILS:
modbus_response = client.write_coils(address, values, unit=unit)
elif op == OPERATION_WRITE_MULTIPLE_REGISTERS:
modbus_response = client.write_registers(address, values, unit=unit)
else:
# Unexpected request
modbus_response[FIELD_ERROR] = "Unexpected request {}".format(op)
return modbus_response
# Run for readwrite multiple registers
# ReadWrite registers requests require a read address and count, and a write address and registers
def readwrite_multiple_registers_request(client, request, op):
modbus_response = {}
error = check_fields(request, [FIELD_READ_ADDRESS, FIELD_READ_COUNT, FIELD_WRITE_ADDRESS, FIELD_WRITE_REGISTERS],
[int, int, int, list])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
if not all(isinstance(x, int) for x in request[FIELD_WRITE_REGISTERS]):
modbus_response[FIELD_ERROR] = "{} must be a non-empty list of integers".format('write_registers')
return modbus_response
# This is the way pymodbus takes in these arguments.
return client.readwrite_registers(**request)
# Run for maskwrite registers
# Maskwrite registers requests require an address, andmask and an ormask
def mask_write_register_request(client, request, op):
modbus_response = {}
error = check_fields(request, [FIELD_ADDRESS, FIELD_AND_MASK, FIELD_OR_MASK], [int, int, int])
if error:
modbus_response[FIELD_ERROR] = error
return modbus_response
return client.mask_write_register(**request)
# Executes the request and returns a json response
def execute(client, request):
json_response = init_default_json_response(request)
# First check if the request has the correct basic fields
error = validate_request(request[FIELD_REQUEST])
if error:
json_response[FIELD_RESPONSE][FIELD_PAYLOAD][FIELD_ERROR] = error
else:
# At this point, 'json_response' should always have a valid Operation and Device field
operation = json_response[FIELD_RESPONSE][FIELD_OPERATION]
json_response[FIELD_RESPONSE][FIELD_PAYLOAD] = function_dict[operation](client, request[FIELD_REQUEST],
operation)
serialize_response(json_response, response_fields_dict[operation])
return update_response_status(json_response)
# All Responses will contain an operation and device default
def init_default_json_response(request):
json_response = {
FIELD_RESPONSE: {
FIELD_PAYLOAD: {},
FIELD_OPERATION: VALUE_OPERATION_DEFAULT,
FIELD_DEVICE: VALUE_DEVICE_DEFAULT
},
FIELD_REQUEST_ID: VALUE_REQUEST_ID_DEFAULT
}
if type(request) is dict and request.get(FIELD_REQUEST):
json_response[FIELD_REQUEST_ID] = request.get(FIELD_REQUEST_ID, VALUE_REQUEST_ID_DEFAULT)
json_response[FIELD_RESPONSE][FIELD_OPERATION] = request[FIELD_REQUEST].get(FIELD_OPERATION,
VALUE_OPERATION_DEFAULT)
json_response[FIELD_RESPONSE][FIELD_DEVICE] = request[FIELD_REQUEST].get(FIELD_DEVICE, VALUE_DEVICE_DEFAULT)
return json_response
# Return meaningful error message for invalid/missing fields, or None if the request is valid.
def validate_request(request):
fields = [FIELD_DEVICE, FIELD_OPERATION]
if sys.version_info[0] < 3:
types = [int, unicode]
else:
types = [int, str]
error = check_fields(request, fields, types)
# At this point we know that the device and operation are present and valid format
if not error:
# The Device address is expected to be an integer from 1-247
# The Operation is expected to be supported and found in function_dict
if request[FIELD_DEVICE] < 1 or request[FIELD_DEVICE] > 247:
error = "Device address {} is out of range (0-247)".format(request[FIELD_DEVICE])
elif request[FIELD_OPERATION] not in function_dict:
error = "Operation {} is not supported".format(request[FIELD_OPERATION])
return error
# Returns whether the specified fields are present in the request and have the expected types.
def check_fields(request, fields, expected_types):
# Assumes that fields and expected_types are the same length when check_fields is called
for field, expected_type in zip(fields, expected_types):
if field not in request:
return "Missing {} field".format(field)
elif not isinstance(request[field], expected_type):
return "Invalid {} field. Expected {}, got {}".format(field, expected_type, type(request[field]))
elif expected_type == list and len(request[field]) < 1:
return "Invalid {} field. List must be non-empty".format(field)
return None
# Turns the response into a serializable JSON.
def serialize_response(json_response, fields):
serial_response = {}
response = json_response[FIELD_RESPONSE][FIELD_PAYLOAD]
if hasattr(response, FIELD_FUNCTION_CODE):
# If the response is an exception, we want to pass the function code and exception code.
if response.function_code > 0x80:
serial_response[FIELD_FUNCTION_CODE] = response.function_code
serial_response[FIELD_EXCEPTION_CODE] = response.exception_code
else:
for field in fields:
serial_response[field] = getattr(response, field)
else:
# This means that the response was a ModbusException error. All ModbusException objects have a 'string' field.
serial_response[FIELD_ERROR] = getattr(response, 'string')
json_response[FIELD_RESPONSE][FIELD_PAYLOAD] = serial_response
return json_response
# Adds corresponding status message to the JSONResponse
def update_response_status(json_response):
response = json_response[FIELD_RESPONSE][FIELD_PAYLOAD]
# The best way to determine if the response was a success is by checking the function_code field.
# Function_code fields above 0x80 indicate an exception.
if FIELD_FUNCTION_CODE not in response:
# No response was received, the error was caught in translator.py, and no request was sent
publishResponse = PublishResponse.generate_error_response(json_response[FIELD_REQUEST_ID], STATUS_NO_RESPONSE,
json_response[FIELD_RESPONSE][FIELD_PAYLOAD][
FIELD_ERROR],
**json_response[FIELD_RESPONSE])
elif response[FIELD_FUNCTION_CODE] > 0x80:
publishResponse = PublishResponse.generate_error_response(json_response[FIELD_REQUEST_ID], STATUS_EXCEPTION,
"Internal Error",
**json_response[FIELD_RESPONSE])
else:
publishResponse = PublishResponse.generate_success_response(json_response[FIELD_REQUEST_ID],
**json_response[FIELD_RESPONSE])
return publishResponse
# Modbus Response Fields
response_fields_dict = {
OPERATION_READ_COILS: [FIELD_FUNCTION_CODE, FIELD_BITS, FIELD_ADDRESS],
OPERATION_READ_INPUT_REGISTERS: [FIELD_FUNCTION_CODE, FIELD_REGISTERS, FIELD_ADDRESS],
OPERATION_READ_DISCRETE_INPUTS: [FIELD_FUNCTION_CODE, FIELD_REGISTERS, FIELD_ADDRESS],
OPERATION_READ_HOLDING_REGISTERS: [FIELD_FUNCTION_CODE, FIELD_REGISTERS, FIELD_ADDRESS],
OPERATION_WRITE_SINGLE_COIL: [FIELD_FUNCTION_CODE, FIELD_ADDRESS, FIELD_VALUE],
OPERATION_WRITE_SINGLE_REGISTER: [FIELD_FUNCTION_CODE, FIELD_ADDRESS, FIELD_VALUE],
OPERATION_WRITE_MULTIPLE_COILS: [FIELD_FUNCTION_CODE, FIELD_ADDRESS, FIELD_COUNT],
OPERATION_WRITE_MULTIPLE_REGISTERS: [FIELD_FUNCTION_CODE, FIELD_COUNT, FIELD_ADDRESS],
OPERATION_READ_WRITE_MULTIPLE_REGISTERS: [FIELD_FUNCTION_CODE, FIELD_REGISTERS, FIELD_ADDRESS],
OPERATION_MASK_WRITE_REGISTER: [FIELD_FUNCTION_CODE, FIELD_AND_MASK, FIELD_OR_MASK]
}
# Modbus Function Mapping
function_dict = {
OPERATION_READ_COILS: read_request,
OPERATION_READ_INPUT_REGISTERS: read_request,
OPERATION_READ_DISCRETE_INPUTS: read_request,
OPERATION_READ_HOLDING_REGISTERS: read_request,
OPERATION_READ_ALL_HOLDING_REGISTERS: read_all_request,
OPERATION_WRITE_SINGLE_COIL: write_single_request,
OPERATION_WRITE_SINGLE_REGISTER: write_single_request,
OPERATION_WRITE_MULTIPLE_COILS: write_multiple_request,
OPERATION_WRITE_MULTIPLE_REGISTERS: write_multiple_request,
OPERATION_READ_WRITE_MULTIPLE_REGISTERS: readwrite_multiple_registers_request,
OPERATION_MASK_WRITE_REGISTER: mask_write_register_request
}
log = logging.getLogger()
log.setLevel(logging.DEBUG)
# Creating a greengrass core sdk client
client = greengrasssdk.client('iot-data')
# Initializing the Modbus Client
if SERIAL_PORT_ENV_VAR in os.environ:
port = os.environ[SERIAL_PORT_ENV_VAR]
log.info("port: {}".format(port))
else:
port = DEFAULT_SERIAL_PORT
log.info("default port")
modbusClient = ModbusClient(method=METHOD, port=port, timeout=DEFAULT_TIMEOUT_SECONDS, baudrate=DEFAULT_BAUDRATE,
stopbits=STOPBITS, bytesize=BYTESIZE, parity=PARITY)
def getVoltageValues(client, deviceid, response):
registers = [0, 4, 8, 12, 16, 20]
length = 2
names = ["voltage_a","voltage_b","voltage_c","voltage_a_b","voltage_c_b", "voltage_a_c"]
for index, reg in enumerate(registers):
result = client.read_holding_registers(reg, length , unit=deviceid)
# if not result.isError():
decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Little)
value = decoder.decode_32bit_float()
response[names[index]] = value
def getCurrentValues(client, deviceid, offset, response, name):
reg = 128
length = 2
skip = 2
regToRead = reg + offset * (length + skip)
result = client.read_holding_registers(regToRead, length , unit=deviceid)
decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Little)
currentValue = decoder.decode_32bit_float()
response[name] = currentValue
def getFrequencyValue(client, deviceid, response):
register = 1016
length = 2
name = "frequency"
result = client.read_holding_registers(register, length , unit=deviceid)
decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Little)
value = decoder.decode_32bit_float()
response[name] = value
# This is used for multiphase type only
def getVcInfo(client, deviceid, offset, response):
global previousValue
registers = [512, 514, 640, 642, 12544, 12546, 12548, 12550]
lengths = [2,2,2,2,2,2,2,2]
skip = [2,2,2,2,2,2,2,2]
names = ["power_total","power_total_fundamental","power_reac_sum","power_reac_sum_fundamental","energy_total_pos","energy_total_neg","energy_reac_sum_import","energy_reac_sum_export"]
types = ['FLOAT', 'FLOAT', 'FLOAT', 'FLOAT', 'INT_32', 'INT_32', 'INT_32', 'INT_32']
for index, reg in enumerate(registers):
regToRead = reg + offset * (lengths[index] + skip[index])
result = client.read_holding_registers(regToRead, lengths[index], unit=deviceid)
decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Little)
if types[index] == 'FLOAT':
currentValue = decoder.decode_32bit_float()
elif types[index] == 'INT_16':
currentValue = decoder.decode_16bit_uint()
elif types[index] == 'INT_32':
currentValue = decoder.decode_32bit_uint()
response[names[index]] = currentValue
# call for both single phase and multiphase
def getCtInfo(client, deviceid, offset, response, names):
registers = [256, 384, 768, 12288, 12290, 12416, 12418]
lengths = [2, 2, 2, 2, 2, 2, 2]
skip = [2, 2, 2, 2, 2, 4, 4]
types = ['FLOAT', 'FLOAT', 'FLOAT', 'INT_32', 'INT_32', 'INT_32', 'INT_32']
for index, reg in enumerate(registers):
regToRead = reg + offset * (lengths[index] + skip[index])
result = client.read_holding_registers(regToRead, lengths[index], unit=deviceid)
decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big, wordorder=Endian.Little)
if types[index] == 'FLOAT':
currentValue = decoder.decode_32bit_float()
elif types[index] == 'INT_16':
currentValue = decoder.decode_16bit_uint()
elif types[index] == 'INT_32':
currentValue = decoder.decode_32bit_uint()
response[names[index]] = currentValue
def function_handler(event, context):
log.info("Client connecting...")
response = {}
try:
modbusClient.connect()
log.info("Connected")
response = execute(modbusClient, event)
log.info(response)
log.info("Closing connection")
modbusClient.close()
time.sleep(1)
except Exception as e:
json_response = init_default_json_response(event)
json_response[FIELD_RESPONSE][FIELD_PAYLOAD][FIELD_ERROR] = str(e)
response = PublishResponse.generate_error_response(json_response[FIELD_REQUEST_ID], STATUS_EXCEPTION,
str(e),
**json_response[FIELD_RESPONSE])
client.publish(topic=DEFAULT_TOPIC_RESPONSE, payload=json.dumps(response))
return
Write, Run & Share Python code online using OneCompiler's Python online compiler for free. It's one of the robust, feature-rich online compilers for python language, supporting both the versions which are Python 3 and Python 2.7. Getting started with the OneCompiler's Python editor is easy and fast. The editor shows sample boilerplate code when you choose language as Python or Python2 and start coding.
OneCompiler's python online editor supports stdin and users can give inputs to programs using the STDIN textbox under the I/O tab. Following is a sample python program which takes name as input and print your name with hello.
import sys
name = sys.stdin.readline()
print("Hello "+ name)
Python is a very popular general-purpose programming language which was created by Guido van Rossum, and released in 1991. It is very popular for web development and you can build almost anything like mobile apps, web apps, tools, data analytics, machine learning etc. It is designed to be simple and easy like english language. It's is highly productive and efficient making it a very popular language.
When ever you want to perform a set of operations based on a condition IF-ELSE is used.
if conditional-expression
#code
elif conditional-expression
#code
else:
#code
Indentation is very important in Python, make sure the indentation is followed correctly
For loop is used to iterate over arrays(list, tuple, set, dictionary) or strings.
mylist=("Iphone","Pixel","Samsung")
for i in mylist:
print(i)
While is also used to iterate a set of statements based on a condition. Usually while is preferred when number of iterations are not known in advance.
while condition
#code
There are four types of collections in Python.
List is a collection which is ordered and can be changed. Lists are specified in square brackets.
mylist=["iPhone","Pixel","Samsung"]
print(mylist)
Tuple is a collection which is ordered and can not be changed. Tuples are specified in round brackets.
myTuple=("iPhone","Pixel","Samsung")
print(myTuple)
Below throws an error if you assign another value to tuple again.
myTuple=("iPhone","Pixel","Samsung")
print(myTuple)
myTuple[1]="onePlus"
print(myTuple)
Set is a collection which is unordered and unindexed. Sets are specified in curly brackets.
myset = {"iPhone","Pixel","Samsung"}
print(myset)
Dictionary is a collection of key value pairs which is unordered, can be changed, and indexed. They are written in curly brackets with key - value pairs.
mydict = {
"brand" :"iPhone",
"model": "iPhone 11"
}
print(mydict)
Following are the libraries supported by OneCompiler's Python compiler
| Name | Description |
|---|---|
| NumPy | NumPy python library helps users to work on arrays with ease |
| SciPy | SciPy is a scientific computation library which depends on NumPy for convenient and fast N-dimensional array manipulation |
| SKLearn/Scikit-learn | Scikit-learn or Scikit-learn is the most useful library for machine learning in Python |
| Pandas | Pandas is the most efficient Python library for data manipulation and analysis |
| DOcplex | DOcplex is IBM Decision Optimization CPLEX Modeling for Python, is a library composed of Mathematical Programming Modeling and Constraint Programming Modeling |