Em um
artigo anterior, prometi mostrar um script que retira o vídeo da câmera e, apesar de um certo período de tempo desde então, as promessas precisam ser cumpridas. Então, eu estou fazendo isso.
Aconteceu que, com a assincronia no mundo da web de servidores, tudo está associado, mas não o PHP.
Bem, porque, você sabe, esse modelo moribundo, a memória vaza e, de fato, não há nada fora da caixa no PHP, exceto
stream_select () e
stream_set_blocking () .
Em algum lugar do PECL, existe algum tipo de
libuv , que, em princípio, é apenas um invólucro para as funções originais da biblioteca original, portanto, usá-lo como está é um desafio. Enfim, quem em sã consciência fará isso?
Mas se pararmos de viver no mundo do PHP4 e voltarmos um pouco para as realidades modernas, veremos que as coisas mudaram um pouco nos últimos anos. Temos ferramentas interessantes como
ReactPHP e
AmPHP , cujos componentes cobrem bem a funcionalidade
Node.js. , e a presença de geradores nos permite escrever código assíncrono em um estilo conveniente, como
async / wait , evitando todos esses retornos intermináveis em retornos de chamada e cadeias de quilômetros
. ) .then (). then () .
É por isso que agora, parece-me, praticamente não existem tarefas do mundo do Node.js que o PHP não conseguiu resolver. Mas, se ainda existem alguns, tudo depende apenas da presença de algumas bibliotecas separadas e não da falta de oportunidades.
Pode surpreender as pessoas saber que a biblioteca padrão do PHP já tem tudo o que precisamos para criar aplicativos orientados a eventos e sem bloqueio. Somente atingimos os limites da funcionalidade nativa do PHP nessa área quando solicitamos que ele faça uma pesquisa de milhares de descritores de arquivos para atividades de IO ao mesmo tempo. Mesmo nesse caso, porém, a falha não ocorre no PHP, mas na chamada select () do sistema subjacente, que é linear na degradação do desempenho à medida que a carga aumenta.
amphp.org/amp/event-loop
Obviamente, a questão de usar tudo isso na produção ainda não está completamente fechada, mas se você não fizer coisas obviamente erradas, que qualquer outro ambiente de tempo de execução assíncrono não o perdoará, tudo ficará bem.
DVRIP
Vamos considerar o protocolo, supostamente inventado pelos chineses, que é usado para comunicar software com câmeras de vigilância.
Tudo funciona em cima do TCP, chamado DVRIP (às vezes Sofia), e quando precisei, encontrei apenas duas bibliotecas mais ou menos sãs, uma das quais descrevia geralmente apenas conectar-se à câmera e trocar algumas mensagens, mas tinha descrição do cabeçalho do pacote que me ajudou muito.
O segundo foi encontrado um pouco mais tarde, no momento da exacerbação da minha síndrome não inventada aqui, e funcionou um pouco diferente do que eu gostaria.
Em geral, armado com um sniffer e uma descrição do cabeçalho do pacote, esbocei brevemente um certo script que é executado em uma forma em algum lugar do meu servidor e será apresentado em uma certa prova de conceito neste artigo.
O protocolo em si não é complexo e o pacote se parece com isso:
Apenas transmitiremos json, mas receberemos ambos.
A resposta (se for um pacote com json) geralmente contém um campo
Ret indicando o sucesso de nossa solicitação. Respostas que contenham um código
Ret diferente de
100 ou
515 serão consideradas deliberadamente erradas.
Armado com esse conhecimento básico, vamos nos conectar à câmera e entender.
Precisamos de duas bibliotecas -
amphp / socket , que
abrirão quase tudo o que precisamos, e
evenement / evenement , que estão disponíveis no packagist e não exigem extensões.
Trabalhar com TCP no Amp é mais ou menos assim:
Loop::run(function () { $socket = yield connect("tcp://{$addr}:{$port}"); yield $socket->write('Hello'); while ($chunk = yield $socket->read()) { echo $chunk; } $socket->close(); });
O DVRIP geralmente roda na porta 34567, portanto, no nosso caso, será algo como isto:
$socket = yield connect('tcp://192.168.0.200:34567');
Em seguida, efetuamos login enviando nosso hash de nome de usuário e senha para a câmera, calculado da seguinte forma:
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))); }
Em caso de autorização bem-sucedida, solicitamos um vídeo e, se a câmera puder nos fornecer, começamos a transmiti-lo.
Cada tipo de pacote possui seu próprio msgId, para que possamos entender qual solicitação recebemos a resposta. Em geral, não é um trabalho empoeirado.
Por conveniência, vamos primeiro descrever a classe do nosso pacote:
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 tente enviar uma solicitação de autorização
yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.login'], [ 'EncryptType' => 'MD5', 'LoginType' => 'DVRIP-Web', 'PassWord' => sofiaHash('admin'), 'UserName' => 'admin', ]));
Sim, ótimo. Conectamos e enviamos uma mensagem, mas nem sabemos se a câmera estava correta e reagimos a ela pelo menos de alguma forma. De alguma forma, precisamos ler a resposta (vamos fazê-lo de forma assíncrona), selecionar pacotes específicos e decodificá-los.
Para separar, de alguma forma, a lógica do processamento de pacotes em resposta, decidi usar a biblioteca
evenement e desenhei uma classe baseada nela, que descartaria pacotes já 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; } }
e também adicionamos alguns métodos estáticos à nossa classe Packet, que, de fato, extrai e decodifica pacotes do fluxo de dados
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; }
Agora, através do uso de
Evenement , podemos descrever nossa lógica na forma
$reader->on($packetType, $handler);
E então eu recebi essa cadeia, onde adicionei outro processamento Keep Alive e 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());
Não é preciso dizer que, em vez de stdout, podemos redirecionar o fluxo de vídeo para onde quisermos.
Roteiro 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()); });
Agora podemos tentar gravar um pedaço de vídeo dessa maneira
$ php recorder.php > output.h264
Adicionamos um manipulador SIGINT; portanto, quando você se cansar, pressione Ctrl + C e o script interromperá a transferência de vídeo e fechará a conexão normalmente.
E podemos, por exemplo, redirecionar o fluxo para ffmpeg e transcodificar em tempo real.
$ 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
E você pode criar cadeias frenéticas do formulário
$ php recorder.php | ffmpeg ... | curl -d @- ...
lançando tudo isso com algum tipo de script que será reiniciado a cada hora, para que possamos obter partes suaves com duração de 1 hora, em movimento, sempre que precisarmos.
Porque
Estou inclinado a acreditar que este artigo irá levantar a questão "Por quê?" Existem soluções prontas para registro de câmeras, as câmeras em geral podem fazer muitas coisas por si mesmas, como objetaram corretamente nos comentários ao artigo anterior e, em geral, isso poderia ser muito mais fácil.
Só que eu tinha tempo livre para fazer em algum lugar, um desejo de seguir meu próprio caminho e um desejo incansável por experimentos.
E isso funciona.
Obviamente, o script apresentado no artigo requer algumas modificações, e você não deve usá-lo desta forma - repito, isso é prova de conceito. Em geral, eu o aconselho a usar produtos caseiros até o joelho para garantir a segurança de sua casa.
No meu caso, a confiabilidade não é tão crítica, portanto essa opção funcionará. Um observador de scripts está girando sobre esse script, que transfere o fluxo para ffmpeg, gravando as peças por hora e enviando-as para o VK, reiniciando o script no caso de uma desconexão ou de circunstâncias imprevistas.
Em geral, tudo isso poderia ter sido evitado simplesmente retirando o vídeo sobre RTSP sobre TCP, como sugeri no final do artigo anterior, mas o ffmpeg gostava de parar estupidamente em um momento aleatório no tempo e começar a responder apenas ao SIGKILL.
Comecei a suspeitar que tudo isso acontecia porque a observação de scripts dos pais era executada pela coroa e recebia uma prioridade mais baixa, mas aumentar a prioridade reduzia apenas a frequência do problema.
No caso do script acima, tudo funciona muito mais estável e ainda não apresentou nenhuma queixa.