Add HTTP Client

Author: maxnz

animatedledstrip/als_http_client.py

@@ -0,0 +1,115 @@
+#  Copyright (c) 2018-2021 AnimatedLEDStrip
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in
+#  all copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+#  THE SOFTWARE.
+import json
+from urllib.request import Request, urlopen
+from typing import TYPE_CHECKING, List, Dict, Any
+
+from animatedledstrip import RunningAnimationParams, Section, StripInfo
+from animatedledstrip.animation_to_run_params import AnimationToRunParams
+from animatedledstrip.json_decoder import ALSJsonDecoder
+from animatedledstrip.json_encoder import ALSJsonEncoder
+
+if TYPE_CHECKING:
+    from animatedledstrip.animation_info import AnimationInfo
+    from animatedledstrip.current_strip_color import CurrentStripColor
+
+
+class ALSHttpClient:
+
+    def __init__(self, ip_address: str):
+        self.ip_address = ip_address
+        self.encoder = ALSJsonEncoder()
+        self.decoder = ALSJsonDecoder()
+
+    def _resolve_url(self, url: str) -> str:
+        return 'http://' + self.ip_address + ':8080' + url
+
+    def _get_data(self, url: str) -> Any:
+        return urlopen(Request(self._resolve_url(url))).read()
+
+    def _post_data(self, url: str, data: Any) -> Any:
+        return urlopen(Request(self._resolve_url(url),
+                               method='POST',
+                               data=bytes(self.encoder.encode(data), 'utf-8'),
+                               headers={'Content-Type': 'application/json'})).read()
+
+    def _delete_data(self, url: str) -> Any:
+        return urlopen(Request(self._resolve_url(url), method='DELETE')).read()
+
+    def get_animation_info(self, anim_name: str) -> 'AnimationInfo':
+        return self.decoder.decode_with_type(self._get_data('/animation/' + anim_name), 'AnimationInfo')
+
+    def get_supported_animation_names(self) -> List[str]:
+        return json.loads(self._get_data('/animations/names'))
+
+    def get_supported_animations(self) -> List['AnimationInfo']:
+        return self.decoder.decode_list_with_type(self._get_data('/animations'), 'AnimationInfo')
+
+    def get_supported_animations_map(self) -> Dict[str, 'AnimationInfo']:
+        return self.decoder.decode_map_with_type(self._get_data('/animations/map'), 'AnimationInfo')
+
+    def get_supported_animations_dict(self) -> Dict[str, 'AnimationInfo']:
+        return self.get_supported_animations_map()
+
+    def get_running_animations(self) -> Dict[str, 'RunningAnimationParams']:
+        return self.decoder.decode_map_with_type(self._get_data('/running'), 'RunningAnimationParams')
+
+    def get_running_animations_ids(self) -> List[str]:
+        return json.loads(self._get_data('/running/ids'))
+
+    def get_running_animation_params(self, anim_id: str) -> 'RunningAnimationParams':
+        return self.decoder.decode_with_type(self._get_data('/running/' + anim_id), 'RunningAnimationParams')
+
+    def end_animation(self, anim_id: str) -> 'RunningAnimationParams':
+        return self.decoder.decode_with_type(self._delete_data('/running/' + anim_id), 'RunningAnimationParams')
+
+    def end_animation_from_params(self, anim_params: 'RunningAnimationParams') -> 'RunningAnimationParams':
+        return self.end_animation(anim_params.anim_id)
+
+    def get_sections(self) -> List['Section']:
+        return self.decoder.decode_list_with_type(self._get_data('/sections'), 'Section')
+
+    def get_sections_map(self) -> Dict[str, 'Section']:
+        return self.decoder.decode_map_with_type(self._get_data('/sections/map'), 'Section')
+
+    def get_sections_dict(self) -> Dict[str, 'Section']:
+        return self.get_sections_map()
+
+    def get_section(self, section_id: str) -> 'Section':
+        return self.decoder.decode_with_type(self._get_data('/sections/' + section_id), 'Section')
+
+    def get_full_strip_section(self) -> 'Section':
+        return self.get_section('fullStrip')
+
+    def create_new_section(self, new_section: 'Section') -> 'Section':
+        return self.decoder.decode_with_type(self._post_data('/sections/newSection', new_section), 'Section')
+
+    def start_animation(self, anim_params: 'AnimationToRunParams') -> 'RunningAnimationParams':
+        return self.decoder.decode_with_type(self._post_data('/start', anim_params), 'RunningAnimationParams')
+
+    def get_strip_info(self) -> 'StripInfo':
+        return self.decoder.decode_with_type(self._get_data('/strip/info'), 'StripInfo')
+
+    def get_current_strip_color(self) -> 'CurrentStripColor':
+        return self.decoder.decode_with_type(self._get_data('/strip/color'), 'CurrentStripColor')
+
+    def clear_strip(self):
+        # TODO: Fix 404
+        self._post_data('/strip/clear', None)

