reformat code with black

Author: Lionel Untereiner

etpclient/__init__.py

@@ -1 +1 @@
-#https://florimond.dev/blog/articles/2020/05/structuring-starlette-projects/
+# https://florimond.dev/blog/articles/2020/05/structuring-starlette-projects/

etpclient/etp/connection_register.py

@@ -28,4 +28,4 @@ def get_ws_n_con(cls, client_info: ClientInfo) -> WebSocket:
         print(client_info.ip, ": #ConnectionRegister.get_ws_n_con")
         if client_info in cls.map_connection:
             return cls.map_connection[client_info]
-        return None
+        return None

etpclient/etp/etp_data_bridge.py

@@ -7,11 +7,21 @@
 from etpproto.messages import Message
 
 from etptypes.helpers import AvroModel
-from etptypes.energistics.etp.v12.protocol.store_notification import UnsolicitedStoreNotifications
-from etptypes.energistics.etp.v12.datatypes.object.subscription_info import SubscriptionInfo
-from etptypes.energistics.etp.v12.datatypes.object.context_info import ContextInfo
-from etptypes.energistics.etp.v12.datatypes.object.relationship_kind import RelationshipKind
-from etptypes.energistics.etp.v12.datatypes.object.context_scope_kind import ContextScopeKind
+from etptypes.energistics.etp.v12.protocol.store_notification import (
+    UnsolicitedStoreNotifications,
+)
+from etptypes.energistics.etp.v12.datatypes.object.subscription_info import (
+    SubscriptionInfo,
+)
+from etptypes.energistics.etp.v12.datatypes.object.context_info import (
+    ContextInfo,
+)
+from etptypes.energistics.etp.v12.datatypes.object.relationship_kind import (
+    RelationshipKind,
+)
+from etptypes.energistics.etp.v12.datatypes.object.context_scope_kind import (
+    ContextScopeKind,
+)
 
 from gabbro.etp.connection_register import ConnectionRegister
 
@@ -20,12 +30,16 @@
 class ETPNotifyer:
 
     # For each ETPConnection, store a list of notifications subscription
-    map_client_to_subscribtion: ClassVar[Mapping[ClientInfo, list[AvroModel]]] = {}
+    map_client_to_subscribtion: ClassVar[
+        Mapping[ClientInfo, list[AvroModel]]
+    ] = {}
 
     unsolicited_notification: ClassVar[list[AvroModel]] = []
 
     @classmethod
-    async def subscribe(cls, client_info: ClientInfo, subscription: Any) -> None:
+    async def subscribe(
+        cls, client_info: ClientInfo, subscription: Any
+    ) -> None:
         print(client_info.ip, ": #ETPNotiyer.subscribe")
 
         if client_info not in cls.map_client_to_subscribtion:
@@ -34,7 +48,9 @@ async def subscribe(cls, client_info: ClientInfo, subscription: Any) -> None:
         cls.map_client_to_subscribtion[client_info].append(subscription)
 
     @classmethod
-    async def unsubscribe(cls, client_info: ClientInfo, subscription: Any) -> None:
+    async def unsubscribe(
+        cls, client_info: ClientInfo, subscription: Any
+    ) -> None:
         print(client_info.ip, ": #ETPNotiyer.unsubscribe")
         cls.map_client_to_subscribtion.pop(client_info, None)
 
@@ -48,22 +64,36 @@ async def trigger(cls, notif_content: Any):
     async def _notify(cls, client_info: ClientInfo, msg: bytes):
         print(client_info.ip, ": #ETPNotiyer._notify")
         websocket, connection = ConnectionRegister.get_ws_n_con(client_info)
-        
+
         if websocket:
             await websocket.send_bytes(msg)
         else:
-            print(client_info.ip, ": #ETPNotiyer._notify => websocket not found")
+            print(
+                client_info.ip, ": #ETPNotiyer._notify => websocket not found"
+            )
 
     @classmethod
     async def send_unsolicited_notification(cls, client_info: ClientInfo):
-        print(client_info.ip, ": #ETPNotiyer.send_unsolicited_notification : <", len(cls.unsolicited_notification), ">")
+        print(
+            client_info.ip,
+            ": #ETPNotiyer.send_unsolicited_notification : <",
+            len(cls.unsolicited_notification),
+            ">",
+        )
         websocket, connection = ConnectionRegister.get_ws_n_con(client_info)
 
         for msg in cls.unsolicited_notification:
             m = Message.get_object_message(etp_object=msg)
-            print(client_info.ip, ": sending unsolicited_notification  ", m, "\n\n")
+            print(
+                client_info.ip,
+                ": sending unsolicited_notification  ",
+                m,
+                "\n\n",
+            )
 
-            async for msg_part in connection.send_msg_and_error_generator(m, None):
+            async for msg_part in connection.send_msg_and_error_generator(
+                m, None
+            ):
                 await websocket.send_bytes(msg_part)
 
     @classmethod
