def handle_request(self, request):<EOL>
<EOL>command_id_bytes = request.ams_header.command_id<EOL>command_id = struct.unpack("<STR_LIT>", command_id_bytes)[<NUM_LIT:0>]<EOL>state = struct.unpack("<STR_LIT>", request.ams_header.state_flags)[<NUM_LIT:0>]<EOL>state = state | <NUM_LIT>  <EOL>state = struct.pack("<STR_LIT>", state)<EOL>def handle_read_device_info():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>major_version = "<STR_LIT>".encode("<STR_LIT:utf-8>")<EOL>minor_version = "<STR_LIT>".encode("<STR_LIT:utf-8>")<EOL>version_build = "<STR_LIT>".encode("<STR_LIT:utf-8>")<EOL>device_name = "<STR_LIT>".encode("<STR_LIT:utf-8>")<EOL>response_content = (<EOL>major_version + minor_version + version_build + device_name<EOL>)<EOL>return response_content<EOL><DEDENT>def handle_read():<EOL><INDENT>"""<STR_LIT>"""<EOL>data = request.ams_header.data<EOL>index_group = struct.unpack("<STR_LIT>", data[:<NUM_LIT:4>])[<NUM_LIT:0>]<EOL>index_offset = struct.unpack("<STR_LIT>", data[<NUM_LIT:4>:<NUM_LIT:8>])[<NUM_LIT:0>]<EOL>plc_datatype = struct.unpack("<STR_LIT>", data[<NUM_LIT:8>:<NUM_LIT:12>])[<NUM_LIT:0>]<EOL>logger.info(<EOL>(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>).format(index_group, index_offset, plc_datatype)<EOL>)<EOL>if index_group == constants.ADSIGRP_SYM_VALBYHND:<EOL><INDENT>response_value = self._named_data[index_offset].value<EOL><DEDENT>else:<EOL><INDENT>response_value = self._data[(index_group, index_offset)][:plc_datatype]<EOL><DEDENT>return struct.pack("<STR_LIT>", len(response_value)) + response_value<EOL><DEDENT>def handle_write():<EOL><INDENT>"""<STR_LIT>"""<EOL>data = request.ams_header.data<EOL>index_group = struct.unpack("<STR_LIT>", data[:<NUM_LIT:4>])[<NUM_LIT:0>]<EOL>index_offset = struct.unpack("<STR_LIT>", data[<NUM_LIT:4>:<NUM_LIT:8>])[<NUM_LIT:0>]<EOL>plc_datatype = struct.unpack("<STR_LIT>", data[<NUM_LIT:8>:<NUM_LIT:12>])[<NUM_LIT:0>]<EOL>value = data[<NUM_LIT:12>:(<NUM_LIT:12> + plc_datatype)]<EOL>logger.info(<EOL>(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>).format(index_group, index_offset, plc_datatype, value)<EOL>)<EOL>if index_group == constants.ADSIGRP_SYM_RELEASEHND:<EOL><INDENT>return b"<STR_LIT>"<EOL><DEDENT>elif index_group == constants.ADSIGRP_SYM_VALBYHND:<EOL><INDENT>self._named_data[index_offset].value = value<EOL>return b"<STR_LIT>"<EOL><DEDENT>self._data[(index_group, index_offset)] = value<EOL>return b"<STR_LIT>"<EOL><DEDENT>def handle_read_write():<EOL><INDENT>"""<STR_LIT>"""<EOL>data = request.ams_header.data<EOL>index_group = struct.unpack("<STR_LIT>", data[:<NUM_LIT:4>])[<NUM_LIT:0>]<EOL>index_offset = struct.unpack("<STR_LIT>", data[<NUM_LIT:4>:<NUM_LIT:8>])[<NUM_LIT:0>]<EOL>read_length = struct.unpack("<STR_LIT>", data[<NUM_LIT:8>:<NUM_LIT:12>])[<NUM_LIT:0>]<EOL>write_length = struct.unpack("<STR_LIT>", data[<NUM_LIT:12>:<NUM_LIT:16>])[<NUM_LIT:0>]<EOL>write_data = data[<NUM_LIT:16>:(<NUM_LIT:16> + write_length)]<EOL>logger.info(<EOL>(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>).format(<EOL>index_group, index_offset, read_length, write_length, write_data<EOL>)<EOL>)<EOL>if index_group == constants.ADSIGRP_SYM_HNDBYNAME:<EOL><INDENT>var_name = write_data.decode()<EOL>names = [x.name for x in self._named_data]<EOL>try:<EOL><INDENT>handle = names.index(var_name)<EOL><DEDENT>except ValueError:<EOL><INDENT>self._named_data.append(PLCVariable(name=var_name, value=bytes(<NUM_LIT:16>)))<EOL>handle = len(self._named_data) - <NUM_LIT:1><EOL><DEDENT>read_data = struct.pack("<STR_LIT>", handle)<EOL><DEDENT>else:<EOL><INDENT>read_data = self._data[(index_group, index_offset)][:read_length]<EOL>self._data[(index_group, index_offset)] = write_data<EOL><DEDENT>return struct.pack("<STR_LIT>", len(read_data)) + read_data<EOL><DEDENT>def handle_read_state():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>ads_state = struct.pack("<STR_LIT>", constants.ADSSTATE_RUN)<EOL>device_state = struct.pack("<STR_LIT>", <NUM_LIT:0>)<EOL>return ads_state + device_state<EOL><DEDENT>def handle_writectrl():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>return b"<STR_LIT>"<EOL><DEDENT>def handle_add_devicenote():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>handle = ("<STR_LIT>" * <NUM_LIT:4>).encode("<STR_LIT:utf-8>")<EOL>return handle<EOL><DEDENT>def handle_delete_devicenote():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>return b"<STR_LIT>"<EOL><DEDENT>def handle_devicenote():<EOL><INDENT>"""<STR_LIT>"""<EOL>logger.info("<STR_LIT>")<EOL>return b"<STR_LIT>"<EOL><DEDENT>function_map = {<EOL>constants.ADSCOMMAND_READDEVICEINFO: handle_read_device_info,<EOL>constants.ADSCOMMAND_READ: handle_read,<EOL>constants.ADSCOMMAND_WRITE: handle_write,<EOL>constants.ADSCOMMAND_READWRITE: handle_read_write,<EOL>constants.ADSCOMMAND_READSTATE: handle_read_state,<EOL>constants.ADSCOMMAND_WRITECTRL: handle_writectrl,<EOL>constants.ADSCOMMAND_ADDDEVICENOTE: handle_add_devicenote,<EOL>constants.ADSCOMMAND_DELDEVICENOTE: handle_delete_devicenote,<EOL>constants.ADSCOMMAND_DEVICENOTE: handle_devicenote,<EOL>}<EOL>try:<EOL><INDENT>response_content = function_map[command_id]()<EOL><DEDENT>except KeyError:<EOL><INDENT>logger.info("<STR_LIT>".format(hex(command_id)))<EOL>error_code = "<STR_LIT>".encode("<STR_LIT:utf-8>")<EOL>return AmsResponseData(state, error_code, "<STR_LIT>".encode("<STR_LIT:utf-8>"))<EOL><DEDENT>error_code = ("<STR_LIT:\x00>" * <NUM_LIT:4>).encode("<STR_LIT:utf-8>")<EOL>response_data = error_code + response_content<EOL>return AmsResponseData(state, request.ams_header.error_code, response_data)<EOL>
Handle incoming requests and create a response.