animatedledstrip/animation_sender.py

@@ -67,7 +67,8 @@ def __init__(self, ip_address: str, port_num: int):
         self.on_new_section_callback: Optional[Callable[['Section'], Any]] = None
         self.on_new_strip_info_callback: Optional[Callable[['StripInfo'], Any]] = None
 
-        self._encoder: 'ALSJsonEncoder' = ALSJsonEncoder()
+        self.encoder: 'ALSJsonEncoder' = ALSJsonEncoder()
+
         self._recv_thread: Optional['Thread'] = None
         self._partial_data: bytes = b''
 
@@ -126,7 +127,7 @@ def end(self) -> 'AnimationSender':
     def send(self, data: Union['AnimationToRunParams', 'ClientParams',
                                'Command', 'EndAnimation', 'Section']) -> 'AnimationSender':
         """Send data to the server"""
-        self.send_json(self._encoder.encode(data))
+        self.send_json(self.encoder.encode(data))
 
         return self
 

animatedledstrip/json_decoder.py

@@ -17,7 +17,7 @@
 #  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 #  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 #  THE SOFTWARE.
-
+import json
 from json import JSONDecoder
 from typing import Dict, Any
 
@@ -52,11 +52,28 @@ def __init__(self, *args, **kwargs):
     def decode_dict(self, json: Dict) -> Any:
         return self.decode(self.encoder.encode(json))
 
+    def decode_with_type(self, obj, data_type: str):
+        return self.decode_dict(add_dicts(json.loads(obj), {"type": data_type}))
+
+    def decode_list_with_type(self, obj, data_type: str):
+        json_list = json.loads(obj)
+        decoded_list = []
+        for o in json_list:
+            decoded_list.append(self.decode_dict(add_dicts(o, {"type": data_type})))
+        return decoded_list
+
+    def decode_map_with_type(self, obj, data_type: str):
+        json_dict = json.loads(obj)
+        decoded_dict = {}
+        for k in json_dict:
+            decoded_dict.update({k: self.decode_dict(add_dicts(json_dict[k], {"type": data_type}))})
+        return decoded_dict
+
     def object_hook(self, obj):
         data_type = obj.get('type', '')
         if data_type == 'AbsoluteDistance':
             return AbsoluteDistance(obj['x'], obj['y'], obj['z'])
-        if data_type == 'AnimationInfo':
+        elif data_type == 'AnimationInfo':
             return AnimationInfo(
                 name=obj['name'],
                 abbr=obj['abbr'],

animatedledstrip/json_encoder.py

@@ -27,7 +27,7 @@ class ALSJsonEncoder(JSONEncoder):
 
     def default(self, o: Any) -> Any:
         # print(type(o))
-        if hasattr(o, "json_dict"):
+        if hasattr(o, 'json_dict'):
             # noinspection PyCallingNonCallable
             return o.json_dict()
         else: