效果

本教程主要是復現了Flexsim 2022最新的官方文檔的樣例。廢話不多說先上優化前后效果（相同倍速），主要是根據等待時間策略學習出了比較好的處理產品的先后順序。

Flexsim強化學習優化前

Flexsim強化學習優化后

環境依賴

Flexsim 2022

Python 3.0，相關庫包括
Gym
Stable-baselines3

模型搭建

Flexsim模型搭建

新建模型，拖拉 Source, Queue, Processor, Sink元素。并連接元素。

在Toolbox中，添加一個Global Table；在table的properties中，將表格重命名為ChangeoverTimes，并將行數和列數改為5，添加以下元素。

表格代表由工件i換到工件j時所消耗的時間。

點擊Processor來編輯它的properties，在Setup Time中，從下拉菜單選擇From/To Lookup Table。為Table選擇ChangeoverTimes/

點擊Source來編輯它的Properties，在triggers中為On creation 添加要素，選擇Data>set Label and color。將值改為duniform(1,5,getstream(current))。隨機產生五種產品。

這個時候保存模型為 ChangeoverTimesRL.fsm。就可以看到模型隨機產生物品。

在Toolbox中加入 Statistics > Model Parameter Table.

將Parameter1表格重命名為Observations，同理創建一個名為Actions的表格。

將Observations的Parameter2重命名為LastItemType; 在這一行的value將值設置為整數，上限值為5.

9. 將Actions表格中的Parameter3命名為ItemType。將value限定為整數，上限5同樣對應5種商品。

點擊Processor，在Properties中，點擊Pull，在Pull Strategy的下拉菜單，選擇Pull Best Item 選項。

在顯示出的Label中，選擇Custom Value，，并輸入

item.Type == Model.parameters["ItemType"].value

保存模型，運行可以看到紅色的優先被拉去。

為模型添加強化學習功能。

在Toolbox，添加Connectivity > Reinforcement Learning

在Observation Space中，選擇Discrete， 在Observation的參數中，選擇LastItem Type；在 Action Space中，選擇Discrete，并選擇ItemType.

點擊Apply，點擊Processor，在Setup Time picklist選擇Codebutton，可以看到“f_lastlabelval”，之后要用到這個值

返回強化學習的屬性窗口，在On observation 的trigger添加，并選擇Code Snippet。將描述文字從Code Snippet改為Set observation parameter。

將下面代碼粘貼到field

Model.parameters["LastItemType"].value = getvarnum(Model.find("Processor1"), "f_lastlabelval");

6. 返回3D視圖，點擊sink，在Labels中，添加一個Number label標簽，并命名為LastTime
7. 再添加一個標簽命名為Reward，勾選Automatically Reset按鈕，并保存
8. 在Triggers中添加On Entry，添加Data>Increment Value選項，在Increment的下拉菜單，選擇current.labels[“Reward”]，在by中輸入 10/(Model.time - current.LastTime)
9. 添加一個Data> Set Label選項，Object 選擇Current，Label選為“LastTime”，在Value選擇Model.time。這就是我們的獎勵計算

10. 返回強化學習的屬性，編輯獎勵方程，將其命名從 Reward Function 改為 Reward based on throughput。粘貼下面的代碼。

double reward = Model.find("Sink1").Reward; Model.find("Sink1").Reward = 0; int done = (Model.time > 1000); return [reward, done];

11. 在On Request Action中，添加Take a Random Action選擇
12. 在Decision Events中添加一個新的event ，選擇Pull Strategy選項。

保存并運行模型。

Python部分

依次，修改并運行下述代碼，分別起到測試接口，訓練，測試功能。



紅色的部分是要修改的地方。
flexsim_env.py

