【連載】LCIの使い方「エレベーター編」#11｜利用解除(Release)

Last updated at 2026-11-10Posted at 2026-10-07

こんにちは！この記事では、株式会社Octa Roboticsが開発・提供しているロボット・設備間連携に特化したマルチベンダー型のインターフェースサービス「LCI」について、ロボットからの利用方法を連載形式で紹介していきます。

前回の記事では、通信プロトコルの「降車・降車完了(RobotStatus)」の実装を行いました。

今回は、「利用解除(Release)」について実装を行っていきます。

前回 : #10 降車・降車完了(RobotStatus)

おさらい

利用解除(Release)のシーケンスについて、再確認していきましょう。

利用解除では、それぞれ以下の手順でロボットとエレベーター（LCI）間でMQTT通信を行います。

利用解除(MQTT)

1. ロボットからエレベーターにReleaseを送信
2. エレベーターからロボットにReleaseResultを送信

プロトコルの仕様

ReleaseとReleaseResultの仕様を確認していきます。

トピック

API名	トピック
Release	lci/<bldg_id>/<bank_id>/<elevator_id>/Release/<robot_id>
ReleaseResult	/lci/<bldg_id>/<bank_id>/<elevator_id>/ReleaseResult/<robot_id>

第3回で使用した設定ファイル：server_config_simulator_1-1.yamlを利用する場合には、以下になります。

API名	トピック
Release	/lci/simulator/1/1/Release/_uClosc2
ReleaseResult	/lci/simulator/1/1/ReleaseResult/_uClosc2

bldg_id : simulator(設定ファイル内のlci_bldg_id)
bank_id : 1（設定ファイル内のlci_bank_id）
elevator_id : 1（設定ファイル内のlci_elevator_id）
robot_id : _uClosc2(開発用ロボットアカウントのID)

ペイロード

API名	ペイロード
Release	{ "timestamp": <timestamp>, "robot_id":<robot_id> }
ReleaseResult	{ "result": <result>, "timestamp": <timestamp>, "requested_robot_id": <requested_robot_id>, "requested_timestamp": <requested_timestamp> }

timestamp : ReleaseもしくはReleaseResultが送信された時間（エポック秒)
robot_id : Releaseを行うロボットのID(今回は_uClosc2)
result : RleaseResultの結果(1:成功、2:失敗、3:その他エラー、99:管制運転中)
requested_robot_id : Releaseを行うロボットのID(今回は_uClosc2)
requested_timestamp : Releaseを送信した時間

サンプルコードの確認

サンプルコードにおける利用解除に関連する箇所を確認してきましょう。

do_release

この関数では、以下のことを行っています。

ペイロードの作成
トピックの作成
メッセージのパブリッシュ(self._publish_and_wait_responce)
ReleaseResultのレスポンスを待つ(self._publish_and_wait_responce)

class LciClient

def do_release(self):
       requested_time = int(time.time() * 1000)/1000
       payload = json.dumps({
           'timestamp': requested_time,

           'robot_id': self.robot_id})

       api = 'Release'
       topic = f'{self.topic_prefix}/{self.elevator_id}/{api}/{self.robot_id}'

       res = self._publish_and_wait_response(topic, payload, api, requested_time, 3.0)

       self.registered = False
       return res

メイン処理(if name = ‘main’)

‘do_release’をメイン処理の最後に実行し、連携を終了します。

main routine

if __name__ == '__main__':

   ...

   #利用解除
   res = lci_client.do_release()
   lci_client.logger.debug(res)
   time.sleep(1)

サンプルコードの改良

今回は、以下の項目をもとにサンプルコードの改良を進めていきます。

メイン処理の開始時に利用解除を一度実行
各コマンドでエラーが発生したら、利用解除を行いメイン処理を終了する
エラーが発生した場合、以下を分かるようにする
- メイン処理のどこでエラーが発生したか
- どのコマンドでエラーが発生したか
- エラーが発生したコマンドにおけるペイロード内の情報

メイン処理開始時に行う利用解除は、利用登録が残っていた場合にエラーが発生しないようにすることが目的です。

作成したコードにおいて、エラーは以下を対象としています。

resultの値が2, 3, 99の場合
resultの値がNoneの場合
- コマンドのレスポンスにおけるペイロードの値がそれぞれNoneの場合
各コマンドで設定したタイムアウトを超えた場合
- コマンドのレスポンス自体がNoneの場合

