Skip to content

Commit

Permalink
Merge pull request #162 from dvm-shlee/main
Browse files Browse the repository at this point in the history
[patch] update type annotations for python <3.9 compatibility
  • Loading branch information
dvm-shlee authored Apr 15, 2024
2 parents a55eece + e5417c0 commit ee23d76
Show file tree
Hide file tree
Showing 9 changed files with 60 additions and 59 deletions.
6 changes: 3 additions & 3 deletions brkraw/api/analyzer/affine.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from .base import BaseAnalyzer
import numpy as np
from copy import copy
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from ..brkobj.scan import ScanInfo

Expand Down Expand Up @@ -42,7 +42,7 @@ def __init__(self, infoobj: 'ScanInfo'):
self.subj_type = infoobj.orientation['subject_type'] if hasattr(infoobj, 'orientation') else None
self.subj_position = infoobj.orientation['subject_position'] if hasattr(infoobj, 'orientation') else None

def get_affine(self, subj_type:str|None=None, subj_position:str|None=None):
def get_affine(self, subj_type: Optional[str] = None, subj_position: Optional[str] = None):
subj_type = subj_type or self.subj_type
subj_position = subj_position or self.subj_position
if isinstance(self.affine, list):
Expand All @@ -51,7 +51,7 @@ def get_affine(self, subj_type:str|None=None, subj_position:str|None=None):
affine = self._correct_orientation(self.affine, subj_position, subj_type)
return affine

def _calculate_affine(self, infoobj: 'ScanInfo', slicepack_id:int|None = None):
def _calculate_affine(self, infoobj: 'ScanInfo', slicepack_id: Optional[int] = None):
sidx = infoobj.orientation['orientation_desc'][slicepack_id].index(2) \
if slicepack_id else infoobj.orientation['orientation_desc'].index(2)
slice_orient = SLICEORIENT[sidx]
Expand Down
4 changes: 2 additions & 2 deletions brkraw/api/analyzer/dataarray.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,15 @@
from .base import BaseAnalyzer
import numpy as np
from copy import copy
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Union
if TYPE_CHECKING:
from ..brkobj import ScanInfo
from io import BufferedReader
from zipfile import ZipExtFile


class DataArrayAnalyzer(BaseAnalyzer):
def __init__(self, infoobj: 'ScanInfo', fileobj: BufferedReader|ZipExtFile):
def __init__(self, infoobj: 'ScanInfo', fileobj: Union[BufferedReader, ZipExtFile]):
infoobj = copy(infoobj)
self._parse_info(infoobj)
self.buffer = fileobj
Expand Down
8 changes: 4 additions & 4 deletions brkraw/api/analyzer/scaninfo.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
from collections import OrderedDict
from brkraw.api import helper
from .base import BaseAnalyzer
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional, Union
if TYPE_CHECKING:
from ..pvobj import PvScan, PvReco, PvFiles

