測距センサーと鉄道おもちゃを使って 2 次元地図を作る簡易スキャナを作ってみた

from m5stack import *
from m5ui import *
from uiflow import *
from IoTcloud.AWS import AWS
import imu
import utime
import json
import ntptime
import time

setScreenColor(0x111111)
lcd.setRotation(2)
remoteInit()

iot_topic = None
recv = None
checksum = None
i = None
distance = None
data = None
status = False
nData = None
dt = None
st = None
starttime = None
endtime = None
records = None
record = None
nSkip = None

title = M5Title(title="Scanner", x=3, fgcolor=0xFFFFFF, bgcolor=0x0000FF)
l_dist = M5TextBox(15, 80, "Distance", lcd.FONT_DejaVu18, 0xFFFFFF, rotate=0)
messages = M5TextBox(0, 20, "Messages", lcd.FONT_Default, 0xFFFFFF, rotate=0)

lcd.qrcode('https://flow.m5stack.com/remote?id=123456789012345678', 0, 100, 130)

#---------------
# 初期化
#---------------
def values():
  global recv, distance, data, status, records, record, nSkip, nData, st, dt
  recv = None
  distance = 0
  data = {}
  status = False
  records = []
  record = {}
  nSkip = 30
  nData = 0
  dt = 0

#------------------------------
# AWS IoT Core へ Publish する
#------------------------------
def pubData():
  global iot_topic, st, records
  
  data = {
    'uuid': st,
    'records': records
  }
  
  aws.publish(str(iot_topic),json.dumps(data))
  
  # 送り終わったreccordsは空にする
  records = []

#---------------
# 計測開始
#---------------
def button_start_callback():
  global status, iot_topic, st, starttime, records
  
  # Set UUID for record  
  st = '' # uuid
  records = []
  try:
    now = time.localtime()
    st = '{}{:0=2}{:0=2}{:0=2}{:0=2}{:0=2}'.format(now[0],now[1],now[2],now[3],now[4],now[5])
  except Exception as e:
    messages.setText(str(e.args[0]))
  
  # Start to scan
  starttime = utime.ticks_ms()
  l_dist.setText('Scanning...')
  status = True

#---------------
# 計測終了
#---------------
def button_stop_callback():
  global status, iot_topic, nData, starttime
  status = False
  
  try:
    pubData()
    messages.setText(str(nData) + ' records were published.\nDuration: ' + str(utime.ticks_ms() - starttime))
    l_dist.setText('Finished')
  except Exception as e:
    messages.setText(str(e))
  
  records = []
  data = {}
  nData = 0
  


try:
  # AWS IoT 用初期設定
  messages.setText('Initializing for AWS...')
  aws = AWS(things_name='Scanner', host='xxxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com', port=8883, keepalive=1200, cert_file_path="/flash/res/certificate.pem.crt", private_key_path="/flash/res/private.pem.key")
  iot_topic = 'm5stickc/scanner' # IoT Topicを定義
  aws.start()
  messages.setText('Initializing for AWS...done!')
  
  # NTP サーバーとの時刻同期
  messages.setText('Get current time using NTP...')
  ntp = ntptime.client(host='jp.pool.ntp.org', timezone=9)
  rtc.setTime((ntp.year()), (ntp.month()), (ntp.day()), (ntp.hour()), (ntp.minute()), (ntp.second()))
  messages.setText('Get current time using NTP...done!')
  
  # 測距センサーの初期化
  messages.setText('Initializing scanning device...')
  uart1 = machine.UART(1, tx=32, rx=33)
  uart1.init(115200, bits=8, parity=None, stop=1)  
  messages.setText('Initializing scanning device...done!')

  # 各種変数の初期化
  values()
  
  # 計測
  while True:
    try:
      # Start ボタンが押されると計測開始
      if status == True:
        
        # 測距センサーからのデータ取得
        if uart1.any():
          recv = uart1.read(9)
          if len(recv) == 9:
            if recv[0] == 0x59 and recv[1] == 0x59:
              checksum = 0
              for i in range(0, 8, 1):
                checksum = checksum + recv[int(i)]
              checksum = checksum % 256
              if checksum == recv[8]:
                distance = recv[2] + recv[3] * 256
                
                # 計測時の Uptime と距離を JSON 形式で纏める
                record = {
                  'time': utime.ticks_ms(),
                  'distance': distance
                }
                
                # 配列に格納
                records.append(record)
                
                # nSkip 回に 1 回まとめて AWS IoT に Publish する
                if nData % nSkip == 0:
                  pubData()
                
                nData = nData + 1
    
    except Exception as e:
      messages.setText("Internal Message: " + str(e))

