ドローンの映像を AWS で簡単に分析して役立てよう !

import socket

SSID = 'Wi-Fi SSID'
PASS = 'Wi-Fi パスワード'

tello_addr = ('192.168.10.1', 8889)

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('', 8889))

# Send command: 'command'
sock.sendto('command'.encode('utf-8'), tello_addr)
response, addr = sock.recvfrom(100)
print(f'{addr}: {response}' % (addr, response))

# Send command: 'ap {ssid} {pass}'
sock.sendto(f'ap {SSID} {PASS}'.encode('utf-8'), tello_addr)
response, addr = sock.recvfrom(100)
print(f'{addr}: {response}' % (addr, response))

sock.close()

import netaddr
import netifaces
import queue
import socket
import subprocess
import threading
import time


class Tello:
    def __init__(self, netmask='255.255.255.0', tello_cmd_port=8889):
        self.cmd_port = tello_cmd_port

        # Queue for received responses from Tello
        self.q = queue.Queue()

        # AF_INET (IPv4), SOCK_DGRAM (UDP)
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind(('', tello_cmd_port))

        # Receiver for responses from Tello
        self.recv_thread = threading.Thread(target=self._udp_receiver)
        self.recv_thread.daemon = True
        self.recv_thread.start()

        # Search for Tello
        self.ip_addr = self._find_tello(self._get_subnets(netmask))
        print(f'Found Tello: {self.ip_addr}')
        
        
    def __del__(self):
        self.sock.close()


    def _udp_receiver(self):
        """
        Puts all data received on self.cmd_port into the queue
        """
        while True:
            try:
                response, addr = self.sock.recvfrom(1024)
                response = response.decode('utf-8')
                print(f'[UDP RECV] {addr} {response}')
                self.q.put((addr, response))
            except Exception as e:
                print(e)


    def _get_subnets(self, netmask):
        """
        Returns a list of CIDRs of subnets with given netmask
        """
        subnets = []
        ifaces = netifaces.interfaces()
        for iface in ifaces:
            my_addr = netifaces.ifaddresses(iface)[netifaces.AF_INET][0]
            if my_addr['netmask'] == netmask:
                cidr = netaddr.IPNetwork(f'{my_addr["addr"]}/{my_addr["netmask"]}')
                subnets.append(cidr)

        return subnets


    def _find_tello(self, subnets):
        """
        Returns (IP Address, Port) of found Tello.
        """
        addrs = []
        for subnet in subnets:
            # Adds addresses to list except network address and broadcast address
            addrs.extend(str(addr) for addr in subnet[1:-1])
        
        for addr in addrs:
            print(f'Trying {addr} ...')
            self.sock.sendto('command'.encode('utf-8'), (addr, self.cmd_port))

            if not self.q.empty():
                tello_addr, response = self.q.get()
                if response == 'ok':
                    # Tello found !
                    return tello_addr[0]

        raise Exception('No Tello found.')


    def send_cmd(self, cmd, timeout=3, attempts=2):
        """
        Sends command (cmd) to Tello (ip_addr).
        Returns response from Tello.
        """
        print(f'Sending command "{cmd}" ...')

        response = None
        for _ in range(attempts):
            self.sock.sendto(cmd.encode('utf-8'), (self.ip_addr, self.cmd_port))
            try:
                _, response = self.q.get(timeout=timeout)
                break
            except queue.Empty as e:
                print(f'Command "{cmd}" timeout.')

        return response


    def emergency_land(self):
        """
        Sends 'land' command to Tello as soon as a flight command fails.
        """
        self.send_cmd('land')
        self.send_cmd('streamoff')

gst-launch-1.0 udpsrc uri=udp://0.0.0.0:11111 \
    ! h264parse ! video/x-h264,stream-format=avc,alignment=au \
    ! kvssink stream-name="{KVSストリーム名}" storage-size=512 \
    access-key="$AWS_ACCESS_KEY_ID" secret-key="$AWS_SECRET_ACCESS_KEY" aws-region="$AWS_DEFAULT_REGION"

{
  "DataSharingPreference": { "OptIn": true },
  "Input": {
    "KinesisVideoStream": {
      "Arn": "{ビデオストリームのARN}"
    }
  },
  "Name": "tello_stream_processor",
  "Output": {
    "S3Destination": {
      "Bucket": "{S3バケットのARN}",
      "KeyPrefix": ""
    }
  },
  "NotificationChannel": {
    "SNSTopicArn": "{SNSトピックのARN}"
  },
  "RoleArn": "{Rekognition用サービスロールのARN}",
  "Settings": {
    "ConnectedHome": {
      "Labels": [
        "PERSON"
      ],
    "MinConfidence": 80
    }
  }
}

$ aws rekognition create-stream-processor --cli-input-json file://input_for_CreateStreamProcessor.json

{
"StreamProcessorArn": "arn:aws:rekognition:us-west-2:{アカウントID}:streamprocessor/tello_stream_processor"
}

import boto3
from botocore.exceptions import ClientError
import json
from datetime import datetime

s3_client  = boto3.client('s3')
sns_client = boto3.client('sns')

def create_presigned_url(bucket_name, object_name, expiration=3600*24*7):
    # Generate a presigned URL for the S3 object
    try:
        response = s3_client.generate_presigned_url('get_object',
                                                    Params={'Bucket': bucket_name,
                                                            'Key': object_name},
                                                    ExpiresIn=expiration)
    except ClientError as e:
        print(e)
        return None

    return response