import gym import os import subprocess import socket import json from gym import error, spaces, utils from gym.utils import seeding import numpy as npclass FlexSimEnv(gym.Env):metadata = {'render.modes': ['human', 'rgb_array', 'ansi']}def __init__(self, flexsimPath, modelPath, address='localhost', port=5005, verbose=False, visible=False):self.flexsimPath = flexsimPathself.modelPath = modelPathself.address = addressself.port = portself.verbose = verboseself.visible = visibleself.lastObservation = ""self._launch_flexsim()self.action_space = self._get_action_space()self.observation_space = self._get_observation_space()def reset(self):self._reset_flexsim()state, reward, done = self._get_observation()return statedef step(self, action):self._take_action(action)state, reward, done = self._get_observation()info = {}return state, reward, done, infodef render(self, mode='human'):if mode == 'rgb_array':return np.array([0,0,0])elif mode == 'human':print(self.lastObservation)elif mode == 'ansi':return self.lastObservationelse:super(FlexSimEnv, self).render(mode=mode)def close(self):self._close_flexsim()def seed(self, seed=None):self.seedNum = seedreturn self.seedNumdef _launch_flexsim(self):if self.verbose:print("Launching " + self.flexsimPath + " " + self.modelPath)args = [self.flexsimPath, self.modelPath, "-training", self.address + ':' + str(self.port)]if self.visible == False:args.append("-maintenance")args.append("nogui")self.flexsimProcess = subprocess.Popen(args)self._socket_init(self.address, self.port)def _close_flexsim(self):self.flexsimProcess.kill()def _release_flexsim(self):if self.verbose:print("Sending StopWaiting message")self._socket_send(b"StopWaiting?")def _get_action_space(self):self._socket_send(b"ActionSpace?")if self.verbose:print("Waiting for ActionSpace message")actionSpaceBytes = self._socket_recv()return self._convert_to_gym_space(actionSpaceBytes)def _get_observation_space(self):self._socket_send(b"ObservationSpace?")if self.verbose:print("Waiting for ObservationSpace message")observationSpaceBytes = self._socket_recv()return self._convert_to_gym_space(observationSpaceBytes)def _reset_flexsim(self):if self.verbose:print("Sending Reset message")resetString = "Reset?"if hasattr(self, "seedNum"):resetString = "Reset:" + str(self.seedNum) + "?"self._socket_send(resetString.encode())def _get_observation(self):if self.verbose:print("Waiting for Observation message")observationBytes = self._socket_recv()self.lastObservation = observationBytes.decode('utf-8')state, reward, done = self._convert_to_observation(observationBytes)return state, reward, donedef _take_action(self, action):actionStr = json.dumps(action, cls=NumpyEncoder)if self.verbose:print("Sending Action message: " + actionStr)actionMessage = "TakeAction:" + actionStr + "?"self._socket_send(actionMessage.encode())def _socket_init(self, host, port):if self.verbose:print("Waiting for FlexSim to connect to socket on " + self.address + ":" + str(self.port))self.serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.serversocket.bind((host, port))self.serversocket.listen();(self.clientsocket, self.socketaddress) = self.serversocket.accept()if self.verbose:print("Socket connected")if self.verbose:print("Waiting for READY message")message = self._socket_recv()if self.verbose:print(message.decode('utf-8'))if message != b"READY":raise RuntimeError("Did not receive READY! message")def _socket_send(self, msg):totalsent = 0while totalsent < len(msg):sent = self.clientsocket.send(msg[totalsent:])if sent == 0:raise RuntimeError("Socket connection broken")totalsent = totalsent + sentdef _socket_recv(self):chunks = []while 1:chunk = self.clientsocket.recv(2048)if chunk == b'':raise RuntimeError("Socket connection broken")if chunk[-1] == ord('!'):chunks.append(chunk[:-1])break;else:chunks.append(chunk)return b''.join(chunks)def _convert_to_gym_space(self, spaceBytes):paramsStartIndex = spaceBytes.index(ord('('))paramsEndIndex = spaceBytes.index(ord(')'), paramsStartIndex)type = spaceBytes[:paramsStartIndex]params = json.loads(spaceBytes[paramsStartIndex+1:paramsEndIndex])if type == b'Discrete':return gym.spaces.Discrete(params)elif type == b'Box':return gym.spaces.Box(np.array(params[0]), np.array(params[1]))elif type == b'MultiDiscrete':return gym.spaces.MultiDiscrete(params)elif type == b'MultiBinary':return gym.spaces.MultiBinary(params)raise RuntimeError("Could not parse gym space string")def _convert_to_observation(self, spaceBytes):observation = json.loads(spaceBytes)state = observation["state"]if isinstance(state, list):state = np.array(observation["state"])reward = observation["reward"]done = (observation["done"] == 1)return state, reward, doneclass NumpyEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, np.integer):return int(obj)elif isinstance(obj, np.floating):return float(obj)elif isinstance(obj, np.ndarray):return obj.tolist()return json.JSONEncoder.default(self, obj)def main():env = FlexSimEnv(flexsimPath = "C:/Program Files/FlexSim 2022/program/flexsim.exe",modelPath = "E:/劉一陽資料/Flexsim/demo/ChangeoverTimesRL.fsm",verbose = True,visible = True)for i in range(2):env.seed(i)observation = env.reset()env.render()done = Falserewards = []while not done:action = env.action_space.sample()observation, reward, done, info = env.step(action)env.render()rewards.append(reward)if done:cumulative_reward = sum(rewards)print("Reward: ", cumulative_reward, "\n")env._release_flexsim()input("Waiting for input to close FlexSim...")env.close()if __name__ == "__main__":main()