except Exception as e:
  messages.setText(str(e))

# Blender の python\bin フォルダ直下の python にて実行します
c:\Program Files\Blender Foundation\Blender 3.4\3.4\python\bin>python -m pip install -U boto3
Collecting boto3
  Using cached boto3-1.26.74-py3-none-any.whl (132 kB)
Collecting botocore<1.30.0,>=1.29.74
  Using cached botocore-1.29.74-py3-none-any.whl (10.4 MB)
Collecting jmespath<2.0.0,>=0.7.1
  Using cached jmespath-1.0.1-py3-none-any.whl (20 kB)
Collecting s3transfer<0.7.0,>=0.6.0
  Using cached s3transfer-0.6.0-py3-none-any.whl (79 kB)
Requirement already satisfied: urllib3<1.27,>=1.25.4 in c:\program files\blender foundation\blender 3.4\3.4\python\lib\site-packages (from botocore<1.30.0,>=1.29.74->boto3) (1.26.8)
Collecting python-dateutil<3.0.0,>=2.1
  Using cached python_dateutil-2.8.2-py2.py3-none-any.whl (247 kB)
Collecting six>=1.5
  Using cached six-1.16.0-py2.py3-none-any.whl (11 kB)
Installing collected packages: six, jmespath, python-dateutil, botocore, s3transfer, boto3
Successfully installed boto3-1.26.74 botocore-1.29.74 jmespath-1.0.1 python-dateutil-2.8.2 s3transfer-0.6.0 six-1.16.0

import boto3
import json
import base64
import math
import bpy


client = boto3.client('s3')
bucket_name = 'scanner-demo'

directory = "データが格納されているディレクトリ/"

response = client.list_objects_v2(
    Bucket=bucket_name,
    Prefix=directory
)
keys = [obj.get('Key') for obj in response['Contents']]

records = []
latest_time = 0
json_stream = None
for key in keys:
    response = client.get_object(
        Bucket=bucket_name,
        Key=key
    )
    json_stream = response['Body'].read().decode()
    json_stream_array = []
    json_stream_array = json_stream.split('\n')
    
    for i_json_stream_array in json_stream_array:
        records.append(json.loads(i_json_stream_array))

distances = []        
for record in records:
    for data in record['records']:
        distances.append(data)

#distances
new_records = []
new_records = sorted(distances, key=lambda tmp_records: tmp_records['time'])

# 角速度の計算
duration = new_records[len(new_records) - 1]['time'] - new_records[0]['time']
omega = 360 / (duration / 1000)

print("Omega: ", omega)
print("Start Time: ", new_records[0]['time'])
print("End Time: ", new_records[len(new_records) - 1]['time'])
print("Duration: ", duration / 1000)


# Calculate Position
radius = 50 / 2
for i_records in new_records:
    degree = (i_records['time'] - new_records[0]['time']) / 1000 * omega
    target_pos_x = (radius + i_records['distance']) * math.cos(-math.radians(degree))
    target_pos_y = (radius + i_records['distance']) * math.sin(-math.radians(degree))
    bpy.ops.mesh.primitive_uv_sphere_add(location=(target_pos_x,target_pos_y,0), scale=(0.05,0.05,0.05))