Add event_stream() which returns stream of camera events

[pnbruckner]

Description

Currently, determining if motion or other events occur requires polling via event_channels_happened() (or one of the other methods or properties that use it.) This can delay response to events, and can even miss events if the event duration is shorter than the polling period.

This adds a new method -- event_stream() -- which will return events as they happen. It is a generator and will normally not return, so it should be run in a separate thread or process.

Example use case

import re
import threading
from amcrest import AmcrestCamera, CommError

_START_STOP = re.compile(r"Code=([^;]+);action=(Start|Stop)", flags=re.S)

def process_events(camera, event_codes):
    event_codes = ",".join(event_codes)
    while True:
        try:
            for event_info in camera.event_stream(event_codes, retries=5):
                print("Event info:", repr(event_info))
                for code, action in _START_STOP.findall(event_info):
                    print("--->", code, action)
        except CommError as error:
            print("Error while processing events:", repr(error))

camera = AmcrestCamera(...).camera
thread = threading.Thread(target=process_events, args=(camera, ["VideoMotion"]))
thread.daemon = True
thread.start()

To do

• Make sure comm errors and camera power cycling are handled properly.
• Get feedback from other users with other camera models and/or firmware versions.
• Prototype usage in Home Assistant's amcrest integration.
• Determine if anything can be done about urllib3 warning:

WARNING:urllib3.connectionpool:Failed to parse headers (url=http://<host>:<port>/cgi-bin/eventManager.cgi?action=attach&codes=%5BVideoMotion%5D): [StartBoundaryNotFoundDefect(), MultipartInvariantViolationDefect()], unparsed data: ''
Traceback (most recent call last):
  File "/home/phil/test/amcrest/venv/lib/python3.7/site-packages/urllib3/connectionpool.py", line 441, in _make_request
    assert_header_parsing(httplib_response.msg)
  File "/home/phil/test/amcrest/venv/lib/python3.7/site-packages/urllib3/util/response.py", line 71, in assert_header_parsing
    raise HeaderParsingError(defects=defects, unparsed_data=unparsed_data)
urllib3.exceptions.HeaderParsingError: [StartBoundaryNotFoundDefect(), MultipartInvariantViolationDefect()], unparsed data: ''

Thanks to @NickWaterton and his groundwork in PR #140!

[coveralls]

Coverage increased (+0.03%) to 31.753% when pulling e8073af on stream-event into a6b6c41 on master.

[pnbruckner]

@tchellomello @dougsland @NickWaterton

I'd appreciate your comments/suggestions/feedback, even though I still have some work to do before I consider it finished. Thx!

[dougsland]

Looks good to me, just two questions.

[pnbruckner]

Regarding the urllib3 warning, I found this old issue that has not been resolved: urllib3/urllib3#800

I also found a suggested work around here: home-assistant/core#17042

I will add a similar Filter here so users don't have to worry about it.

[pnbruckner]

I did some experimenting by powering off the camera while in a call to event_stream() (which is sitting in _get_lines() -> ret.iter_content().) It will just sit there forever. And since there's no guarantee there will ever be a motion event, it's not really practical to set a read timeout on the connection.

(I did consider setting a read timeout anyway, e.g., 60 seconds. Then, when the timeout occurs, the generator would end, and another instance of the generator could be started. This might be one way to handle the camera going away. Unfortunately that would mean events could be missed. So, that's not really a practical solution.)

I did some research and discovered a TCP connection will (by default) stay established forever, even if there is no data sent in either direction. Therefore there's no way, in that scenario, to tell if the server (i.e., camera) is alive and healthy.

The solution that I found suggested is to set SO_KEEPALIVE (and set a few related TCP_KEEPxxx parameters) on the TCP connection. Unfortunately the requests package doesn't provide a simple way to set socket parameters like these. Apparently the only practical way to do so is to create a custom HTTPAdapter. I tested with the following:

class SOHTTPAdapter(requests.adapters.HTTPAdapter):
    def __init__(self, *args, **kwargs):
        self.socket_options = kwargs.pop("socket_options", None)
        super().__init__(*args, **kwargs)

    def init_poolmanager(self, *args, **kwargs):
        if self.socket_options is not None:
            kwargs["socket_options"] = self.socket_options
        super().init_poolmanager(*args, **kwargs)


so_adapter = SOHTTPAdapter(socket_options=(
    HTTPConnection.default_socket_options +
    [
        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
        (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10),
        (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 5),
        (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 4),
    ]
))

Then, instead of using requests.get, first a session is created, then the custom adapter mounted on it, then session.get is used. That seemed to work. While the camera is on the generator returns events normally. When the camera goes off, after about a half minute there is a timeout exception, which kicks it out of the event_stream() -> _get_lines() -> ret.iter_content() call stack. I'd catch that and turn it into a usual amcrest.CommError, which the user can catch and restart the generator if/when they want.

I'll play with this some more to make sure it handles all the scenarios I can think of, then I'll commit another change that uses this technique.

[pnbruckner]

Got some feedback in this community forum topic.

[pnbruckner]

I modified Home Assistant's amcrest integration to use this new event_stream method and it seems to work great! PR in HA coming soon. I believe this is ready to go. I'll wait a bit for more review comments, but will probably go ahead and merge it in the next couple of days if there's no objection.

[dougsland]

Patch looks good to me, the workaround from request library also looks good.

[dougsland]

Fell free to merge after testing.

[pnbruckner]

Thanks! BTW, I've been testing since before I created the PR. 🤣