import ast
from pathlib import Path
import re
import threading
from types import SimpleNamespace
from typing import Optional
import unittest


class CameraTests(unittest.TestCase):
    def setUp(self):
        self.now = 10.0
        source = ast.parse(Path(__file__).with_name('bridge.py').read_text())
        node = next(item for item in source.body if isinstance(item, ast.ClassDef) and item.name == 'SlamNode')
        names = ('_discover_cameras', '_on_camera_image', 'camera_sources', 'camera_frame_jpeg')
        methods = [item for item in node.body if isinstance(item, ast.FunctionDef) and item.name in names]
        scope = {'re': re, 'Optional': Optional, 'CompressedImage': object,
                 'time': SimpleNamespace(time=lambda: self.now), '_CAMERA_FRAME_MAX_AGE_SECONDS': 3,
                 'qos_profile_sensor_data': object()}
        exec(compile(ast.Module(body=methods, type_ignores=[]), 'camera-bridge', 'exec'), scope)
        self.node = type('CameraNode', (), {name: scope[name] for name in names})()
        self.node._lock = threading.Lock()
        self.node._camera_frames = {}
        self.node._camera_subscriptions = {}
        self.node._camera_image = None
        self.node._last_camera_image_wall_time = None
        self.node._camera_image_count = 0
        self.node.camera_image_topic = '/camera_test/image_jpeg'
        self.topics = []
        self.callbacks = {}
        self.node.get_topic_names_and_types = lambda: self.topics
        def subscribe(kind, topic, callback, qos):
            self.callbacks[topic] = callback
            return topic
        self.node.create_subscription = subscribe
        self.node.destroy_subscription = lambda sub: self.callbacks.pop(sub)

    def test_independent_frames_and_disconnect(self):
        for key in ['video0', 'video2']:
            self.topics.append((f'/camera_test/cameras/{key}/image_jpeg', ['sensor_msgs/msg/CompressedImage']))
        self.topics += [('/other/image_jpeg', ['sensor_msgs/msg/CompressedImage']),
                        ('/camera_test/cameras/video9/image_jpeg', ['sensor_msgs/msg/Image'])]
        self.node._discover_cameras()
        self.assertEqual(len(self.callbacks), 2)
        for topic, callback in self.callbacks.items():
            callback(SimpleNamespace(data=topic.encode(), format='jpeg'))
        self.assertEqual([item['id'] for item in self.node.camera_sources()], ['video0', 'video2'])
        self.assertIn(b'video2', self.node.camera_frame_jpeg('video2'))
        self.assertIsNone(self.node.camera_frame_jpeg('../video0'))
        self.now += 4
        self.assertTrue(all(not item['available'] for item in self.node.camera_sources()))
        self.assertIsNone(self.node.camera_frame_jpeg('video0'))
        self.topics = self.topics[:1]
        self.node._discover_cameras()
        self.assertNotIn('video2', self.node._camera_frames)

    def test_single_camera_compatibility_and_non_jpeg(self):
        self.node._on_camera_image(SimpleNamespace(data=b'jpeg', format='jpeg'))
        self.assertEqual(self.node.camera_frame_jpeg(), b'jpeg')
        self.assertEqual(self.node.camera_frame_jpeg('legacy'), b'jpeg')
        self.node._on_camera_image(SimpleNamespace(data=b'png', format='png'))
        self.assertIsNone(self.node.camera_frame_jpeg('legacy'))
        self.assertFalse(self.node.camera_sources()[0]['available'])


if __name__ == '__main__':
    unittest.main()