def lambda_handler(event, context):
    msg = json.loads(event['Records'][0]['Sns']['Message'])
    if msg['eventNamespace']['type'] != 'LABEL_DETECTED':
        return None
        
    detected = msg['labels'][0]
    email_subject = f'ドローンが {detected["name"]} を検出しました'
    timestamp = detected['videoMapping']['kinesisVideoMapping']['producerTimestamp']
    time_detected = datetime.fromtimestamp(timestamp//1000).strftime('%Y-%m-%d %H:%M:%S')
    email_body  = f'Time:  {time_detected}\n'
    email_body += f'Label: {detected["name"]}\n'
    email_body += f'Confidence: {detected["confidence"]} %\n\n'
    
    # Get image URL
    s3_uri = detected['frameImageUri']
    bucket_name = s3_uri.partition('//')[2].partition('/')[0]
    object_key  = s3_uri.partition('//')[2].partition('/')[2]
    image_url = create_presigned_url(bucket_name, object_key)
    if image_url:
        email_body += f'Image URL: {image_url}\n'
        
    response = sns_client.publish(
        TopicArn='{SNSトピックuser-notificationのARN}',
        Message=email_body,
        Subject=email_subject
    )

    return response

import boto3
import subprocess
import time
from tello_util import *


"""
Sequence of flight commands for Tello.
List of tuple (command, timeout [sec]).
To find available flight commands, see https://dl-cdn.ryzerobotics.com/downloads/Tello/Tello%20SDK%202.0%20User%20Guide.pdf
"""
FLIGHT_CMDS = [
   ('takeoff', 15),
   ('up 50', 10),
   ('forward 160', 20),
   ('cw 180', 20),
   ('forward 100', 20),
   ('land', 10)
]
KVS_STREAM_NAME = 'tello-kvstream'
STREAM_PROCESSOR_NAME = 'tello_stream_processor'
STREAM_PROCESSOR_MAX_DURATION = 60
GSTREAMER_CMD = 'gst-launch-1.0 udpsrc uri=udp://0.0.0.0:11111 '\
    '! h264parse ! video/x-h264,stream-format=avc,alignment=au '\
    f'! kvssink stream-name="{KVS_STREAM_NAME}" storage-size=512 '\
    'access-key="$AWS_ACCESS_KEY_ID" secret-key="$AWS_SECRET_ACCESS_KEY" aws-region="$AWS_DEFAULT_REGION"'


def list_latest_video_fragments(attempts=10):
    """
    Performs ListFragments API for Kinesis Video Streams
    """
    kvs = boto3.client('kinesisvideo')
    endpoint = kvs.get_data_endpoint(
        APIName='LIST_FRAGMENTS',
        StreamName=KVS_STREAM_NAME
    )['DataEndpoint']

    kvam = boto3.client('kinesis-video-archived-media', endpoint_url=endpoint)

    # Some attempts may be needed until media fragments are persisted in KVS
    fragments = []
    for _ in range(attempts):
        timestamp_now = time.time()
        fragments = kvam.list_fragments(
            StreamName=KVS_STREAM_NAME,
            MaxResults=10,
            FragmentSelector={
                'FragmentSelectorType': 'PRODUCER_TIMESTAMP',
                'TimestampRange': {
                    'StartTimestamp': timestamp_now-5,
                    'EndTimestamp': timestamp_now
                }
            }
        )['Fragments']

        if fragments:
            return fragments
        
        print('Failed to list fragments. Retrying ...')
        time.sleep(0.5)
    
    raise Exception('Failed to list fragments.')


def main():
    # (1) Create Tello instance and find Tello's IP address
    tello = Tello()

    # (2) Video stream on (Tello -> PC)
    response = tello.send_cmd('streamon')
    if response != 'ok':
        print(response)
        raise Exception('Failed to start video stream.')
    
    # (2) Start video stream to Kinesis Video Streams using GStreamer
    print('Start GStreamer process')
    devnull = open('/dev/null', 'w')
    gst_proc = subprocess.Popen('exec ' + GSTREAMER_CMD, shell=True, stdout=devnull, stderr=devnull)

    # (3) Get latest video fragments persisted in Kinesis Video Streams
    fragments = list_latest_video_fragments()
    print(f'Got fragments: {fragments}')
    fragment_numbers = [f['FragmentNumber'] for f in fragments]
    latest_fragment_number = max(fragment_numbers)

    # (4) Start stream processor for Rekognition Streaming Video Events
    print('Start stream processor')
    rekognition = boto3.client('rekognition')
    response = rekognition.start_stream_processor(
        Name=STREAM_PROCESSOR_NAME,
        StartSelector={
            'KVSStreamStartSelector': {
                'FragmentNumber': latest_fragment_number
            }
        },
        StopSelector={
            'MaxDurationInSeconds': STREAM_PROCESSOR_MAX_DURATION
        }
    )
    print(response)

    # (5) Start a flight !
    for cmd, timeout in FLIGHT_CMDS:
        response = tello.send_cmd(cmd, timeout=timeout)
        if not response or 'error' in response:
            gst_proc.kill()
            tello.emergency_land()
            raise Exception(f'Failed to execute command "{cmd}". Landing ...')

    # (6) Wait until stream processor stops
    while True:
        response = rekognition.describe_stream_processor(
            Name=STREAM_PROCESSOR_NAME
        )
        status = response["Status"]
        print(f'Stream processor {STREAM_PROCESSOR_NAME}: {status}')
        if status == 'STOPPED':
            break
        elif status == 'FAILED':
            print(f'Stream processor {STREAM_PROCESSOR_NAME} failed.')
            break

        time.sleep(5)

    # (6) Stop video stream to Kinesis Video Streams
    print('Stop Gstreamer process')
    gst_proc.kill()

    # (6) Video stream off (Tello -> PC)
    response = tello.send_cmd('streamoff')
    if response != 'ok':
        raise Exception('Failed to stop video stream.')


if __name__ == '__main__':
    main()