エラーが起きた際の各情報を取得できるようにする理由は、主に以下になります。

ロボットの異常系シーケンスを、エラーが発生したタイミングに応じて適切に実施するため
ログをもとにしたエラーの原因解明をより行いやすくするため

実装したコード

上記の項目をもとに、サンプルコードに追記・変更を加えた箇所を記載します。

追加した変数

self.error_info : エラー発生時の情報を格納する変数
self.lock_error : self.error_infoの値を取り出す・上書きする際の排他処理に使用

class LciClient

def __init__(self):
    ...
    self.error_info = None
    self.lock_error = threading.Lock()

get_error_info

self.error_infoの中身を返します。

この関数は、スレッドで実行しているRequestElevatorStatus/ElevatorStatusとRobotStatus/RobotStatusResult(state:5)におけるエラー情報を取得する目的で使用します。

class LciClient

def get_error_info(self):
    with self.lock_error:
        return self.error_info

set_error_info

self.error_infoにエラー情報を上書きします。

この関数は、スレッドで実行しているRequestElevatorStatus/ElevatorStatusとRobotStatus/RobotStatusResult(state:5)におけるエラー情報を更新する目的で使用します。

class LciClient

def set_error_info(self, res):
       with self.lock_error:
           if res:
               self.error_info = res
           else:
               self.error_info = None

do_registration

制限時間(self.timeout_sec)を超えても利用登録が成功しなかった場合に、呼び出し元にresを返すよう一部変更しました。
加えて、res.resultに応じた条件式も簡略化しています。

class LciClient

  def do_registration(self, dry_run: bool = False):
        max_timeout = time.time() + self.timeout_sec
        
-       while time.time() < max_timeout:
+       while True:
            res = self.registration(dry_run)

+           if time.time() > max_timeout:
+               return res

            if res is not None:
                if res.result == 2:
                    time.sleep(5)
                elif res.result == 1:
                    if dry_run:
                        self.dry_run_registered = True
                    else:
                        self.registered = True
                    return res
                else:
                    return res
            else:
                break

do_request_elevator_status

res.resultが1でない場合とresがNoneの場合に、set_error_infoを実行するよう追記しました。

class LciClient

def do_request_elevator_status(self):
       time_out = 10

       while not self.stop_event.is_set():
           res = self.request_elevator_status(time_out)
           self.logger.debug(res)
          
           if res is not None:
               if res.result != 1:
+                  self.set_error_info(res)
                   self.registered = False
                   break
           else:
+              self.set_error_info(res)
               self.registered = False
               break
          
           time.sleep(2)

robot_status_open_hold_worker

res.resultが1でない場合とresがNoneの場合に、set_error_infoを実行するよう追記しました。

class LciClient

def robot_status_open_hold_worker(self):
       time_out = 10
       limit_time = self.timeout_sec + time.time()

       while True:
           if time.time() < limit_time:
               with self.lock_robot_status:
                   if self.stop_event_robot_status.is_set():
                       break
                  
                   # 乗り込み
                   res = self.robot_status(RobotStatus.KEEP_DOOR_OPEN, time_out)
                   self.logger.debug(res)

                   if res is not None:
                       if res.result != 1:
+                          self.set_error_info(res)
                           self.registered = False
                           break
                   else:
+                      self.set_error_info(res)
                       self.registered = False
                       break
              
               self.stop_event_robot_status.wait(5)
                  
           else:
               self.registered = False
               break

wait_for_arrival

コマンドのレスポンスとしてget_error_infoからエラー情報を取得し、resに格納します。
その後、エレベーター到着およびドア開閉の確認結果であるenableとresを呼び出しもとに返すよう追記しました。

class LciClient

def wait_for_arrival(self, target_floor: str, target_door: int = 1):
       #エレベーター到着確認
       limit_time = self.timeout_sec + time.time()
       enable = False
      
       #追記:Release True -> self.is_registered()
       #self.registeredは各コマンドでリザルトに応じてFalseに変更しているので、エラーが発生したことと同義と判断
       while self.is_registered():
           if time.time() < limit_time:
               if self.floor_value == target_floor \
                   and self.door_value == target_door:
                   enable = True
                   break
          
           else:
               self.registered = False
               if self.floor_value != target_floor:
                   self.logger.error("Elevator hasn't arrived")
               if self.door_value != target_door:
                   self.logger.error("Door hasn't opened")
               break
              
           time.sleep(0.05)