Expand All @@ -18,8 +18,8 @@ class ScanInfoAnalyzer(BaseAnalyzer):
NotImplementedError: If an operation is not implemented.
"""
def __init__(self,
pvobj: 'PvScan'|'PvReco'|'PvFiles',
reco_id:int|None = None,
pvobj: Union['PvScan', 'PvReco', 'PvFiles'],
reco_id:Optional[int] = None,
debug:bool = False):

self._set_pars(pvobj, reco_id)
Expand All @@ -28,7 +28,7 @@ def __init__(self,
if self.visu_pars:
self._parse_info()

def _set_pars(self, pvobj: 'PvScan'|'PvReco'|'PvFiles', reco_id: int|None):
def _set_pars(self, pvobj: Union['PvScan', 'PvReco', 'PvFiles'], reco_id: Optional[int]):
for p in ['acqp', 'method']:
try:
vals = getattr(pvobj, p)
Expand Down
15 changes: 8 additions & 7 deletions brkraw/api/brkobj/scan.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from __future__ import annotations
from typing import Optional
import ctypes
from ..pvobj import PvScan
from ..analyzer import ScanInfoAnalyzer, AffineAnalyzer, DataArrayAnalyzer, BaseAnalyzer
Expand All @@ -14,8 +15,8 @@ def num_warns(self):


class ScanObj(PvScan):
def __init__(self, pvscan: 'PvScan', reco_id: int|None = None,
loader_address: int|None = None, debug: bool=False):
def __init__(self, pvscan: 'PvScan', reco_id: Optional[int] = None,
loader_address: Optional[int] = None, debug: bool=False):
super().__init__(pvscan._scan_id,
(pvscan._rootpath, pvscan._path),
pvscan._contents,
Expand Down Expand Up @@ -45,25 +46,25 @@ def get_info(self, reco_id:int, get_analyzer:bool = False):
infoobj.warns.extend(attr_vals['warns'])
return infoobj

def get_affine_info(self, reco_id:int|None = None):
def get_affine_info(self, reco_id:Optional[int] = None):
if reco_id:
info = self.get_info(reco_id)
else:
info = self.info if hasattr(self, 'info') else self.get_info(self.reco_id)
return AffineAnalyzer(info)

def get_data_info(self, reco_id: int|None = None):
def get_data_info(self, reco_id: Optional[int] = None):
reco_id = reco_id or self.avail[0]
recoobj = self.get_reco(reco_id)
fileobj = recoobj.get_2dseq()
info = self.info if hasattr(self, 'info') else self.get_info(self.reco_id)
return DataArrayAnalyzer(info, fileobj)

def get_affine(self, reco_id:int|None = None,
subj_type:str|None = None, subj_position:str|None = None):
def get_affine(self, reco_id:Optional[int] = None,
subj_type:Optional[str] = None, subj_position:Optional[str] = None):
return self.get_affine_info(reco_id).get_affine(subj_type, subj_position)

def get_dataarray(self, reco_id: int|None = None):
def get_dataarray(self, reco_id: Optional[int] = None):
return self.get_data_info(reco_id).get_dataarray()

def retrieve_pvscan(self):
Expand Down
2 changes: 1 addition & 1 deletion brkraw/api/brkobj/study.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ def get_scan(self, scan_id, reco_id=None, debug=False):
return ScanObj(pvscan=pvscan, reco_id=reco_id,
loader_address=id(self), debug=debug)

def _parse_header(self) -> (Dict | None):
def _parse_header(self):
if not self.contents or 'subject' not in self.contents['files']:
self.header = None
return
Expand Down
4 changes: 2 additions & 2 deletions brkraw/api/helper/orientation.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations
import math
import numpy as np
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
from .base import BaseHelper, is_all_element_same
from .slicepack import SlicePack
if TYPE_CHECKING:
Expand Down Expand Up @@ -137,7 +137,7 @@ def _case_multi_slicepacks_multi_slices(self):
self._orient = orientation
self._position = positions

def _est_volume_origin(self, id: int|None =None):
def _est_volume_origin(self, id: Optional[int] =None):
"""Estimate the origin coordinates of the Volume matrix.
Notes:
Expand Down
30 changes: 15 additions & 15 deletions brkraw/app/tonifti/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from brkraw.api.brkobj import ScanObj, ScanInfo
from brkraw.api.analyzer import ScanInfoAnalyzer, DataArrayAnalyzer, AffineAnalyzer
from .header import Header
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional, Union
if TYPE_CHECKING:
from pathlib import Path

Expand All @@ -25,7 +25,7 @@ class ScaleMode(Enum):
class BaseMethods:
info, fileobj = (None, None)

def set_scale_mode(self, scale_mode:ScaleMode|None):
def set_scale_mode(self, scale_mode:Optional[ScaleMode]=None):
if scale_mode:
self.scale_mode = scale_mode
else:
Expand All @@ -45,9 +45,9 @@ def _set_info(self):
self.analysed = analysed

