La jornada de trabajo a distancia del viernes ya estaba llegando a su fin cuando llamaron a la puerta para anunciar la instalación de un nuevo intercomunicador. Al enterarme de que el nuevo intercomunicador tiene una aplicación móvil que te permite contestar llamadas sin estar en casa, me interesé y de inmediato la descargué a mi teléfono. Después de iniciar sesión, descubrí una característica interesante de esta aplicación: incluso sin una llamada activa a mi apartamento, podía mirar por la cámara del intercomunicador y abrir la puerta en cualquier momento. "¡Sí, esto es ARI en línea en la puerta de entrada!" - hizo clic en mi cabeza. El destino del próximo fin de semana estaba sellado.
Video de demostración al final del artículo.
Descargo de responsabilidad
. , - — .
API
, , . - — , . lkit — , http(s) Android .
— Android- Certificate authority , . , Android 7 .
root , Android, Android Studio. ADB , Certificate pinning .
, — , .
:
: POST
/rest/v1/places/{place_id}/accesscontrols/{control_id}/actions
JSON-{"name": "accessControlOpen"}
() : GET
/rest/v1/places/{place_id}/accesscontrols/{control_id}/videosnapshots
: GET
/rest/v1/forpost/cameras/{camera_id}/video?LightStream=0
HTTP Authorization — , . Advanced REST Client, , Authorization API , , .
Python requests
, :
HEADERS = {"Authorization": "Bearer ###"}
ACTION_URL = "https://###.ru/rest/v1/places/###/accesscontrols/###/"
VIDEO_URL = "https://###.ru/rest/v1/forpost/cameras/###/video?LightStream=0"
def get_image():
result = requests.get(f'{ACTION_URL}/videosnapshots', headers=HEADERS)
if result.status_code != 200:
logging.error(f"Failed to get an image with status code {result.status_code}")
return None
logging.warning(f"Image received successfully in {result.elapsed.total_seconds()}sec")
return result.content
def open_door():
result = requests.post(
f'{ACTION_URL}/actions', headers=HEADERS, json={"name": "accessControlOpen"})
if result.status_code != 200:
logging.error(f"Failed to open the door with status code {result.status_code}")
return False
logging.warning(f"Door opened successfully in {result.elapsed.total_seconds()}sec")
return True
def get_videostream_link():
result = requests.get(VIDEO_URL, headers=HEADERS)
if result.status_code != 200:
logging.error(f"Failed to get stream link with status code {result.status_code}")
return False
logging.warning(f"Stream link received successfully in {result.elapsed.total_seconds()}sec")
return result.json()['data']['URL']
, — Intel(R) Xeon(R) CPU E5-2650L v3 @ 1.80GHz
, 1GB 0 GPU. , , .
, . OpenVINO Toolkit — Intel, CPU.
Interactive Face Recognition Demo — , . , - 2020.3, pip 2021.1. OpenVINO .
, . ( ), , , :
class ImageProcessor:
def __init__(self):
self.frame_processor = FrameProcessor()
def process(self, image):
detections = self.frame_processor.process(image)
labels = []
for roi, landmarks, identity in zip(*detections):
label = self.frame_processor.face_identifier.get_identity_label(
identity.id)
labels.append(label)
return labels
. , get_image()
.
100 runs on an image with known face:
Total time: 7.356s
Time per frame: 0.007s
FPS: 135.944
100 runs on an image without faces:
Total time: 2.985s
Time per frame: 0.003s
FPS: 334.962
, .
1 FPS:
, , . , MVP get_image()
.
class ImageProcessor:
# <...>
def process_single_image(self, image):
nparr = np.fromstring(image, np.uint8)
img_np = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
labels = self.process(img_np)
return labels
def snapshot_based_intercom_id():
processor = ImageProcessor()
last_open_door_time = time.time()
while True:
start_time = time.time()
image = get_image()
result = processor.process_single_image(image)
logging.info(f'{result} in {time.time() - start_time}s')
# Successfull detections are "face{N}"
if any(['face' in res for res in result]):
if start_time - last_open_door_time > 5:
open_door()
with open(f'images/{start_time}_OK.jfif', 'wb') as f:
f.write(image)
last_open_door_time = start_time
, , . , .. .
! , . , — , .. API . , 0.7 0.6 , .
30 FPS:
:
vcap = cv2.VideoCapture(link) success, frame = vcap.read()
, 30 FPS. : read()
. , , , . , , 30 — , .
: vcap.set(CV_CAP_PROP_BUFFERSIZE, 0);
. , OpenCV 3.4, - , . , StackOverflow — , ( , ).
ImageProcessor
3 :
class CameraBufferCleanerThread(threading.Thread):
def __init__(self, camera, name='camera-buffer-cleaner-thread'):
self.camera = camera
self.last_frame = None
self.finished = False
super(CameraBufferCleanerThread, self).__init__(name=name)
self.start()
def run(self):
while not self.finished:
ret, self.last_frame = self.camera.read()
def __enter__(self): return self
def __exit__(self, type, value, traceback):
self.finished = True
self.join()
class ImageProcessor:
# <...>
def process_stream(self, link):
vcap = cv2.VideoCapture(link)
interval = 0.3 # ~3 FPS
with CameraBufferCleanerThread(vcap) as cam_cleaner:
while True:
frame = cam_cleaner.last_frame
if frame is not None:
yield (self.process(frame), frame)
else:
yield (None, None)
time.sleep(interval)
snapshot_based_intercom_id
:
def stream_based_intercom_id():
processor = ImageProcessor()
link = get_videostream_link()
# To notify about delays
last_time = time.time()
last_open_door_time = time.time()
for result, np_image in processor.process_stream(link):
current_time = time.time()
delta_time = current_time - last_time
if delta_time < 1:
logging.info(f'{result} in {delta_time}')
else:
logging.warning(f'{result} in {delta_time}')
last_time = current_time
if result is None:
continue
if any(['face' in res for res in result]):
if current_time - last_open_door_time > 5:
logging.warning(
f'Hey, I know you - {result[0]}! Opening the door...')
last_open_door_time = current_time
open_door()
cv2.imwrite(f'images/{current_time}_OK.jpg', np_image)
— , .
Telegram
/. .
python-telegram-bot
, callback / .
class TelegramInterface:
def __init__(self, login_whitelist, state_callback):
self.state_callback = state_callback
self.login_whitelist = login_whitelist
self.updater = Updater(
token = "###", use_context = True)
self.run()
def run(self):
dispatcher = self.updater.dispatcher
dispatcher.add_handler(CommandHandler("start", self.start))
dispatcher.add_handler(CommandHandler("run", self.run_intercom))
dispatcher.add_handler(CommandHandler("stop", self.stop_intercom))
self.updater.start_polling()
def run_intercom(self, update: Update, context: CallbackContext):
user = update.message.from_user
update.message.reply_text(
self.state_callback(True) if user.username in self.login_whitelist else 'not allowed',
reply_to_message_id=update.message.message_id)
def stop_intercom(self, update: Update, context: CallbackContext):
user = update.message.from_user
update.message.reply_text(
self.state_callback(False) if user.username in self.login_whitelist else 'not allowed',
reply_to_message_id=update.message.message_id)
def start(self, update: Update, context: CallbackContext) -> None:
update.message.reply_text('Hi!')
class TelegramBotThreadWrapper(threading.Thread):
def __init__(self, state_callback, name='telegram-bot-wrapper'):
self.whitelist = ["###", "###"]
self.state_callback = state_callback
super(TelegramBotThreadWrapper, self).__init__(name=name)
self.start()
def run(self):
self.bot = TelegramInterface(self.whitelist, self.state_callback)
intercom_id
, :
def stream_based_intercom_id_with_telegram():
processor = ImageProcessor()
loop_state_lock = threading.Lock()
loop_should_run = False
loop_should_change_state_cv = threading.Condition(loop_state_lock)
is_loop_finished = True
loop_changed_state_cv = threading.Condition(loop_state_lock)
def stream_processing_loop():
nonlocal loop_should_run
nonlocal loop_should_change_state_cv
nonlocal is_loop_finished
nonlocal loop_changed_state_cv
while True:
with loop_should_change_state_cv:
loop_should_change_state_cv.wait_for(lambda: loop_should_run)
is_loop_finished = False
loop_changed_state_cv.notify_all()
logging.warning(f'Loop is started')
link = get_videostream_link()
last_time = time.time()
last_open_door_time = time.time()
for result, np_image in processor.process_stream(link):
with loop_should_change_state_cv:
if not loop_should_run:
is_loop_finished = True
loop_changed_state_cv.notify_all()
logging.warning(f'Loop is stopped')
break
current_time = time.time()
delta_time = current_time - last_time
if delta_time < 1:
logging.info(f'{result} in {delta_time}')
else:
logging.warning(f'{result} in {delta_time}')
last_time = current_time
if result is None:
continue
if any(['face' in res for res in result]):
if current_time - last_open_door_time > 5:
logging.warning(f'Hey, I know you - {result[0]}! Opening the door...')
last_open_door_time = current_time
open_door()
cv2.imwrite(f'images/{current_time}_OK.jpg', np_image)
def state_callback(is_running):
nonlocal loop_should_run
nonlocal loop_should_change_state_cv
nonlocal is_loop_finished
nonlocal loop_changed_state_cv
with loop_should_change_state_cv:
if is_running == loop_should_run:
return "Intercom service state is not changed"
loop_should_run = is_running
if loop_should_run:
loop_should_change_state_cv.notify_all()
loop_changed_state_cv.wait_for(lambda: not is_loop_finished)
return "Intercom service is up"
else:
loop_changed_state_cv.wait_for(lambda: is_loop_finished)
return "Intercom service is down"
telegram_bot = TelegramBotThreadWrapper(state_callback)
logging.warning("Bot is ready")
stream_processing_loop()
:
A pesar de las posibilidades que brinda la tecnología de intercomunicación inteligente a los residentes, cientos (¿miles?) De puertas de entrada con cámaras y micrófonos (¡sí, hay audio en la transmisión de video recibida al azar!), Lo que abre nuevas oportunidades para violaciones de privacidad.
Preferiría que el acceso a la transmisión de video se proporcionara solo en el momento de una llamada al apartamento y la grabación en curso de tres días, posicionada como un medio para revelar violaciones, no se almacenara en los servidores de la compañía, sino directamente en el intercomunicador. , con la posibilidad de acceder a él a pedido. O no del todo.