+       res = self.get_error_info()

-       return res
+       return enable, res

reset

self.error_infoをNoneにリセットするよう追記します。

class LciClient

def reset(self):
       self.stop_robot_status_open_hold()
       self.stop_request_elevator_status()
       self.clear_robot_status_stop_event()
+      self.error_info = None

SequenceError

今回、作成した独自の例外クラスになります。
エラーが発生した際に、このクラスに各エラー情報を格納します。
今回は、以下の情報を格納するものとしました。

self.step : メイン処理上でエラーが発生したステップ
self.command : エラーが発生したコマンド（Registrationなど）
self.result : エラーが発生したコマンドにおけるresult値
self.timestamp : コマンドに対するレスポンスが送信された時間
self.requested_timestamp : コマンドを送信した時間
self.requested_robot_id : コマンドを送信したロボットのrobot_id

乗り込み中止および降車中止はエラー情報を格納しません。

class LciClient

class SequenceError(Exception):
   def __init__(self, step, error_info):
       self.step = step

       if error_info:
           self.command = error_info.orig_api
           self.result = error_info.result
           self.timestamp = error_info.timestamp
           self.requested_timestamp = error_info.requested_timestamp
           self.requested_robot_id = error_info.requested_robot_id
       else:
           self.command = None
           self.result = None
           self.timestamp = None
           self.requested_timestamp = None
           self.requested_robot_id = None
      
       if not self.step in ("RobotStatus(state:3)", "RobotStatus(state:4)"):
           message = "Error occured at " + self.step + "."
       else:
           message = self.step + " is send."

       # exceptionクラスのmessageをオーバーライド
       super().__init__(message)

メイン処理(if name=='main':)

変更点は以下になります。

利用登録開始前に利用解除を実施
初回の利用解除から本番の利用登録までは、resがNone、もしくはresultの値が1以外の場合にraise SequenceErrorを実施
本番の利用登録以降はself.registeredがFalseの場合に、raise SequenceErrorを実施
try: except SequenceError as e:で、raise SequenceErrorをキャッチし、エラー情報を表示
try: finally:でメイン処理の最後に、def resetの実行と利用解除を実施
wait_for_arrivalからの返り値をenableとresに格納

メイン処理で扱うres(各コマンドにおけるレスポンス)は以下のパターンで中身が異なります。

タイムアウトせず、LCIからのレスポンスを受け取った場合
- 中身は各コマンドごとに定義されたペイロード
  例：Registration/RegistrationResult

orig_api: 'Registration', result: '1', elevator_id: '1', timestamp: '1791166603.02', requested_robot_id: '_uClosc2', requested_timestamp: '1791166602.979'

タイムアウトした場合
- 中身はNone
self.registeredがFalseの場合
- 中身はLciResponceクラスのインスタンス
  - 各変数の値がNoneの状態

main routine

if __name__ == '__main__':

   ...
  
   try:
       #初回利用解除　
+      res = lci_client.do_release()
+      lci_client.logger.debug(res)
+      if not res or res.result != 1:
+          raise SequenceError("1st Release", res)
+      time.sleep(1)

       #dry runによる疎通確認　
       res = lci_client.do_registration(True)
       lci_client.logger.debug(res)
+      if not res or res.result != 1:
+          raise SequenceError("DryRun Registration", res)
       time.sleep(1)

       #本番の利用登録　
       res = lci_client.do_registration()
       lci_client.logger.debug(res)
+      if not res or res.result != 1:
+          raise SequenceError("Registration", res)
       time.sleep(1)
      
       if res != None:
           lci_client.set_elevator_id(res.elevator_id)

       #出発階への呼び出し　
       res = lci_client.do_call_elevator('B1', '1F')
       lci_client.logger.debug(res)
+      if not lci_client.is_registered():
+          raise SequenceError("1st CallElevator", res)
       time.sleep(1)

       #エレベーターの状態取得
       lci_client.start_request_elevator_status()
       time.sleep(1)

       #エレベーターの出発階到着・ドアの開閉状態確認　