@@ -72,13 +102,16 @@ def add_unsolicited_notification(cls, msg: AvroModel):
 
 
 u_store_notif = UnsolicitedStoreNotifications(
-        subscriptions=[
-            SubscriptionInfo( context=ContextInfo(uri="eml:///", depth=3, navigable_edges=RelationshipKind.BOTH),
-                            scope=ContextScopeKind.SOURCES,
-                            request_uuid=pyUUID.uuid4().bytes,
-                            start_time=0,
-                            include_object_data=False,
-            )
-        ]
-    )
+    subscriptions=[
+        SubscriptionInfo(
+            context=ContextInfo(
+                uri="eml:///", depth=3, navigable_edges=RelationshipKind.BOTH
+            ),
+            scope=ContextScopeKind.SOURCES,
+            request_uuid=pyUUID.uuid4().bytes,
+            start_time=0,
+            include_object_data=False,
+        )
+    ]
+)
 ETPNotifyer.add_unsolicited_notification(msg=u_store_notif)

etpclient/etp/etp_notifyer.py

@@ -1,54 +1,61 @@
 import h5py
 
 from etpclient.utils import (
-    search_all_element_value, 
-    get_xml_dict_from_path, 
-    get_xml_dict_from_string, 
-    xml_dict_get_uri
+    search_all_element_value,
+    get_xml_dict_from_path,
+    get_xml_dict_from_string,
+    xml_dict_get_uri,
 )
 
 from etptypes.energistics.etp.v12.protocol.data_array.put_data_arrays import (
     PutDataArrays,
 )
 
-def descend_obj(obj,sep='\t'):
+
+def descend_obj(obj, sep="\t"):
     """
     Iterate through groups in a HDF5 file and prints the groups and datasets names and datasets attributes
     """
-    if type(obj) in [h5py._hl.group.Group,h5py._hl.files.File]:
+    if type(obj) in [h5py._hl.group.Group, h5py._hl.files.File]:
         for key in obj.keys():
-            print(sep,'-',key,':',obj[key])
-            descend_obj(obj[key],sep=sep+'\t')
+            print(sep, "-", key, ":", obj[key])
+            descend_obj(obj[key], sep=sep + "\t")
     elif type(obj) == h5py._hl.dataset.Dataset:
         print("-------------------")
         for key in obj.attrs.keys():
-            print(sep+'\t','-',key,':',obj.attrs[key])
+            print(sep + "\t", "-", key, ":", obj.attrs[key])
         print(obj[...])
 
 
-def h5dump(path,group='/'):
+def h5dump(path, group="/"):
     """
     print HDF5 file metadata
 
     group: you can give a specific group, defaults to the root group
     """
-    with h5py.File(path,'r') as f:
-         descend_obj(f[group])
+    with h5py.File(path, "r") as f:
+        descend_obj(f[group])
+
 
 def h5_search_dataset(h5_file_path, path_in_hdf):
-    with h5py.File(h5_file_path,'r') as f:
+    with h5py.File(h5_file_path, "r") as f:
         # _h5_search_dataset_in_obj(f[group])
         dataset = f[path_in_hdf]
         if dataset is not None:
             return dataset[...], dataset.shape, dataset.dtype
     print("Dataset", path_in_hdf, "not found in", h5_file_path)
     return None
 
-def generate_put_data_arrays(xml_obj: str, h5_file_path: str, dataspace: str=None):
+
+def generate_put_data_arrays(
+    xml_obj: str, h5_file_path: str, dataspace: str = None
+):
     res = []
-    
+
     obj_dict = get_xml_dict_from_string(xml_obj)
-    for path_in_hdf in search_all_element_value(obj_dict, "PathInExternalFile") + search_all_element_value(obj_dict, "PathInHdfFile"):
+    for path_in_hdf in search_all_element_value(
+        obj_dict, "PathInExternalFile"
+    ) + search_all_element_value(obj_dict, "PathInHdfFile"):
         data, shape, dtype = h5_search_dataset(h5_file_path, path_in_hdf)
         pda_dict = {
             "dataArrays": {
@@ -59,11 +66,7 @@ def generate_put_data_arrays(xml_obj: str, h5_file_path: str, dataspace: str=Non
                     },
                     "array": {
                         "dimensions": list(shape),
-                        "data": {
-                            "item": {
-                                "values": data.flatten().tolist()
-                            }
-                        },
+                        "data": {"item": {"values": data.flatten().tolist()}},
                     },
                     "customData": {},
                 }
@@ -72,4 +75,4 @@ def generate_put_data_arrays(xml_obj: str, h5_file_path: str, dataspace: str=Non
         res.append(PutDataArrays.parse_obj(pda_dict))
         # res.append(h5_search_dataset(h5_file_path, path_in_hdf))
 
-    return res
+    return res