Dans un
article précédent, j'ai promis de montrer un script qui extrait la vidéo de la caméra et, même si un certain temps s'est écoulé depuis, les promesses doivent être tenues. Je le fais donc.
Il se trouve que, avec l'asynchronie dans le monde du serveur Web, tout est associé, mais pas PHP.
Eh bien, parce que, vous savez, ce modèle en train de mourir, des fuites de mémoire, et en effet il n'y a rien hors de la boîte en PHP à l'exception de
stream_select () et
stream_set_blocking () .
Quelque part là-bas, sur PECL, il y a une sorte de
libuv , qui en principe n'est qu'un wrapper pour les fonctions originales de la bibliothèque d'origine, donc l'utiliser tel quel vous pose des défis. Quoi qu'il en soit, qui dans leur bon sens fera cela?
Mais si nous cessons de vivre dans le monde de PHP4 et retournons un peu aux réalités modernes, nous verrons que les choses ont quelque peu changé ces dernières années. Nous avons des outils intéressants tels que
ReactPHP et
AmPHP , dont les composants couvrent bien la fonctionnalité Node.js, et la présence de générateurs nous permet d'écrire du code asynchrone dans un style pratique, comme
async / wait , en évitant tous ces rappels sans fin dans les rappels et les chaînes de kilomètres
. ) .puis (). puis () .
C'est pourquoi maintenant, il me semble, il n'y a pratiquement aucune tâche du monde de Node.js que PHP ne puisse pas résoudre. Mais s'il en existe encore, alors tout ne dépend que de la présence de bibliothèques distinctes, et non du manque d'opportunités en tant que telles.
Cela peut surprendre les gens d'apprendre que la bibliothèque standard PHP possède déjà tout ce dont nous avons besoin pour écrire des applications événementielles et non bloquantes. Nous n'atteignons les limites de la fonctionnalité native de PHP dans ce domaine que lorsque nous lui demandons d'interroger des milliers de descripteurs de fichiers pour l'activité d'E / S en même temps. Même dans ce cas, cependant, la faute ne vient pas de PHP mais de l'appel système select () sous-jacent qui est linéaire dans sa dégradation des performances à mesure que la charge augmente.
amphp.org/amp/event-loop
Bien sûr, la question de l'utilisation de tout cela en production n'est pas encore complètement close, mais si vous ne faites pas de mauvaises choses, ce que tout autre environnement d'exécution asynchrone ne vous pardonnera pas, alors tout ira bien.
DVRIP
Nous considérerons le protocole, prétendument inventé par les Chinois, qui est utilisé pour communiquer les logiciels avec les caméras de surveillance.
Tout fonctionne au-dessus de TCP, appelé DVRIP (parfois Sofia), et au moment où j'en avais besoin, je n'ai trouvé que deux bibliothèques plus ou moins saines, dont l'une décrivait généralement uniquement la connexion à la caméra et l'échange de quelques messages, mais avait description de l'en-tête du package qui m'a beaucoup aidé.
Le second a été retrouvé un peu plus tard, au moment de l'exacerbation de mon syndrome non inventé ici, et il fonctionnait un peu différemment de ce que j'aimerais.
En général, armé d'un renifleur et d'une description de l'en-tête du package, j'ai brièvement esquissé un certain script qui tourne quelque part sur mon serveur quelque part, et sera présenté dans une certaine preuve de concept dans cet article.
Le protocole lui-même n'est pas complexe et le package ressemble à ceci:
Nous ne transmettrons que json, mais nous recevrons les deux.
La réponse (s'il s'agit d'un package avec json) contient généralement un champ
Ret indiquant le succès de notre demande. Les réponses contenant un code
Ret différent de
100 ou
515 seront considérées comme délibérément erronées.
Armés de ces connaissances de base, connectons-nous à la caméra et comprenons.
Nous avons besoin de deux bibliothèques -
amphp / socket , qui affichera presque tout ce dont nous avons besoin, et
evenement / evenement , qui sont disponibles dans packagist et ne nécessitent aucune extension.
Travailler avec TCP dans Amp ressemble à ceci:
Loop::run(function () { $socket = yield connect("tcp://{$addr}:{$port}"); yield $socket->write('Hello'); while ($chunk = yield $socket->read()) { echo $chunk; } $socket->close(); });
DVRIP fonctionne généralement sur le port 34567, donc dans notre cas, ce sera quelque chose comme ceci:
$socket = yield connect('tcp://192.168.0.200:34567');
Ensuite, nous nous connectons en envoyant notre hachage de nom d'utilisateur et de mot de passe à la caméra, qui est calculé comme suit:
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 cas d'autorisation réussie, nous demandons une vidéo et, si la caméra peut nous la donner, nous commençons à la transmettre.
Chaque type de package a son propre msgId, afin que nous puissions comprendre quelle demande nous avons reçu la réponse. En général, ce n'est pas un travail poussiéreux.
Pour plus de commodité, décrivons d'abord la classe de notre package:
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]; } }
et essayez d'envoyer une demande d'autorisation
yield $socket->write((string) new Packet(Packet::MESSAGE_CODES['packet.request.login'], [ 'EncryptType' => 'MD5', 'LoginType' => 'DVRIP-Web', 'PassWord' => sofiaHash('admin'), 'UserName' => 'admin', ]));
Ouais, super. Nous nous sommes connectés et avons envoyé un message, mais nous ne savons même pas si la caméra était correcte et y avons réagi au moins d'une manière ou d'une autre. Nous devons en quelque sorte lire la réponse (faisons-le de manière asynchrone), sélectionner des paquets spécifiques et les décoder.
Afin de séparer en quelque sorte la logique de traitement des paquets venant en réponse, j'ai décidé d'utiliser la bibliothèque
evenement et esquissé une classe basée sur celle-ci qui jettera les paquets déjà décodés
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; } }
et également ajouté quelques méthodes statiques à notre classe Packet, qui, en fait, extraira et décodera les paquets du flux de données
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; }
Maintenant, grâce à l'utilisation d'
Evenement , nous pouvons décrire notre logique sous la forme
$reader->on($packetType, $handler);
Et donc j'ai eu cette chaîne, où j'ai ajouté un autre traitement Keep Alive et 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());
Il va sans dire qu'au lieu de stdout, nous pouvons rediriger le flux vidéo où nous voulons.
Script 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()); });
Maintenant, nous pouvons essayer d'enregistrer un morceau de vidéo de cette façon
$ php recorder.php > output.h264
Nous avons ajouté un gestionnaire SIGINT, donc lorsque vous êtes fatigué, appuyez simplement sur Ctrl + C et le script arrêtera le transfert vidéo et fermera la connexion normalement.
Et nous pouvons, par exemple, rediriger le flux vers ffmpeg et transcoder à la volée.
$ 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
Et vous pouvez faire des chaînes frénétiques de la forme
$ php recorder.php | ffmpeg ... | curl -d @- ...
en lançant tout cela avec une sorte de script qui redémarrera toutes les heures, nous obtenons donc des morceaux fluides d'une heure, à la volée, partout où nous en avons besoin.
Pourquoi?
Je suis porté à croire que cet article soulèvera la question "Pourquoi?" Il existe des solutions toutes faites pour l'enregistrement à partir de caméras, les caméras en général peuvent faire beaucoup de choses elles-mêmes, comme elles l'ont objecté à juste titre dans les commentaires de l'article précédent, et en général, cela pourrait être beaucoup plus facile.
C'est juste que j'avais du temps libre que je devais faire quelque part, un désir de faire mon propre chemin et un infatigable désir d'expérimenter.
Et ça marche.
Bien sûr, le script présenté dans l'article nécessite quelques modifications, et vous ne devez pas l'utiliser sous cette forme - je le répète, c'est une preuve de concept. Et en général, je vous conseille fortement d'utiliser des produits faits maison à hauteur du genou pour assurer la sécurité de votre maison.
Dans mon cas, la fiabilité n'est pas si critique, donc cette option fonctionnera. Un observateur de script tourne sur ce script, qui transfère le flux vers ffmpeg, enregistre les morceaux à l'heure et les télécharge sur le VK, redémarre le script en cas de déconnexion ou de toute circonstance imprévue.
En général, tout cela aurait pu être évité en retirant simplement la vidéo sur RTSP sur TCP, comme je l'ai suggéré à la fin de l'article précédent, mais ffmpeg aimait stupidement caler à un moment aléatoire et commencer à répondre uniquement à SIGKILL.
J'ai commencé à soupçonner que tout cela était dû au fait que le script-watch parent était géré par la couronne et recevait une priorité inférieure, mais l'augmentation de la priorité ne faisait que réduire la fréquence du problème.
Dans le cas du script ci-dessus, tout fonctionne beaucoup plus stable et n'a pas encore soulevé de plaintes.