-      res = lci_client.wait_for_arrival('B1', 1)
+      enable, res = lci_client.wait_for_arrival('B1', 1)

       #乗り込み
       if enable:
           #戸開継続要求
           lci_client.start_robot_status_open_hold()
           """
           ~~~ロボットのエレベーター乗り込み動作~~~
           ここでロボットによるエレベーターの乗り込みを実施します.
           今回は20秒で、ロボットが乗り込みを行ったとみなします。
           """
           roop_time = 20 + time.time()
           while time.time() < roop_time:
+              if not lci_client.is_registered():
+                  raise SequenceError("1st RobotStatus(state:5)", lci_client.get_error_info())
               time.sleep(1)
      
       else:
+           raise SequenceError("1st WaitForArrival", res)

       #ロボットの乗り込みが終了した場合→True
       #ロボットの乗り込みを中止したい場合→False
       end_boarding = True
      
       with lci_client.lock_robot_status:
           lci_client.stop_robot_status_open_hold()
          
           #乗り込み完了
           if end_boarding:
               res = lci_client.do_robot_status_success_enter()
               lci_client.logger.debug(res)
+              if not lci_client.is_registered():
+                  raise SequenceError("RobotStatus(state:1)", res)

           #乗り込み中止
           else:
               res = lci_client.do_robot_status_cancel_enter()
               lci_client.logger.debug(res)
+              raise SequenceError("RobotStatus(state:3)", res)
          
       time.sleep(1)
      
       #RobotStatusに関するイベントのリセット
       lci_client.clear_robot_status_stop_event()
       time.sleep(1)

       #行き先階への登録
       res = lci_client.do_call_elevator('B1', '1F')
       lci_client.logger.debug(res)
+      if not lci_client.is_registered():
           raise SequenceError("2nd CallElevator", res)
       time.sleep(1)

       #エレベーターの目的階到着・ドアの開閉状態確認
-      res = lci_client.wait_for_arrival('1F', 1)
+      enable, res = lci_client.wait_for_arrival('1F', 1)

       if enable:
           #戸開継続要求
           lci_client.start_robot_status_open_hold()
           """
           ~~~ロボットのエレベーター降車~~~
           ここでロボットによるエレベーターの降車を実施します.
           今回は20秒で、ロボットが降車を行ったとみなします。
           """
           roop_time = 20 + time.time()
           while time.time() < roop_time:
+              if not lci_client.is_registered():
+                  raise SequenceError("2nd RobotStatus(state:5)", lci_client.get_error_info())
               time.sleep(1)
       else:
+          raise SequenceError("2nd WaitForArrival", res)

       #エレベーター状態取得終了
       lci_client.stop_request_elevator_status()
       time.sleep(1)

       #ロボットの降車が終了した場合→True
       #ロボットの降車を中止したい場合→False
       end_alighting = False
      
       with lci_client.lock_robot_status:
           lci_client.stop_robot_status_open_hold()
          
           #降車完了
           if end_alighting:
               res = lci_client.do_robot_status_success_exit()
               lci_client.logger.debug(res)
+              if not lci_client.is_registered():
+                  raise SequenceError("RobotStatus(state:2)", res)
                  
           #降車中止
           else:
               res = lci_client.do_robot_status_cancel_exit()
               lci_client.logger.debug(res)
+              raise SequenceError("RobotStatus(state:4)", res)
                  
       time.sleep(1)
      
       #RobotStatusに関するイベントのリセット
       lci_client.clear_robot_status_stop_event()
       time.sleep(1)
  
+  except SequenceError as e:
+      if not e.step in ("RobotStatus(state:3)", "RobotStatus(state:4)"):
+          lci_client.logger.debug(f"{e} "
+                                  f"error command : {e.command}, "
+                                  f"result : {e.result}, "
+                                  f"timestamp : {e.timestamp}, "
+                                  f"requested timestamp : {e.requested_timestamp}, "
+                                  f"requested robot id : {e.requested_robot_id}"
+                                  )
+      else:
+          lci_client.logger.debug(f"{e}")

+  finally:
+      lci_client.reset()
+      res = lci_client.do_release()
+      lci_client.logger.debug(res)

動作確認

エラーが発生した際に、各エラー情報がログに出力されるかを確認します。
今回は初回のCallElevatorにおいて、resultが2であったケースで確認します。

コード変更箇所

_publish_and_wait_responseにおいて以下のように、加筆を行います。
これにより、CallElevatorResultのペイロードの中身を書き換えます。