flexsim_training.py

import gym from flexsim_env import FlexSimEnv from stable_baselines3.common.env_checker import check_env from stable_baselines3 import PPO from stable_baselines3.common.env_util import make_vec_envdef main():print("Initializing FlexSim environment...")# Create a FlexSim OpenAI Gym Environmentenv = FlexSimEnv(flexsimPath = "C:/Program Files/FlexSim 2022/program/flexsim.exe",modelPath = "E:/劉一陽資料/Flexsim/demo/ChangeoverTimesRL.fsm",verbose = False,visible = False)check_env(env) # Check that an environment follows Gym API.# Training a baselines3 PPO model in the environmentmodel = PPO("MlpPolicy", env, verbose=1)print("Training model...")model.learn(total_timesteps=50000)# save the modelprint("Saving model...")model.save("ChangeoverTimesModel")input("Waiting for input to do some test runs...")# Run test episodes using the trained modelfor i in range(4):env.seed(i)observation = env.reset()env.render()done = Falserewards = []while not done:action, _states = model.predict(observation)observation, reward, done, info = env.step(action)env.render()rewards.append(reward)if done:cumulative_reward = sum(rewards)print("Reward: ", cumulative_reward, "\n")env._release_flexsim()input("Waiting for input to close FlexSim...")env.close()if __name__ == "__main__":main() import json from stable_baselines3 import PPO from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse, parse_qs import numpy as npclass FlexSimInferenceServer(BaseHTTPRequestHandler):def do_GET(self):params = parse_qs(urlparse(self.path).query)self._handle_reply(params)def do_POST(self):content_length = int(self.headers['Content-Length'])body = self.rfile.read(content_length)params = parse_qs(body)self._handle_reply(params)def _handle_reply(self, params):if len(params):observation = []if b'observation' in params.keys():observationBytes = params[b'observation'][0]observation = np.array(json.loads(observationBytes))elif 'observation' in params.keys():observationBytes = params['observation'][0]observation = np.array(json.loads(observationBytes))if isinstance(observation, list):observation = np.array(observation)action, _states = FlexSimInferenceServer.model.predict(observation)self.send_response(200)self.send_header("Content-type", "application/json")self.end_headers()self.wfile.write(bytes(json.dumps(action, cls=NumpyEncoder), "utf-8"))returnself.send_response(200)self.send_header("Content-type", "text/html")self.end_headers()self.wfile.write(bytes("", "utf-8"))class NumpyEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, np.integer):return int(obj)elif isinstance(obj, np.floating):return float(obj)elif isinstance(obj, np.ndarray):return obj.tolist()return json.JSONEncoder.default(self, obj)def main():print("Loading model...")model = PPO.load("ChangeoverTimesModel.zip")FlexSimInferenceServer.model = model# Create server objectprint("Starting server...")hostName = "localhost"serverPort = 8890webServer = HTTPServer((hostName, serverPort), FlexSimInferenceServer)print("Server started http://%s:%s" % (hostName, serverPort))# Start the web servertry:webServer.serve_forever()except KeyboardInterrupt:passwebServer.server_close()print("Server stopped.")if __name__ == "__main__":main()

運行到最后一個模型可以得到一個本地的ip接口。
并在模型中修改。
現在再運行就可以了。

總結

以上是生活随笔為你收集整理的Flexsim 强化学习的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。