@staticmethod
def get_dataobj(scanobj:'ScanInfo'|'ScanObj',
fileobj:'BufferedReader'|'ZipExtFile'|None = None,
reco_id:int|None = None,
def get_dataobj(scanobj:Union['ScanInfo','ScanObj'],
fileobj:Union['BufferedReader', 'ZipExtFile', None] = None,
reco_id:Optional[int] = None,
scale_correction:bool = False):
data_dict = BaseMethods.get_data_dict(scanobj, fileobj, reco_id)
dataobj = data_dict['data_array']
Expand All @@ -62,17 +62,17 @@ def get_dataobj(scanobj:'ScanInfo'|'ScanObj',
return dataobj

@staticmethod
def get_affine(scanobj:'ScanInfo'|'ScanObj', reco_id:int|None = None,
subj_type:str|None=None, subj_position:str|None=None):
def get_affine(scanobj:Union['ScanInfo', 'ScanObj'], reco_id:Optional[int] = None,
subj_type:Optional[str]=None, subj_position:Optional[str]=None):
return BaseMethods.get_affine_dict(scanobj, reco_id, subj_type, subj_position)['affine']

@staticmethod
def get_data_dict(scanobj:'ScanInfo'|'ScanObj',
fileobj:'BufferedReader'|'ZipExtFile'|None = None,
reco_id:int|None = None):
def get_data_dict(scanobj:Union['ScanInfo', 'ScanObj'],
fileobj:Union['BufferedReader', 'ZipExtFile'] = None,
reco_id:Optional[int] = None):
if isinstance(scanobj, ScanObj):
data_info = scanobj.get_data_info(reco_id)
elif isinstance(scanobj, ScanInfo) and isinstance(scanobj, BufferedReader|ZipExtFile):
elif isinstance(scanobj, ScanInfo) and isinstance(scanobj, Union[BufferedReader, ZipExtFile]):
data_info = DataArrayAnalyzer(scanobj, fileobj)
else:
raise TypeError(
Expand All @@ -94,8 +94,8 @@ def get_data_dict(scanobj:'ScanInfo'|'ScanObj',
}

@staticmethod
def get_affine_dict(scanobj:'ScanInfo'|'ScanObj', reco_id:int|None = None,
subj_type:str|None = None, subj_position:str|None = None):
def get_affine_dict(scanobj:Union['ScanInfo','ScanObj'], reco_id:Optional[int] = None,
subj_type:Optional[str] = None, subj_position:Optional[str] = None):
if isinstance(scanobj, ScanObj):
affine_info = scanobj.get_affine_info(reco_id)
elif isinstance(scanobj, ScanInfo):
Expand Down Expand Up @@ -131,11 +131,11 @@ def get_bdata(analobj:'ScanInfoAnalyzer'):
return bvals, bvecs

@staticmethod
def get_bids_metadata(scaninfo:'ScanInfo', bids_recipe:'Path'|None=None):
def get_bids_metadata(scaninfo:'ScanInfo', bids_recipe:Optional['Path']=None):
print(isinstance(scaninfo, ScanInfo), bids_recipe)

@staticmethod
def get_nifti1header(scaninfo:'ScanInfo', scale_mode:'ScaleMode'|None):
def get_nifti1header(scaninfo:'ScanInfo', scale_mode:Optional['ScaleMode']=None):
scale_mode = scale_mode or ScaleMode.HEADER
return Header(scaninfo, scale_mode).get()

Expand Down
46 changes: 23 additions & 23 deletions brkraw/app/tonifti/brkraw.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,18 @@
from __future__ import annotations
from brkraw.api.brkobj import StudyObj
from .base import BaseMethods, ScaleMode
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from pathlib import Path


class BrkrawToNifti(StudyObj, BaseMethods):
def __init__(self, path:'Path', scale_mode: 'ScaleMode'|None = None):
def __init__(self, path:'Path', scale_mode: Optional['ScaleMode'] = None):
"""_summary_
Args:
path (Path): _description_
scale_mode (ScaleMode | None, optional): _description_. Defaults to None.
scale_mode (ScaleMode , None, optional): _description_. Defaults to None.
"""

super().__init__(path)
Expand All @@ -32,40 +32,40 @@ def get_scan(self, scan_id:int):
self._cache[scan_id] = super().get_scan(scan_id)
return self._cache[scan_id]

def get_scan_analyzer(self, scan_id:int, reco_id:int|None=None):
def get_scan_analyzer(self, scan_id:int, reco_id:Optional[int]=None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
"""
return self.get_scan(scan_id).get_info(reco_id, get_analyzer=True)

def get_affine(self, scan_id:int, reco_id:int|None=None, subj_type:str|None=None, subj_position:str|None=None):
def get_affine(self, scan_id:int, reco_id:Optional[int]=None, subj_type:Optional[str]=None, subj_position:Optional[str]=None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
subj_type (str | None, optional): _description_. Defaults to None.
subj_position (str | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
subj_type (str , None, optional): _description_. Defaults to None.
subj_position (str , None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
"""
scanobj = self.get_scan(scan_id)
return super().get_affine(scanobj=scanobj, reco_id=reco_id, subj_type=subj_type, subj_position=subj_position)

def get_dataobj(self, scan_id:int, reco_id:int|None=None, scale_mode:'ScaleMode'|None = None):
def get_dataobj(self, scan_id:int, reco_id:Optional[int]=None, scale_mode:Optional['ScaleMode'] = None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
scale_mode (ScaleMode; | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
scale_mode (ScaleMode; , None, optional): _description_. Defaults to None.
Raises:
ValueError: _description_
Expand All @@ -78,27 +78,27 @@ def get_dataobj(self, scan_id:int, reco_id:int|None=None, scale_mode:'ScaleMode'
scanobj = self.get_scan(scan_id)
return super().get_dataobj(scanobj=scanobj, fileobj=None, reco_id=reco_id, scale_correction=scale_correction)

def get_data_dict(self, scan_id:int, reco_id:int|None=None):
def get_data_dict(self, scan_id:int, reco_id:Optional[int]=None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
"""
scanobj = self.get_scan(scan_id)
return super().get_data_dict(scanobj=scanobj, reco_id=reco_id)

def get_affine_dict(self, scan_id:int, reco_id:int|None=None, subj_type:str|None=None, subj_position:str|None=None):
def get_affine_dict(self, scan_id:int, reco_id:Optional[int]=None, subj_type:Optional[str]=None, subj_position:Optional[str]=None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
subj_type (str | None, optional): _description_. Defaults to None.
subj_position (str | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
subj_type (str , None, optional): _description_. Defaults to None.
subj_position (str , None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
Expand All @@ -107,13 +107,13 @@ def get_affine_dict(self, scan_id:int, reco_id:int|None=None, subj_type:str|None
return super().get_affine_dict(scanobj=scanobj, reco_id=reco_id,
subj_type=subj_type, subj_position=subj_position)

def get_nifti1header(self, scan_id:int, reco_id:int|None=None, scale_mode:'ScaleMode'|None = None):
def get_nifti1header(self, scan_id:int, reco_id:Optional[int]=None, scale_mode:Optional['ScaleMode'] = None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
scale_mode (ScaleMode | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
scale_mode (ScaleMode , None, optional): _description_. Defaults to None.
Returns:
_type_: _description_
Expand All @@ -134,12 +134,12 @@ def get_bdata(self, scan_id:int):
analobj = self.get_scan_analyzer(scan_id)
return super().get_bdata(analobj)

def get_bids_metadata(self, scan_id:int, reco_id:int|None=None, bids_recipe=None):
def get_bids_metadata(self, scan_id:int, reco_id:Optional[int]=None, bids_recipe=None):
"""_summary_
Args:
scan_id (int): _description_
reco_id (int | None, optional): _description_. Defaults to None.
reco_id (int , None, optional): _description_. Defaults to None.
bids_recipe (_type_, optional): _description_. Defaults to None.
Returns:
Expand Down
4 changes: 2 additions & 2 deletions brkraw/app/tonifti/header.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
from __future__ import annotations
import warnings
from nibabel.nifti1 import Nifti1Header
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Union
if TYPE_CHECKING:
from brkraw.api.brkobj import ScanInfo
from .base import ScaleMode


class Header:
def __init__(self, scaninfo:'ScanInfo', scale_mode:'ScaleMode'|int):
def __init__(self, scaninfo:'ScanInfo', scale_mode:Union['ScaleMode', int]):
self.info = scaninfo
self.scale_mode = int(scale_mode)
self.nifti1header = Nifti1Header()
Expand Down

0 comments on commit ee23d76

Please sign in to comment.