class LciClient

 def _publish_and_wait_response(self, topic: str, payload: str, orig_api: str, requested_timestamp: float, timeout_sec: float):
       with self.mqtt_lock:
           self._publish(topic, payload)
           res = self.wait_response(orig_api, requested_timestamp, timeout_sec)

+          if res.orig_api == "CallElevator":
+              res.result = 2

           return res

実行結果

下記にコード実行時のログを一部記載しています。
このログから読み取れる動作として、CallElevatorにエラーが発生した場合に各エラー情報がログとして出力されています。

13:01:40,418 ~ 13:01:40,580 : 初回の利用解除
13:01:41,586 ~ 13:01:41,718 : 利用登録（ドライラン）
13:01:42,755 ~ 13:01:44,206 : 利用登録（本番）
13:01:45,221 ~ 13:01:45,417 : 初回のエレベーター呼び出し（エラー発生）
13:01:45,442 : エラー情報の表示
13:01:45,442 ~ 13:01:45,651 : 利用解除

2026-10-05 13:01:40,418    DEBUG Sending PUBLISH (d0, q1, r0, m3), 'b'/lci/simulator/1/1/Release/_uClosc2'', ... (53 bytes)
2026-10-05 13:01:40,580    DEBUG Received PUBLISH (d0, q1, r0, m1), '/lci/simulator/1/1/ReleaseResult/_uClosc2', ...  (115 bytes)
2026-10-05 13:01:41,586    DEBUG Sending PUBLISH (d0, q1, r0, m4), 'b'/lci/simulator/1/1/Registration/_uClosc2'', ... (70 bytes)
2026-10-05 13:01:41,718    DEBUG Received PUBLISH (d0, q1, r0, m1), '/lci/simulator/1/1/RegistrationResult/_uClosc2', ...  (152 bytes)
2026-10-05 13:01:42,755    DEBUG Sending PUBLISH (d0, q1, r0, m5), 'b'/lci/simulator/1/1/Registration/_uClosc2'', ... (53 bytes)
2026-10-05 13:01:44,206    DEBUG Received PUBLISH (d0, q1, r0, m1), '/lci/simulator/1/1/RegistrationResult/_uClosc2', ...  (134 bytes)
2026-10-05 13:01:45,221    DEBUG Sending PUBLISH (d0, q1, r0, m6), 'b'/lci/simulator/1/1/CallElevator/_uClosc2'', ... (141 bytes)
2026-10-05 13:01:45,417    DEBUG Received PUBLISH (d0, q1, r0, m1), '/lci/simulator/1/1/CallElevatorResult/_uClosc2', ...  (115 bytes)
2026-10-05 13:01:45,442    DEBUG Error occured at 1st CallElevator. error command : CallElevator, result : 2, timestamp : 1791172905.294, requested timestamp : 1791172905.221, requested robot id : _uClosc2
2026-10-05 13:01:45,442    DEBUG Sending PUBLISH (d0, q1, r0, m7), 'b'/lci/simulator/1/1/Release/_uClosc2'', ... (53 bytes)
2026-10-05 13:01:45,651    DEBUG Received PUBLISH (d0, q1, r0, m1), '/lci/simulator/1/1/ReleaseResult/_uClosc2', ...  (115 bytes)

最後に

今回の利用解除でLCIの使い方「エレベーター編」は終了となります。
ここまで長らくお付き合い頂きまして誠にありがとうございました！

エレベーター編を通して改良を行ったコードについては、以下リンク先から確認できます。
→lci_client_tech_blog.py

次は、LCIを使用した「Door API」の使い方についてご紹介していく予定です。

ご興味のある方は、引き続きぜひご閲覧ください！

その他

本記事は、LCIプロトコル仕様書(オンライン)のv2025-10-04を参考に作成・加筆しています
LCI関連の開発に必要な情報が記載されている開発者用ポータルは以下からアクセスできます
- LCI Developer Portal
各種資料は以下から閲覧・ダウンロードできます
- LCIプロトコル仕様書(オンライン)
- サンプルコード・設定ファイル

サンプルコードは、以下の二つがアップロードされています。

lci_client.py : オリジナル
lci_client_tech_blog.py : 本ブログを通して改良したもの

開発用アカウントの発行は、以下からお問い合わせください
- 株式会社Octa Roboticsへの問い合わせ
- 開発用アカウント発行に関する参考資料

連載記事リスト

他の記事もチェックしてみてください！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up