blakeblackshear.frigate/frigate/test/test_config.py

import json
from unittest import TestCase, main
import voluptuous as vol
from frigate.config import FRIGATE_CONFIG_SCHEMA, FrigateConfig

class TestConfig(TestCase):
    def setUp(self):
        self.minimal = {
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
    def test_empty(self):
        FRIGATE_CONFIG_SCHEMA({})

    def test_minimal(self):
        FRIGATE_CONFIG_SCHEMA(self.minimal)
    
    def test_config_class(self):
        FrigateConfig(config=self.minimal)
    
    def test_inherit_tracked_objects(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'objects': {
                'track': ['person', 'dog']
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('dog' in frigate_config.cameras['back'].objects.track)
    
    def test_override_tracked_objects(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'objects': {
                'track': ['person', 'dog']
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'track': ['cat']
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('cat' in frigate_config.cameras['back'].objects.track)
    
    def test_default_object_filters(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'objects': {
                'track': ['person', 'dog']
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('dog' in frigate_config.cameras['back'].objects.filters)
    
    def test_inherit_object_filters(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'objects': {
                'track': ['person', 'dog'],
                'filters': {
                    'dog': {
                        'threshold': 0.7
                    }
                }
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('dog' in frigate_config.cameras['back'].objects.filters)
        assert(frigate_config.cameras['back'].objects.filters['dog'].threshold == 0.7)
    
    def test_override_object_filters(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'track': ['person', 'dog'],
                        'filters': {
                            'dog': {
                                'threshold': 0.7
                            }
                        }
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('dog' in frigate_config.cameras['back'].objects.filters)
        assert(frigate_config.cameras['back'].objects.filters['dog'].threshold == 0.7)
    
    def test_global_object_mask(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'objects': {
                'track': ['person', 'dog']
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'mask': '0,0,1,1,0,1',
                        'filters': {
                            'dog': {
                                'mask': '1,1,1,1,1,1'
                            }
                        }
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('dog' in frigate_config.cameras['back'].objects.filters)
        assert(len(frigate_config.cameras['back'].objects.filters['dog']._raw_mask) == 2)
        assert(len(frigate_config.cameras['back'].objects.filters['person']._raw_mask) == 1)
    
    def test_ffmpeg_params_global(self):
        config = {
            'ffmpeg': {
                'input_args': ['-re']
            },
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'track': ['person', 'dog'],
                        'filters': {
                            'dog': {
                                'threshold': 0.7
                            }
                        }
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('-re' in frigate_config.cameras['back'].ffmpeg_cmds[0]['cmd'])

    def test_ffmpeg_params_camera(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ],
                        'input_args': ['-re']
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'track': ['person', 'dog'],
                        'filters': {
                            'dog': {
                                'threshold': 0.7
                            }
                        }
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('-re' in frigate_config.cameras['back'].ffmpeg_cmds[0]['cmd'])

    def test_ffmpeg_params_input(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'], 'input_args': ['-re'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'objects': {
                        'track': ['person', 'dog'],
                        'filters': {
                            'dog': {
                                'threshold': 0.7
                            }
                        }
                    }
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert('-re' in frigate_config.cameras['back'].ffmpeg_cmds[0]['cmd'])
    
    
    def test_inherit_clips_retention(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'clips': {
                'retain': {
                    'default': 20,
                    'objects': {
                        'person': 30
                    }
                }
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
        frigate_config = FrigateConfig(config=config)
        assert(frigate_config.cameras['back'].clips.retain.objects['person'] == 30)
    
    def test_roles_listed_twice_throws_error(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'clips': {
                'retain': {
                    'default': 20,
                    'objects': {
                        'person': 30
                    }
                }
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] },
                            { 'path': 'rtsp://10.0.0.1:554/video2', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }
        self.assertRaises(vol.MultipleInvalid, lambda: FrigateConfig(config=config))
    
    def test_zone_matching_camera_name_throws_error(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'clips': {
                'retain': {
                    'default': 20,
                    'objects': {
                        'person': 30
                    }
                }
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'zones': {
                        'back': {
                            'coordinates': '1,1,1,1,1,1'
                        }
                    }
                }
            }
        }
        self.assertRaises(vol.MultipleInvalid, lambda: FrigateConfig(config=config))
    
    def test_clips_should_default_to_global_objects(self):
        config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'clips': {
                'retain': {
                    'default': 20,
                    'objects': {
                        'person': 30
                    }
                }
            },
            'objects': {
                'track': ['person', 'dog']
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920,
                    'clips': {
                        'enabled': True
                    }
                }
            }
        }
        config = FrigateConfig(config=config)
        assert(config.cameras['back'].clips.objects is None)
    
    def test_role_assigned_but_not_enabled(self):
        json_config = {
            'mqtt': {
                'host': 'mqtt'
            },
            'cameras': {
                'back': {
                    'ffmpeg': {
                        'inputs': [
                            { 'path': 'rtsp://10.0.0.1:554/video', 'roles': ['detect', 'rtmp'] },
                            { 'path': 'rtsp://10.0.0.1:554/record', 'roles': ['record'] }
                        ]
                    },
                    'height': 1080,
                    'width': 1920
                }
            }
        }

        config = FrigateConfig(config=json_config)
        ffmpeg_cmds = config.cameras['back'].ffmpeg_cmds
        assert(len(ffmpeg_cmds) == 1)
        assert(not 'clips' in ffmpeg_cmds[0]['roles'])


if __name__ == '__main__':
    main(verbosity=2)