En un
artículo anterior, prometí mostrar un guión que saca el video de la cámara y, aunque ha pasado un cierto tiempo desde entonces, las promesas deben cumplirse. Entonces lo estoy haciendo.
Dio la casualidad de que con la asincronía en el mundo de los servidores web todo está asociado, pero no PHP.
Bueno, porque, ya sabes, este modelo moribundo, la memoria pierde, y de hecho no hay nada fuera de la caja en PHP excepto
stream_select () y
stream_set_blocking () .
En algún lugar allí, en PECL, hay algún tipo de
libuv , que, en principio, es solo un contenedor para las funciones originales de la biblioteca original, por lo que usarlo tal como es le brinda algunos desafíos. De todos modos, ¿quién en su sano juicio hará esto?
Pero si dejamos de vivir en el mundo de PHP4 y volvemos un poco a las realidades modernas, veremos que las cosas han cambiado un poco en los últimos años. Tenemos herramientas tan interesantes como
ReactPHP y
AmPHP , cuyos componentes cubren bien la funcionalidad de Node.js, y la presencia de generadores nos permite escribir código asíncrono en un estilo conveniente, como
async / wait , evitando todas estas devoluciones de llamadas sin fin en devoluciones de llamada y cadenas de kilómetros
. ) .then (). then () .
Entonces, por eso ahora, me parece, prácticamente no hay tareas del mundo de Node.js que PHP no pueda resolver. Pero si todavía hay algunas, entonces todo depende solo de la presencia de algunas bibliotecas separadas, y no de la falta de oportunidades como tal.
Puede sorprender a la gente saber que la biblioteca estándar de PHP ya tiene todo lo que necesitamos para escribir aplicaciones controladas por eventos y sin bloqueo. Solo alcanzamos los límites de la funcionalidad nativa de PHP en esta área cuando le pedimos que sondee miles de descriptores de archivos para la actividad de E / S al mismo tiempo. Sin embargo, incluso en este caso, el error no está en PHP sino en la llamada select () del sistema subyacente, que es lineal en su degradación del rendimiento a medida que aumenta la carga.
amphp.org/amp/event-loop
Por supuesto, la cuestión de usar todo esto en la producción aún no está completamente cerrada, pero si no sabe cosas erróneas que cualquier otro entorno de tiempo de ejecución asincrónico no le perdonará, entonces todo estará bien.
DVRIP
Consideraremos el protocolo, supuestamente inventado por los chinos, que se utiliza para comunicar software con cámaras de vigilancia.
Todo funciona sobre TCP, llamado DVRIP (a veces Sofía), y cuando lo necesité, encontré solo dos bibliotecas más o menos sanas, una de las cuales describía que generalmente solo se conectaba a la cámara e intercambiaba un par de mensajes, pero tenía Descripción del encabezado del paquete que me ayudó mucho.
El segundo fue encontrado un poco más tarde, en el momento de la exacerbación de mi síndrome no inventado aquí, y funcionó de manera un poco diferente de lo que me gustaría.
En general, armado con un sniffer y una descripción del encabezado del paquete, bosquejé brevemente un cierto script que está girando en algún lugar de mi servidor en algún lugar, y se presentará en una cierta prueba de concepto en este artículo.
El protocolo en sí no es complejo y el paquete se ve así:
Solo transmitiremos json, pero recibiremos ambos.
La respuesta (si es un paquete con json) generalmente contiene un campo
Ret que indica el éxito de nuestra solicitud. Las respuestas que contengan un código
Ret que no sea igual a
100 o
515 se considerarán deliberadamente erróneas.
Armados con este conocimiento básico, conectemos a la cámara y comprendamos.
Necesitamos dos bibliotecas:
amphp / socket , que
extraerá casi todo lo que necesitamos, y
evenement / evenement , que están disponibles en packagist y no requieren ninguna extensión.
Trabajar con TCP en Amp se parece a esto:
Loop::run(function () { $socket = yield connect("tcp://{$addr}:{$port}"); yield $socket->write('Hello'); while ($chunk = yield $socket->read()) { echo $chunk; } $socket->close(); });
DVRIP generalmente se ejecuta en el puerto 34567, por lo que en nuestro caso será algo como esto:
$socket = yield connect('tcp://192.168.0.200:34567');
Luego, iniciamos sesión enviando a la cámara nuestro hash de nombre de usuario y contraseña, que se calcula de la siguiente manera:
function sofiaHash(string $password) : string { $md5 = md5($password, true); return implode('', array_map(function ($i) use ($md5) { $c = (ord($md5[2 * $i]) + ord($md5[2 * $i + 1])) % 62; $c += $c > 9 ? ($c > 35 ? 61 : 55) : 48; return chr($c); }, range(0, 7))); }
En caso de autorización exitosa, solicitamos un video y, si la cámara nos lo puede dar, entonces comenzamos a transmitirlo.
Cada tipo de paquete tiene su propio msgId, por lo que podemos entender qué solicitud recibimos la respuesta. En general, no es un trabajo polvoriento.
Para mayor comodidad, describamos primero la clase de nuestro paquete:
class Packet { public const SUCCESS_CODES = [100, 515]; public const MESSAGE_CODES = [ 'packet.request.login' => 1000, 'packet.request.keepAlive' => 1006, 'packet.request.claim' => 1413, 'packet.response.login' => 1001, 'packet.response.keepAlive' => 1007, 'packet.response.claim' => 1414, 'packet.videoControl' => 1410, 'packet.binary' => 1412, ]; protected int $head = 255; protected int $version = 0; protected int $sessionId; protected int $sequence; protected int $msgId; protected ?string $rawData; protected ?array $data; public function __construct(int $msgId, ?array $data = null, int $sequence = 0, int $sessionId = 0) { $this->msgId = $msgId; $this->data = $data; $this->sequence = $sequence; $this->sessionId = $sessionId; } public function __toString() : string { if ($this->data !== null) { $this->rawData = json_encode($this->data, JSON_THROW_ON_ERROR) . "\x0a\x00"; } return pack('CCx2IIx2SI', $this->head, $this->version, $this->sessionId, $this->sequence, $this->msgId, strlen($this->rawData)) . $this->rawData; } public function getData() : ?array { return $this->data; } public function getRawData() : ?string { return $this->rawData; } public function getSession() : int { return $this->sessionId; } public function getSequence() : int { return $this->sequence; } public function getMessageType() : string { $types = array_flip(static::MESSAGE_CODES); if (!isset($types[$this->msgId])) { throw new \Exception('Unknown message type: ' . $this->msgId); } return $types[$this->msgId]; } }
e intente enviar una solicitud de autorización
yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.login'], [ 'EncryptType' => 'MD5', 'LoginType' => 'DVRIP-Web', 'PassWord' => sofiaHash('admin'), 'UserName' => 'admin', ]));
Si genial Nos conectamos y enviamos un mensaje, pero ni siquiera sabemos si la cámara era correcta y reaccionó al menos de alguna manera. Necesitamos leer de alguna manera la respuesta (hagámoslo de forma asincrónica), seleccionar paquetes específicos y decodificarlos.
Para separar de alguna manera la lógica de procesamiento de los paquetes que vienen en respuesta, decidí usar la biblioteca
evenement y esbocé una clase basada en ella que arrojará paquetes ya decodificados
class Reader extends EventEmitter { protected InputStream $socket; public function __construct(InputStream $socket) { $this->socket = $socket; } public function start() : \Generator { $buffer = ''; while (!$this->socket->isClosed() && $result = yield Packet::read($this->socket, $buffer)) { [$packet, $chunk] = $result; $buffer = $chunk; $chunk = null; $this->emit('packet', [$packet]); $this->emit($packet->getMessageType(), [$packet]); } $buffer = null; } }
y también agregamos un par de métodos estáticos a nuestra clase Packet, que, de hecho, extraerá y decodificará paquetes del flujo de datos
public static function read(InputStream $socket, string $buffer = '') : Promise { return call(function () use ($socket, $buffer) { $header = null; do { if ($header === null && strlen($buffer) > 20) { $header = unpack('Chead/Cversion/x2/IsessionId/Isequence/x2/SmsgId/Ilen', substr($buffer, 0, 20)); $buffer = substr($buffer, 20); } if ($header !== null && strlen($buffer) >= (int) $header['len']) { return [ static::fromRaw($header, (int) $header['len'] > 0 ? substr($buffer, 0, (int) $header['len']) : null), substr($buffer, (int) $header['len']) ]; } $buffer .= yield $socket->read(); } while (!$socket->isClosed()); }); } protected static function fromRaw(array $header, ?string $rawData) : self { $packet = new static($header['msgId'], null, $header['sequence'], $header['sessionId']); $packet->rawData = $rawData; $packet->data = (int) $packet->msgId === static::MESSAGE_CODES['packet.binary'] || $rawData === null ? null : json_decode(substr($rawData, 0, -2), true, 512, JSON_THROW_ON_ERROR); return $packet; }
Ahora, mediante el uso de
Evenement , podemos describir nuestra lógica en la forma
$reader->on($packetType, $handler);
Y así obtuve esta cadena, donde agregué otro procesamiento Keep Alive y SIGINT / SIGTERM:
$reader = new Reader($socket = yield connect('tcp://10.0.5.100:49152')); $reader->on('packet.response.login', asyncCoroutine(function (Packet $packet) use ($socket) { if (empty($packet->getData()['Ret']) || !in_array($packet->getData()['Ret'], Packet::SUCCESS_CODES)) { throw new \Exception('Wrong login data'); } yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.claim'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Claim', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], 1, $packet->getSession())); })); $reader->once('packet.response.login', function (Packet $packet) use ($socket) { Loop::unreference(Loop::repeat($packet->getData()['AliveInterval'] * 1000, function ($watcherId) use ($socket, $packet) { if ($socket->getResource() === null) { return Loop::cancel($watcherId); } yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.keepAlive'], [ 'Name' => 'KeepAlive', 'SessionID' => $packet->getData()['SessionID'] ], 0, $packet->getSession())); })); }); $reader->on('packet.response.claim', asyncCoroutine(function (Packet $packet) use ($socket) { yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.videoControl'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Start', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], $packet->getSequence() + 1, $packet->getSession())); })); $emitter = new Emitter(); $reader->on('packet.binary', function ($packet) use ($emitter) { $emitter->emit($packet->getRawData()); }); $reader->once('packet.binary', function (Packet $packet) use ($socket) { $signalHandler = function () use ($socket, $packet) { yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.videoControl'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Stop', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], 0, $packet->getSession())); $socket->close(); }; Loop::unreference(Loop::onSignal(defined('SIGINT') ? SIGINT : 2, $signalHandler, 'SIGINT')); Loop::unreference(Loop::onSignal(defined('SIGTERM') ? SIGTERM : 15, $signalHandler, 'SIGTERM')); }); asyncCall([$reader, 'start']); yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.login'], [ 'EncryptType' => 'MD5', 'LoginType' => 'DVRIP-Web', 'PassWord' => sofiaHash('123qwea'), 'UserName' => 'admin', ])); yield pipe(new IteratorStream($emitter->iterate()), getStdout());
No hace falta decir que en lugar de stdout, podemos redirigir la transmisión de video donde queramos.
Guión final <?php ini_set('display_errors', 'stderr'); use Amp\Emitter; use Amp\Loop; use Amp\Promise; use Amp\ByteStream\InputStream; use Amp\ByteStream\IteratorStream; use Evenement\EventEmitter; use function Amp\call; use function Amp\asyncCall; use function Amp\asyncCoroutine; use function Amp\ByteStream\getStderr; use function Amp\ByteStream\getStdout; use function Amp\ByteStream\pipe; use function Amp\Socket\connect; require 'vendor/autoload.php'; function sofiaHash(string $password) : string { $md5 = md5($password, true); return implode('', array_map(function ($i) use ($md5) { $c = (ord($md5[2 * $i]) + ord($md5[2 * $i + 1])) % 62; $c += $c > 9 ? ($c > 35 ? 61 : 55) : 48; return chr($c); }, range(0, 7))); } class Reader extends EventEmitter { protected InputStream $socket; public function __construct(InputStream $socket) { $this->socket = $socket; } public function start() : \Generator { $buffer = ''; while (!$this->socket->isClosed() && $result = yield Packet::read($this->socket, $buffer)) { [$packet, $chunk] = $result; $buffer = $chunk; $chunk = null; $this->emit('packet', [$packet]); $this->emit($packet->getMessageType(), [$packet]); } $buffer = null; } } class Packet { public const SUCCESS_CODES = [100, 515]; public const MESSAGE_CODES = [ 'packet.request.login' => 1000, 'packet.request.keepAlive' => 1006, 'packet.request.claim' => 1413, 'packet.response.login' => 1001, 'packet.response.keepAlive' => 1007, 'packet.response.claim' => 1414, 'packet.videoControl' => 1410, 'packet.binary' => 1412, ]; public static function read(InputStream $socket, string $buffer = '') : Promise { return call(function () use ($socket, $buffer) { $header = null; do { if ($header === null && strlen($buffer) > 20) { $header = unpack('Chead/Cversion/x2/IsessionId/Isequence/x2/SmsgId/Ilen', substr($buffer, 0, 20)); $buffer = substr($buffer, 20); } if ($header !== null && strlen($buffer) >= (int) $header['len']) { return [ static::fromRaw($header, (int) $header['len'] > 0 ? substr($buffer, 0, (int) $header['len']) : null), substr($buffer, (int) $header['len']) ]; } $buffer .= yield $socket->read(); } while (!$socket->isClosed()); }); } protected static function fromRaw(array $header, ?string $rawData) : self { $packet = new static($header['msgId'], null, $header['sequence'], $header['sessionId']); $packet->rawData = $rawData; $packet->data = (int) $packet->msgId === static::MESSAGE_CODES['packet.binary'] || $rawData === null ? null : json_decode(substr($rawData, 0, -2), true, 512, JSON_THROW_ON_ERROR); return $packet; } protected int $head = 255; protected int $version = 0; protected int $sessionId; protected int $sequence; protected int $msgId; protected ?string $rawData; protected ?array $data; public function __construct(int $msgId, ?array $data = null, int $sequence = 0, int $sessionId = 0) { $this->msgId = $msgId; $this->data = $data; $this->sequence = $sequence; $this->sessionId = $sessionId; } public function __toString() : string { if ($this->data !== null) { $this->rawData = json_encode($this->data, JSON_THROW_ON_ERROR) . "\x0a\x00"; } return pack('CCx2IIx2SI', $this->head, $this->version, $this->sessionId, $this->sequence, $this->msgId, strlen($this->rawData)) . $this->rawData; } public function getData() : ?array { return $this->data; } public function getRawData() : ?string { return $this->rawData; } public function getSession() : int { return $this->sessionId; } public function getSequence() : int { return $this->sequence; } public function getMessageType() : string { $types = array_flip(static::MESSAGE_CODES); if (!isset($types[$this->msgId])) { throw new \Exception('Unknown message type: ' . $this->msgId); } return $types[$this->msgId]; } } Loop::run(function () : \Generator { $reader = new Reader($socket = yield connect('tcp://10.0.5.100:49152')); $reader->on('packet.response.login', asyncCoroutine(function (Packet $packet) use ($socket) { if (empty($packet->getData()['Ret']) || !in_array($packet->getData()['Ret'], Packet::SUCCESS_CODES)) { throw new \Exception('Wrong login data'); } yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.claim'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Claim', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], 1, $packet->getSession())); })); $reader->once('packet.response.login', function (Packet $packet) use ($socket) { Loop::unreference(Loop::repeat($packet->getData()['AliveInterval'] * 1000, function ($watcherId) use ($socket, $packet) { if ($socket->getResource() === null) { return Loop::cancel($watcherId); } yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.keepAlive'], [ 'Name' => 'KeepAlive', 'SessionID' => $packet->getData()['SessionID'] ], 0, $packet->getSession())); })); }); $reader->on('packet.response.claim', asyncCoroutine(function (Packet $packet) use ($socket) { yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.videoControl'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Start', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], $packet->getSequence() + 1, $packet->getSession())); })); $emitter = new Emitter(); $reader->on('packet.binary', function ($packet) use ($emitter) { $emitter->emit($packet->getRawData()); }); $reader->once('packet.binary', function (Packet $packet) use ($socket) { $signalHandler = function () use ($socket, $packet) { yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.videoControl'], [ 'Name' => 'OPMonitor', 'OPMonitor' => ['Action' => 'Stop', 'Parameter' => ['Channel' => 0, 'CombinMode' => 'NONE', 'StreamType' => 'Main', 'TransMode' => 'TCP']], ], 0, $packet->getSession())); $socket->close(); }; Loop::unreference(Loop::onSignal(defined('SIGINT') ? SIGINT : 2, $signalHandler, 'SIGINT')); Loop::unreference(Loop::onSignal(defined('SIGTERM') ? SIGTERM : 15, $signalHandler, 'SIGTERM')); }); asyncCall([$reader, 'start']); yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.login'], [ 'EncryptType' => 'MD5', 'LoginType' => 'DVRIP-Web', 'PassWord' => sofiaHash('123qwea'), 'UserName' => 'admin', ])); yield pipe(new IteratorStream($emitter->iterate()), getStdout()); });
Ahora podemos intentar grabar un video de esta manera
$ php recorder.php > output.h264
Agregamos un controlador SIGINT, así que cuando te canses, solo presiona Ctrl + C y el script detendrá la transferencia de video y cerrará la conexión normalmente.
Y podemos, por ejemplo, redirigir la transmisión a ffmpeg y transcodificar sobre la marcha.
$ php recorder.php | ffmpeg -nostdin -y -hide_banner -loglevel verbose -f h264 -i pipe:0 -pix_fmt yuv420p -c copy -movflags +frag_keyframe+empty_moov+default_base_moof -reset_timestamps 1 -f mpegts output.mpegts
Y puedes hacer cadenas frenéticas de la forma
$ php recorder.php | ffmpeg ... | curl -d @- ...
iniciando todo esto con algún tipo de script que se reiniciará cada hora, por lo que obtenemos piezas suaves que duran 1 hora, sobre la marcha, donde sea que necesitemos.
Por qué
Me inclino a creer que este artículo planteará la pregunta "¿Por qué?" Existen soluciones preparadas para el registro desde cámaras, las cámaras en general pueden hacer muchas cosas por sí mismas, ya que objetaron correctamente en los comentarios al artículo anterior, y en general podría hacerse mucho más fácil.
Es solo que tenía tiempo libre que tenía que hacer en algún lugar, un deseo de hacer mi propio camino y un deseo infatigable de experimentar.
Y funciona
Por supuesto, el guión presentado en el artículo requiere algunas modificaciones, y no debe usarlo de esta forma. Repito, esta es una prueba de concepto. Y en general, le advierto encarecidamente que use productos caseros hasta la rodilla para garantizar la seguridad de su hogar.
En mi caso, la confiabilidad no es tan crítica, por lo que esta opción funcionará. Un observador de guiones está girando sobre este guión, que transfiere la transmisión a ffmpeg, graba piezas por hora y las carga en el VK, reiniciando el guión en caso de desconexión o cualquier circunstancia imprevista.
En general, todo esto podría haberse evitado simplemente sacando el video a través de RTSP a través de TCP, como sugerí al final del artículo anterior, pero a ffmpeg le gustaba estúpidamente detenerse en un momento aleatorio y comenzar a responder solo a SIGKILL.
Comencé a sospechar que todo esto se debía a que la corona guió el reloj de guiones principal y recibió una prioridad más baja, pero aumentar la prioridad solo redujo la frecuencia del problema.
En el caso del script anterior, todo funciona mucho más estable y aún no ha presentado ninguna queja.