こんにちは。フレクシェのミヤモトです。 弊社Webページで

FLEXSCHE CarryOutは工場のあらゆるデバイスと連携する機能を持った製品です。

とお伝えしていますが、具体的にどのように使えるのかイメージが湧かない、というご意見を頂くことがあります。

我々も工場を持っていれば、お連れして実際に動作しているところを見せられるのですがそうはいきません。そこで色々なデバイスを揃えて、弊社内でも実際の運用を想定した検証をしています。

その一環で、先日ロボットアーム(DOBOT Magician®)が届きました。研究用なので、デスクの上に乗る小型のものです。
ちなみにこのロボットは出力が小さく、ぶつかったら停止するようになっているので、いわゆる"協働ロボット"の括りに入るそうです。

検証する内容を決めるため、工場で協働ロボットがどのように活用されるか、ChatGPTへ聞いてみました。

協働ロボットは製造現場で様々な用途に活用されています。具体的な例を3つ挙げて解説いたします。
- 梱包工程 協働ロボットは、ピック・プレースや箱詰め、蓋締め、段ボール梱包など、梱包工程の自動化に利用されます。特に多品目小ロットの梱包現場では、効率化と省人化が期待されています。
- 検査工程 品質を一定に保つ再現性のある検査を実現するために、協働ロボットが導入されています。操作性が高いため、ティーチングが容易で、作業者の負担軽減にもつながります。
- 充填・包装工程 食品工場などで見られる充填・包装工程では、協働ロボットが作業者の労力を省き、生産性の向上に貢献しています。作業環境の改善にもつながっており、人手不足の解消にも役立っています。

今回の記事では、梱包工程のピック・プレースに絞って検証してみます。
実際にやることは、①ブロックで部品をつくって置き場におく、②置き場から拾ってトレイに配置する、の２つです。作業員として私が①を、ロボットが②を担当します。

流れとしては、
1. FLEXSCHE GPで計画を立案し、FLEXSCHE CarryOutサーバーに送信
2. CarryOutサーバーから、CarryOutクライアントに作業指示を送信
3. 作業員が部品を組み立て、部品を置き場におく
4. CarryOutクライアントから、部品を置いた場所をCarryOutサーバーに送信
5. CarryOutサーバーから、ロボットサーバーに部品の置き場を送信
6. ロボットサーバーから、ロボットアームへ指定された置き場からトレイへ運搬する命令を送信
7. ロボットがトレイへ配置
のようになります。併せて各作業の開始・完了についても、CarryOutサーバーに送信します。

CarryOutサーバーからロボットサーバーに作業指示を送る部分は、CarryOutクライアントから組立完了の実績を送信したらスケジューリングルールを実行するようにします。そのルールの中で外部プロセス実行をしてロボットサーバーに作業の情報をpostします。
ここについてはもっと良いやり方がありそうですが、今回はお試しということで良しとします。

ロボットサーバーについては、これはhttpでpostされたリクエストを処理できて、CarryOutサーバーに実績を入力できればよいです。
私はPythonが好きなので、Pythonで実装しました。せっかくなのでコードを載せておきます。

コードはこちら

import threading
import json
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import requests

POST_URL = r'{CarryOutサーバーのURL}/data/events?uuid=robot_server'
JST = timezone(timedelta(hours=+9), 'JST')


class HttpRequestHandler(BaseHTTPRequestHandler):
	execute_lock = threading.Lock()

	def do_OPTIONS(self):
		self.send_response(200, "ok")
		self.send_header('Access-Control-Allow-Origin', '*')
		self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS, GET')
		self.send_header("Access-Control-Allow-Headers", "X-Requested-With")
		self.send_header("Access-Control-Allow-Headers", "Content-Type")
		self.end_headers()

	def do_POST(self) -> requests.Response:
		# 受け取ったリクエストの解析
		content_len = int(self.headers.get('content-length'))
		req_body: str = self.rfile.read(content_len).decode("utf-8")
		request_data: dict = json.loads(req_body)

		# 製造完了実績の取得
		try:
			place_no: int = int(request_data["place_no"])
			original_operation_code: str = request_data["operation_code"]
		except KeyError as e:
			self.send_response(400)
			self.send_header('Content-type', 'application/json')
			self.end_headers()
			responseBody = f"Bad Request, {e}"
			self.wfile.write(responseBody.encode('utf-8'))
			return responseBody
		order_code: str = original_operation_code.split(":")[0]
		result_operation_code: str = order_code + ":輸送"

		with self.execute_lock:
			# 開始実績の送信
			cos_request_helper = COSRequestHelper()
			_ = cos_request_helper.send_manu_start_request_to_cos(result_operation_code, POST_URL)

			# ロボットを動かして輸送の処理
			some_robot_operation()

			# 終了実績の送信
			_ = cos_request_helper.send_manu_end_request_to_cos(result_operation_code, POST_URL)
		return


class COSRequestHelper:
	def __init__(self):
		pass

	def fetch_request_body(self, operation_code: str, event: str) -> dict:
		time_now_str = datetime.now(JST).strftime("%Y/%m/%d %H:%M:%S")
		request_body = {
			'target': operation_code,
			'event': event,
			'target_type': 'operation',
			'ev_datetime': time_now_str,
			'sent_at': time_now_str,
			'sender': 'robot_server'
		}
		return request_body

	def send_manu_start_request_to_cos(
			self,
			operation_code: str,
			post_url: str
			) -> requests.Response:
		request_body: dict = self.fetch_request_body(operation_code, 'manu_start')
		response = requests.post(post_url, json=request_body)
		return response

	def send_manu_end_request_to_cos(
			self,
			operation_code: str,
			post_url: str
			) -> requests.Response:
		request_body: str = self.fetch_request_body(operation_code, 'manu_end')
		response = requests.post(post_url, json=request_body)
		return response


def main():
	server = ThreadingHTTPServer(("", 8000), HttpRequestHandler)
	server.serve_forever()


if __name__ == "__main__":
	main()
  

これらを実装して動かしてみた結果はこんな感じになりました。私とロボットでうまく役割分担できていますね。

動画の右奥に映っているのはFLEXSCHE Viewerで、上段が私、下段がロボットです。
作業の前後で実績をCarryOutサーバーへ送信することで、ここにリアルタイムな進捗状況が反映されます。
実はやっているときにも気づいていましたが、ロボットが若干遅いのが良くわかりますね。

今回はCarryOutとロボットを連携させる部分に着目して検証してみました。ロボットを動かしたのは初めてですが、動かす部分さえ作ってしまえば、CarryOutと連携する部分は比較的簡単に実装できました。
検証したデバイスは、順次フレクシェスマート工場ワークショップで公開していきます。興味のある方は参加いただけると、より実際の工場で運用する際のイメージをつかみやすくなると思いますので、ご検討いただけますと幸いです。